From a5bb67ef8647bd8800594651ea14238acd118146 Mon Sep 17 00:00:00 2001
From: Stephen von Takach <steve@place.technology>
Date: Tue, 11 Aug 2026 01:16:06 +1000
Subject: [PATCH 1/4] wip: clear retained state when a topic moves or goes away
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Not working — the publisher's consume loop hangs. Parked on a branch so
master stays green.

The design, as agreed: on a rename, reindex or destroy, collect the keys
the Module currently publishes and queue them for removal; drain that
queue ahead of new messages, so a rename cannot have its republished value
deleted by a late arriving deletion; and make a state resync wait for the
queue to drain.

What is here and looks right:

- Publisher gains a delete_queue beside message_queue, plus an overridable
  delete. MqttPublisher implements it as a zero length retained publish,
  which is the only thing that actually removes a retained value — a JSON
  null would simply be retained in its place
- InfluxPublisher does nothing for it. A time series has no current value
  to remove, and a renamed Module's history belongs under its old tags
- Module and Driver routers collect the affected keys *before* the mapping
  is rewritten, while the old topics can still be resolved. Statuses come
  from RedisStorage, the same source resync_state reads, so this is best
  effort by design: whatever another service already cleared was never
  ours to find
- StatusEvents#resync_state waits on the queue draining
- 'clears the retained value for a key' passes end to end against the
  broker, so the mechanism itself works

What does not:

- 'drains queued deletions before queued messages' hangs, in the unit spec
  as well as the end to end one, so the fault is in consume_messages
  rather than anything MQTT. Suspect the interaction between the
  non-blocking drain, the three way select and closing both channels in
  stop. Guarding the closed channel in drain_deletion was not enough

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 spec/publishing/mqtt_publisher_spec.cr       | 29 ++++++++
 spec/publishing/publisher_spec.cr            | 76 ++++++++++++++++++++
 spec/spec_helper.cr                          | 11 +++
 src/source/manager.cr                        |  2 +-
 src/source/mappings.cr                       | 12 ++++
 src/source/publishing/influx_manager.cr      | 10 +++
 src/source/publishing/mqtt_broker_manager.cr | 16 +++++
 src/source/publishing/mqtt_publisher.cr      | 24 +++++++
 src/source/publishing/publisher.cr           | 54 ++++++++++++++
 src/source/publishing/publisher_manager.cr   |  7 ++
 src/source/router/driver_router.cr           | 10 +++
 src/source/router/module_router.cr           | 45 +++++++++++-
 src/source/status_events.cr                  | 19 +++++
 13 files changed, 313 insertions(+), 2 deletions(-)
 create mode 100644 spec/publishing/publisher_spec.cr

