diff --git a/spec/publishing/mqtt_publisher_spec.cr b/spec/publishing/mqtt_publisher_spec.cr
index c3e7bae..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
@@ -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(
diff --git a/spec/publishing/publisher_spec.cr b/spec/publishing/publisher_spec.cr
new file mode 100644
index 0000000..579a3e4
--- /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.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
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/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
 
diff --git a/spec/spec_helper.cr b/spec/spec_helper.cr
index 0f65112..aa53be7 100644
--- a/spec/spec_helper.cr
+++ b/spec/spec_helper.cr
@@ -36,15 +36,147 @@ 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
 
     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..f147053 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.queue_delete(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..e03d782 100644
--- a/src/source/publishing/publisher.cr
+++ b/src/source/publishing/publisher.cr
@@ -12,24 +12,66 @@ 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
+
+    # 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.
+    # 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
+      @pending_deletes.get > 0
+    end
+
     def start
       spawn { consume_messages }
     end
 
     def stop
       message_queue.close
+      delete_queue.close
     end
 
     private def consume_messages
       while !message_queue.closed?
+        # 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
@@ -45,5 +87,14 @@ module PlaceOS::Source
         end
       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
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