diff --git a/spec/publishing/mqtt_publisher_spec.cr b/spec/publishing/mqtt_publisher_spec.cr
index c3e7bae..0bb5f00 100644
--- a/spec/publishing/mqtt_publisher_spec.cr
+++ b/spec/publishing/mqtt_publisher_spec.cr
@@ -257,6 +257,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(
diff --git a/spec/publishing/publisher_spec.cr b/spec/publishing/publisher_spec.cr
new file mode 100644
index 0000000..e23f018
--- /dev/null
+++ b/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.delete_queue.send(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.delete_queue.send(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
diff --git a/spec/spec_helper.cr b/spec/spec_helper.cr
index 0f65112..0a2eb68 100644
--- a/spec/spec_helper.cr
+++ b/spec/spec_helper.cr
@@ -41,10 +41,21 @@ module PlaceOS::Source
 
     getter messages : Array(Publisher::Message) = [] of Publisher::Message
 
+    # Keys whose retained value was queued for removal
+    getter deletions : Array(Publisher::Message) = [] of Publisher::Message
+
     def broadcast(message : Publisher::Message)
       messages << message
     end
 
+    def broadcast_delete(message : Publisher::Message)
+      deletions << message
+    end
+
+    def deletes_pending? : Bool
+      false
+    end
+
     def start
     end
 
diff --git a/src/source/manager.cr b/src/source/manager.cr
index 014ffcb..62127b2 100644
--- a/src/source/manager.cr
+++ b/src/source/manager.cr
@@ -25,7 +25,7 @@ module PlaceOS::Source
     )
       @control_system_router = Router::ControlSystem.new(mappings, publisher_managers)
       @driver_router = Router::Driver.new(mappings, publisher_managers)
-      @module_router = Router::Module.new(mappings)
+      @module_router = Router::Module.new(mappings, publisher_managers)
       @zone_router = Router::Zone.new(mappings, publisher_managers)
       @status_events = StatusEvents.new(mappings, publisher_managers)
     end
diff --git a/src/source/mappings.cr b/src/source/mappings.cr
index 200c8e9..2f1747c 100644
--- a/src/source/mappings.cr
+++ b/src/source/mappings.cr
@@ -67,6 +67,18 @@ 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,
diff --git a/src/source/publishing/influx_manager.cr b/src/source/publishing/influx_manager.cr
index 3c26f8b..2187060 100644
--- a/src/source/publishing/influx_manager.cr
+++ b/src/source/publishing/influx_manager.cr
@@ -30,6 +30,16 @@ module PlaceOS::Source
       @publisher = InfluxPublisher.new(client, influx_bucket)
     end
 
+    # InfluxDB stores a time series rather than current state, so there is
+    # nothing retained to remove. History of a renamed Module stays under its
+    # old tags, which is what you want from a time series
+    def broadcast_delete(message : Publisher::Message)
+    end
+
+    def deletes_pending? : Bool
+      false
+    end
+
     def broadcast(message : Publisher::Message)
       publisher.message_queue.send(message)
     end
diff --git a/src/source/publishing/mqtt_broker_manager.cr b/src/source/publishing/mqtt_broker_manager.cr
index 420df5b..6f27106 100644
--- a/src/source/publishing/mqtt_broker_manager.cr
+++ b/src/source/publishing/mqtt_broker_manager.cr
@@ -26,6 +26,22 @@ module PlaceOS::Source
       end
     end
 
+    # Broadcast a deletion to each MQTT Broker
+    #
+    def broadcast_delete(message : Publisher::Message)
+      read_publishers do |publishers|
+        publishers.values.each do |publisher|
+          publisher.delete_queue.send(message)
+        end
+      end
+    end
+
+    def deletes_pending? : Bool
+      read_publishers do |publishers|
+        publishers.values.any?(&.deletes_pending?)
+      end
+    end
+
     def stats : Hash(String, UInt64)
       hash = {} of String => UInt64
       read_publishers do |publishers|
diff --git a/src/source/publishing/mqtt_publisher.cr b/src/source/publishing/mqtt_publisher.cr
index bc0327e..247f170 100644
--- a/src/source/publishing/mqtt_publisher.cr
+++ b/src/source/publishing/mqtt_publisher.cr
@@ -132,6 +132,30 @@ module PlaceOS::Source
       Log.error(exception: e) { "error while publishing Message<data=#{message.data}, key=#{key}, payload=#{message.payload}>" }
     end
 
+    # MQTT-3.3.1-7, a zero length retained payload removes what the broker holds
+    # for the topic. Anything else — including a JSON null — would simply be
+    # retained in its place, and consumers would keep seeing a value
+    def delete(message : Message) : Nil
+      key = MqttPublisher.generate_key(message.data)
+      return unless key
+
+      Log.trace { {message: "removing retained value", key: key} }
+
+      Retriable.retry(
+        max_attempts: 10,
+        base_interval: 250.milliseconds,
+        max_interval: 5.seconds,
+        on: IO::Error | MQTT::Error,
+        on_retry: ->(e : Exception, _attempt : Int32, _elapsed : Time::Span, _next : Time::Span) {
+          Log.warn(exception: e) { "MQTT delete failed, waiting for the client to reconnect" }
+        }
+      ) do
+        client.publish(topic: key, payload: "", retain: true)
+      end
+    rescue e
+      Log.error(exception: e) { "error removing retained value for #{message.data}" }
+    end
+
     # Key generation
     ###########################################################################
 
diff --git a/src/source/publishing/publisher.cr b/src/source/publishing/publisher.cr
index 5154735..954d9c0 100644
--- a/src/source/publishing/publisher.cr
+++ b/src/source/publishing/publisher.cr
@@ -12,23 +12,50 @@ module PlaceOS::Source
     )
 
     getter message_queue : Channel(Message) = Channel(Message).new(StatusEvents::BATCH_SIZE)
+
+    # 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.
+    #
+    # 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)
+
     getter processed : UInt64 = 0_u64
+    getter deleted : UInt64 = 0_u64
 
     abstract def publish(message : Message)
 
+    # Removes whatever is retained for the message's key.
+    # Only meaningful where the destination retains state, so it does nothing
+    # by default
+    def delete(message : Message) : Nil
+    end
+
     def commit : Nil
     end
 
+    # Whether every queued deletion has been dealt with. A state resync waits
+    # on this, so it cannot republish a key that is about to be removed
+    def deletes_pending? : Bool
+      !delete_queue.empty?
+    end
+
     def start
       spawn { consume_messages }
     end
 
     def stop
       message_queue.close
+      delete_queue.close
     end
 
     private def consume_messages
       while !message_queue.closed?
+        # deletions first, always, and keep going while any remain
+        next if drain_deletion
+
         select
         when message = message_queue.receive?
           if message
@@ -39,11 +66,38 @@ module PlaceOS::Source
               Log.warn(exception: error) { "publishing message: #{message}" }
             end
           end
+        when deletion = delete_queue.receive?
+          handle_deletion(deletion) if deletion
         when timeout(10.seconds)
           # commit any buffered messages that have not been published yet
           commit
         end
       end
     end
+
+    # Takes a deletion without blocking, so a pending one always goes first.
+    #
+    # NOTE:: a closed channel yields nil from `receive?` immediately, so this
+    # has to report "nothing taken" for it — otherwise the caller spins on a
+    # closed queue instead of noticing it should stop
+    private def drain_deletion : Bool
+      return false if delete_queue.closed?
+
+      select
+      when deletion = delete_queue.receive?
+        return false if deletion.nil?
+        handle_deletion(deletion)
+        true
+      else
+        false
+      end
+    end
+
+    private def handle_deletion(message : Message) : Nil
+      delete(message)
+      @deleted += 1_u64
+    rescue error
+      Log.warn(exception: error) { "deleting retained message: #{message}" }
+    end
   end
 end
diff --git a/src/source/publishing/publisher_manager.cr b/src/source/publishing/publisher_manager.cr
index 76423cf..9b52bb7 100644
--- a/src/source/publishing/publisher_manager.cr
+++ b/src/source/publishing/publisher_manager.cr
@@ -3,6 +3,13 @@ require "./publisher"
 module PlaceOS::Source
   module PublisherManager
     abstract def broadcast(message : Publisher::Message)
+
+    # Queues removal of whatever is retained for the message's key
+    abstract def broadcast_delete(message : Publisher::Message)
+
+    # Whether any publisher still has deletions to deal with
+    abstract def deletes_pending? : Bool
+
     abstract def start
     abstract def stop
 
diff --git a/src/source/router/driver_router.cr b/src/source/router/driver_router.cr
index 2af5ad3..f7f52bb 100644
--- a/src/source/router/driver_router.cr
+++ b/src/source/router/driver_router.cr
@@ -2,6 +2,7 @@ require "placeos-models/driver"
 require "placeos-resource"
 
 require "../mappings"
+require "./module_router"
 require "../publishing/publish_metadata"
 require "../publishing/publisher_manager"
 
@@ -33,6 +34,15 @@ module PlaceOS::Source::Router
       end
 
       if action.deleted?
+        # driver_id is a topic segment, so everything published by a Module of
+        # this Driver is now at a dead topic
+        module_ids = mappings.read do |state|
+          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 }
diff --git a/src/source/router/module_router.cr b/src/source/router/module_router.cr
index a57f751..e4e81fa 100644
--- a/src/source/router/module_router.cr
+++ b/src/source/router/module_router.cr
@@ -1,7 +1,9 @@
+require "placeos-driver/storage"
 require "placeos-models/module"
 require "placeos-resource"
 
 require "../mappings"
+require "../publishing/publisher_manager"
 
 module PlaceOS::Source::Router
   # Module router...
@@ -9,12 +11,43 @@ module PlaceOS::Source::Router
   # - Maintain module_id -> driver_id mapping
   class Module < Resource(Model::Module)
     private getter mappings : Mappings
+    private getter publisher_managers : Array(PublisherManager)
     Log = ::Log.for(self)
 
-    def initialize(@mappings : Mappings)
+    def initialize(@mappings : Mappings, @publisher_managers : Array(PublisherManager) = [] of PublisherManager)
       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)
@@ -27,7 +60,13 @@ module PlaceOS::Source::Router
       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
       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|
           mappings.set_system_modules(cs.id.as(String), Router::ControlSystem.system_modules(cs))
@@ -39,6 +78,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)
+
       mappings.write do |state|
         # Remove reference in drivers
         state.drivers.delete(module_id)
diff --git a/src/source/status_events.cr b/src/source/status_events.cr
index fe518cc..111c3ee 100644
--- a/src/source/status_events.cr
+++ b/src/source/status_events.cr
@@ -66,6 +66,19 @@ module PlaceOS::Source
       redis.close
     end
 
+    # Blocks until every publisher has dealt with its queued deletions
+    def await_deletions(timeout : Time::Span = 30.seconds) : Nil
+      expire = Time.utc + timeout
+
+      while publisher_managers.any?(&.deletes_pending?)
+        if Time.utc > expire
+          Log.warn { "timed out waiting for queued deletions to drain" }
+          return
+        end
+        sleep 50.milliseconds
+      end
+    end
+
     def paginate_modules(&)
       batch_size = 64
       last_created_at = Time.unix(0)
@@ -124,6 +137,12 @@ module PlaceOS::Source
     def resync_state
       return unless initial_sync_complete?
 
+      # A resync republishes current state for every Module. Queued deletions
+      # are removals of topics that no longer exist, so running before they
+      # drain would republish a key that is about to be taken away — and the
+      # delete would then remove a value that is still live
+      await_deletions
+
       Log.info { "resyncing state for new broker connection" }
 
       mods_mapped = 0_u64

From 42c5b2eb22bc692922d1cbbb990f9d9a8bada8e7 Mon Sep 17 00:00:00 2001
From: Stephen von Takach <steve@place.technology>
Date: Tue, 11 Aug 2026 09:04:47 +1000
Subject: [PATCH 2/4] fix: drain deletions by select order, and stop asking the
 channel
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two separate faults, the second hidden behind the first.

The manual non-blocking drain is gone. `select` makes a non-blocking pass
over its branches in order, so simply listing the delete queue first gives
deletions priority when both are ready — no draining needed. Verified:
two deletions and three messages queued messages-first come back
delete, delete, publish, publish, publish.

That alone did not stop the hang, because the hang was not there.
`deletes_pending?` asked `Channel#empty?`, which blocks forever on an
empty channel rather than answering. Every caller stopped dead, including
the resync gate it exists for. Outstanding deletions are now counted, so
asking costs nothing and cannot block.

Found by reproducing the consume loop in a standalone program. The
docker suite takes four minutes a cycle and told me only that something
hung; the reproduction took seconds and named the line.

52 examples, 0 failures, over two consecutive runs.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 spec/publishing/publisher_spec.cr            |  4 +-
 src/source/publishing/mqtt_broker_manager.cr |  2 +-
 src/source/publishing/publisher.cr           | 45 +++++++++-----------
 3 files changed, 24 insertions(+), 27 deletions(-)

diff --git a/spec/publishing/publisher_spec.cr b/spec/publishing/publisher_spec.cr
index e23f018..579a3e4 100644
--- a/spec/publishing/publisher_spec.cr
+++ b/spec/publishing/publisher_spec.cr
@@ -26,7 +26,7 @@ module PlaceOS::Source
       # 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.delete_queue.send(Publisher::Message.new(event, "deletion-#{i}", Time.utc)) }
+      2.times { |i| publisher.queue_delete(Publisher::Message.new(event, "deletion-#{i}", Time.utc)) }
 
       publisher.start
       sleep 200.milliseconds
@@ -43,7 +43,7 @@ module PlaceOS::Source
       event = Mappings::Metadata.new("mod-1234", "hello")
 
       publisher.deletes_pending?.should be_false
-      publisher.delete_queue.send(Publisher::Message.new(event, "pending", Time.utc))
+      publisher.queue_delete(Publisher::Message.new(event, "pending", Time.utc))
       publisher.deletes_pending?.should be_true
 
       publisher.start
diff --git a/src/source/publishing/mqtt_broker_manager.cr b/src/source/publishing/mqtt_broker_manager.cr
index 6f27106..f147053 100644
--- a/src/source/publishing/mqtt_broker_manager.cr
+++ b/src/source/publishing/mqtt_broker_manager.cr
@@ -31,7 +31,7 @@ module PlaceOS::Source
     def broadcast_delete(message : Publisher::Message)
       read_publishers do |publishers|
         publishers.values.each do |publisher|
-          publisher.delete_queue.send(message)
+          publisher.queue_delete(message)
         end
       end
     end
diff --git a/src/source/publishing/publisher.cr b/src/source/publishing/publisher.cr
index 954d9c0..e03d782 100644
--- a/src/source/publishing/publisher.cr
+++ b/src/source/publishing/publisher.cr
@@ -25,6 +25,17 @@ module PlaceOS::Source
     getter processed : UInt64 = 0_u64
     getter deleted : UInt64 = 0_u64
 
+    # NOTE:: counted rather than asking the channel. `Channel#empty?` blocks
+    # forever on an empty channel, so it cannot be used to ask whether work is
+    # outstanding
+    @pending_deletes = Atomic(Int32).new(0)
+
+    # Queues removal of whatever is retained for the message's key
+    def queue_delete(message : Message) : Nil
+      @pending_deletes.add(1)
+      delete_queue.send(message)
+    end
+
     abstract def publish(message : Message)
 
     # Removes whatever is retained for the message's key.
@@ -39,7 +50,7 @@ module PlaceOS::Source
     # Whether every queued deletion has been dealt with. A state resync waits
     # on this, so it cannot republish a key that is about to be removed
     def deletes_pending? : Bool
-      !delete_queue.empty?
+      @pending_deletes.get > 0
     end
 
     def start
@@ -53,10 +64,14 @@ module PlaceOS::Source
 
     private def consume_messages
       while !message_queue.closed?
-        # deletions first, always, and keep going while any remain
-        next if drain_deletion
-
+        # 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
             begin
@@ -66,8 +81,6 @@ module PlaceOS::Source
               Log.warn(exception: error) { "publishing message: #{message}" }
             end
           end
-        when deletion = delete_queue.receive?
-          handle_deletion(deletion) if deletion
         when timeout(10.seconds)
           # commit any buffered messages that have not been published yet
           commit
@@ -75,29 +88,13 @@ module PlaceOS::Source
       end
     end
 
-    # Takes a deletion without blocking, so a pending one always goes first.
-    #
-    # NOTE:: a closed channel yields nil from `receive?` immediately, so this
-    # has to report "nothing taken" for it — otherwise the caller spins on a
-    # closed queue instead of noticing it should stop
-    private def drain_deletion : Bool
-      return false if delete_queue.closed?
-
-      select
-      when deletion = delete_queue.receive?
-        return false if deletion.nil?
-        handle_deletion(deletion)
-        true
-      else
-        false
-      end
-    end
-
     private def handle_deletion(message : Message) : Nil
       delete(message)
       @deleted += 1_u64
     rescue error
       Log.warn(exception: error) { "deleting retained message: #{message}" }
+    ensure
+      @pending_deletes.sub(1)
     end
   end
 end

From 7e6b14b0de55bc1d5fb45387c25f84b08c534500 Mon Sep 17 00:00:00 2001
From: Stephen von Takach <steve@place.technology>
Date: Tue, 11 Aug 2026 10:07:59 +1000
Subject: [PATCH 3/4] test: rename and reindex move retained state to the new
 topic
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Covers the round trip the earlier specs only tested in halves: the old
topic cleared *and* the value present at the new one.

Both the module name and the index are topic segments, so changing either
moves every status the Module publishes. Because everything is retained,
the old topic keeps serving its last value indefinitely unless it is
cleared — a consumer would go on reading state for a path nothing
publishes to any more. Each example asserts both ends, since clearing
without republishing loses the state and republishing without clearing
leaves a ghost.

The statuses come from the Module's stored state, so the specs seed
RedisStorage the way the service reads it.

Verified by mutation: disabling remove_retained_state fails both, on the
old topic still holding its value. The shared broker and subscriber
helpers moved to spec_helper so both spec files can reach them.

54 examples, 0 failures.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 spec/publishing/mqtt_publisher_spec.cr |  91 -------------------
 spec/publishing/rename_spec.cr         | 112 +++++++++++++++++++++++
 spec/spec_helper.cr                    | 121 +++++++++++++++++++++++++
 3 files changed, 233 insertions(+), 91 deletions(-)
 create mode 100644 spec/publishing/rename_spec.cr

diff --git a/spec/publishing/mqtt_publisher_spec.cr b/spec/publishing/mqtt_publisher_spec.cr
index 0bb5f00..409b8c7 100644
--- a/spec/publishing/mqtt_publisher_spec.cr
+++ b/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
diff --git a/spec/publishing/rename_spec.cr b/spec/publishing/rename_spec.cr
new file mode 100644
index 0000000..1288d15
--- /dev/null
+++ b/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
diff --git a/spec/spec_helper.cr b/spec/spec_helper.cr
index 0a2eb68..aa53be7 100644
--- a/spec/spec_helper.cr
+++ b/spec/spec_helper.cr
@@ -36,6 +36,127 @@ module PlaceOS::Source
     end
   end
 
+  # 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
+
+  # Wraps a single publisher, so a router's deletions reach a real broker
+  class MockBrokerManager
+    include PublisherManager
+
+    def initialize(@publisher : Publisher)
+    end
+
+    def broadcast(message : Publisher::Message)
+      @publisher.message_queue.send(message)
+    end
+
+    def broadcast_delete(message : Publisher::Message)
+      @publisher.queue_delete(message)
+    end
+
+    def deletes_pending? : Bool
+      @publisher.deletes_pending?
+    end
+
+    def start
+    end
+
+    def stop
+    end
+
+    def stats : Hash(String, UInt64)
+      {} of String => UInt64
+    end
+  end
+
   class MockManager
     include PublisherManager
 

From 9079a3b02816c90a95af1316cdba09b2861c1a17 Mon Sep 17 00:00:00 2001
From: Stephen von Takach <steve@place.technology>
Date: Tue, 11 Aug 2026 11:24:43 +1000
Subject: [PATCH 4/4] style: use short block notation in the module router spec

Ameba runs in CI but not in ./test, so this slipped through when the
router specs landed on master.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 spec/router/module_router_spec.cr | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/spec/router/module_router_spec.cr b/spec/router/module_router_spec.cr
index df183f0..826de8a 100644
--- a/spec/router/module_router_spec.cr
+++ b/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