diff --git a/spec/publishing/influx_ingest_spec.cr b/spec/publishing/influx_ingest_spec.cr
new file mode 100644
index 0000000..5d55217
--- /dev/null
+++ b/spec/publishing/influx_ingest_spec.cr
@@ -0,0 +1,375 @@
+require "csv"
+require "http/client"
+
+require "../spec_helper"
+
+# End to end: status written to Redis the way a driver writes it, carried by
+# `StatusEvents` to the InfluxDB the test stack runs, and read back with Flux
+module PlaceOS::Source
+  # Every point written for a system, one row per point with its fields
+  # pivoted into columns alongside the tags.
+  #
+  # NOTE:: read as plain CSV rather than with `Flux::Client#query`, which
+  # cannot parse a result holding more than one schema — and points with
+  # different fields and tags come back as one table schema each
+  def self.influx_points(system_id : String) : Array(Hash(String, String))
+    flux = <<-FLUX
+      from(bucket: "#{INFLUX_BUCKET}")
+        |> range(start: 0)
+        |> filter(fn: (r) => r.pos_system == "#{system_id}")
+        |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
+      FLUX
+
+    uri = URI.parse(INFLUX_HOST.not_nil!)
+    headers = HTTP::Headers{
+      "Authorization" => "Token #{INFLUX_API_KEY}",
+      "Accept"        => "application/csv",
+      "Content-Type"  => "application/json",
+    }
+    body = {query: flux, dialect: {annotations: [] of String, header: true}}.to_json
+    response = HTTP::Client.new(uri).post("/api/v2/query?org=#{URI.encode_www_form(INFLUX_ORG)}", headers, body)
+    raise "query failed: #{response.status_code} #{response.body}" unless response.success?
+
+    # each table is a header row and its records, tables separated by a blank line
+    response.body.split(/\r?\n\r?\n/).flat_map do |table|
+      next [] of Hash(String, String) if table.strip.empty?
+      rows = CSV.parse(table.strip)
+      header = rows.shift
+      rows.map { |row| header.zip(row).to_h.reject { |_key, value| value.empty? } }
+    end
+  end
+
+  def self.eventually(timeout : Time::Span = 20.seconds, interval : Time::Span = 250.milliseconds, &)
+    expire = Time.instant + timeout
+    loop do
+      return if yield
+      raise "condition not met within #{timeout}" if Time.instant > expire
+      sleep interval
+    end
+  end
+
+  # A Module's status events, as published by a driver, flowing to `managers`
+  class IngestPipeline
+    getter module_id : String
+    getter system_id : String = "cs-#{UUID.random}"
+    # the measurement, unique so field types never clash with earlier runs
+    getter module_name : String = "ingest_#{Random::Secure.hex(6)}"
+    getter status_events : StatusEvents
+    getter mappings : Mappings
+
+    def initialize(managers : Array(PublisherManager), @module_id = "mod-#{UUID.random}")
+      state = PlaceOS::Source.mock_state(
+        module_id: module_id,
+        module_name: module_name,
+        control_system_id: system_id,
+        org_id: "org-#{UUID.random}",
+      )
+      @mappings = Mappings.new(state)
+      @status_events = StatusEvents.new(@mappings, managers)
+      spawn { @status_events.start }
+      PlaceOS::Source.eventually(10.seconds) { @status_events.initial_sync_complete? }
+      # time for the pattern subscription to be established
+      sleep 500.milliseconds
+    end
+
+    def store : PlaceOS::Driver::RedisStorage
+      PlaceOS::Driver::RedisStorage.new(module_id)
+    end
+
+    def set(status : String, value) : Nil
+      store[status] = value.to_json
+    end
+
+    # The MQTT topic a status is published to
+    def topic(status : String) : String
+      MqttPublisher.generate_key(mappings.status_events?(module_id, status).not_nil!.first).not_nil!
+    end
+
+    def points : Array(Hash(String, String))
+      PlaceOS::Source.influx_points(system_id)
+    end
+
+    def stop
+      status_events.stop
+      store.clear rescue nil
+    end
+  end
+
+  # Relays TCP to the MQTT broker until it is cut, standing in for a broker
+  # that goes away under a connected publisher. Once cut nothing listens on the
+  # port, so reconnection attempts are refused
+  class BrokerRelay
+    getter port : Int32
+
+    @sockets = [] of TCPSocket
+    @lock = Mutex.new
+
+    def initialize(@upstream_host : String, @upstream_port : Int32)
+      @server = TCPServer.new("127.0.0.1", 0)
+      @port = @server.local_address.port
+      spawn { accept_loop }
+    end
+
+    private def accept_loop
+      while client = @server.accept?
+        upstream = TCPSocket.new(@upstream_host, @upstream_port)
+        @lock.synchronize { @sockets << client << upstream }
+        spawn { pipe(client, upstream) }
+        spawn { pipe(upstream, client) }
+      end
+    end
+
+    private def pipe(from, to)
+      IO.copy(from, to)
+    rescue
+    ensure
+      from.close rescue nil
+      to.close rescue nil
+    end
+
+    def cut : Nil
+      @server.close rescue nil
+      @lock.synchronize do
+        @sockets.each { |socket| socket.close rescue nil }
+        @sockets.clear
+      end
+    end
+
+    def broker : Model::Broker
+      broker = Model::Broker.new(
+        name: "relay-#{UUID.random}",
+        host: "127.0.0.1",
+        port: port,
+        auth_type: :no_auth,
+      )
+      broker.id = "spec-relay-#{UUID.random}"
+      broker
+    end
+  end
+
+  def self.influx_manager : InfluxManager
+    manager = InfluxManager.new(INFLUX_HOST.not_nil!, INFLUX_API_KEY.not_nil!)
+    # rather than the production interval, so specs are not waiting on it
+    manager.publisher.commit_interval = 500.milliseconds
+    manager
+  end
+
+  describe "InfluxDB ingest" do
+    it "writes each supported payload format to InfluxDB" do
+      influx = influx_manager
+      influx.start
+      pipeline = IngestPipeline.new([influx] of PublisherManager)
+
+      begin
+        pipeline.set "power", true
+        pipeline.set "volume", 42.5
+        pipeline.set "input", "hdmi"
+        pipeline.set "owner", "someone@example.com"
+        pipeline.set "climate", {temperature: 21.5, humidity: 40, measurement: "climate"}
+        pipeline.set "rooms", {"r1" => {people: 3}, "r2" => {people: 5}}
+        pipeline.set "desks", [{desk: "d1", occupied: true}, {desk: "d2", occupied: false}]
+        seen = Time.utc.to_unix - 60
+        pipeline.set "locations", {
+          ts_hint:      "complex",
+          measurement:  "locations",
+          ts_map:       {x: "xloc"},
+          ts_tag_keys:  ["mac"],
+          ts_tags:      {pos_building: "custom-building"},
+          ts_fields:    {source: "wifi"},
+          ts_timestamp: "seen",
+          value:        [{x: 1.5, mac: "aa", seen: seen}, {x: 2.5, mac: "bb", seen: seen + 1}],
+        }
+        # neither of these can be stored, so nothing is written for them
+        pipeline.set "nothing", nil
+        pipeline.set "control", "bad\u0001value"
+
+        points = [] of Hash(String, String)
+        eventually { (points = pipeline.points).size >= 11 }
+        sleep 1.second
+        points = pipeline.points
+
+        by_key = points.group_by { |point| point["pos_key"] }
+        by_key.keys.sort!.should eq %w[climate desks input locations owner power rooms volume]
+
+        # every point carries the hierarchy and module tags
+        points.each do |point|
+          point["pos_module"].should eq pipeline.module_name
+          point["pos_index"].should eq "1"
+          point["pos_driver"].should eq "12345"
+        end
+
+        # scalars, one field named after the status
+        power = by_key["power"].first
+        power["_measurement"].should eq pipeline.module_name
+        power["power"].should eq "true"
+        power["pos_level"].should eq "nek"
+        by_key["volume"].first["volume"].should eq "42.5"
+        by_key["input"].first["input"].should eq "hdmi"
+
+        # strings matching the default filters are hashed before storage
+        owner = by_key["owner"].first["owner"]
+        owner.should_not eq "someone@example.com"
+        owner.size.should eq 64
+
+        # a hash, one field per key, with a measurement of its own
+        climate = by_key["climate"].first
+        climate["_measurement"].should eq "climate"
+        climate["temperature"].should eq "21.5"
+        climate["humidity"].should eq "40"
+
+        # a hash of hashes, one point per entry
+        rooms = by_key["rooms"].sort_by(&.["pos_uniq"])
+        rooms.map(&.["parent_hash_key"]).should eq %w[r1 r2]
+        rooms.map(&.["people"]).should eq %w[3 5]
+
+        # an array of hashes, one point per element
+        desks = by_key["desks"].sort_by(&.["pos_uniq"])
+        desks.map(&.["desk"]).should eq %w[d1 d2]
+        desks.map(&.["occupied"]).should eq %w[true false]
+
+        # complex: mapped field names, fields promoted to tags, tags and fields
+        # added to every point, and a timestamp taken from the value
+        locations = by_key["locations"].sort_by(&.["pos_uniq"])
+        locations.size.should eq 2
+        locations.each do |point|
+          point["_measurement"].should eq "locations"
+          point["pos_building"].should eq "custom-building"
+          point["source"].should eq "wifi"
+        end
+        locations.map(&.["xloc"]).should eq %w[1.5 2.5]
+        locations.map(&.["mac"]).should eq %w[aa bb]
+        locations.map { |point| Time.parse_rfc3339(point["_time"]).to_unix }.should eq [seen, seen + 1]
+      ensure
+        pipeline.stop
+        influx.stop
+      end
+    end
+
+    it "writes a slow trickle of events without waiting for a full batch" do
+      influx = influx_manager
+      influx.start
+      pipeline = IngestPipeline.new([influx] of PublisherManager)
+
+      begin
+        # an event more often than the commit interval, never enough to fill a
+        # batch: the buffer is never idle for a whole interval
+        spawn do
+          20.times do |index|
+            pipeline.set "trickle", index
+            sleep 200.milliseconds
+          end
+        end
+
+        eventually(3.seconds) { pipeline.points.size > 0 }
+      ensure
+        pipeline.stop
+        influx.stop
+      end
+    end
+
+    it "writes buffered points when stopped" do
+      influx = influx_manager
+      influx.publisher.commit_interval = 1.minute
+      influx.start
+      pipeline = IngestPipeline.new([influx] of PublisherManager)
+
+      begin
+        pipeline.set "power", true
+        eventually { influx.publisher.processed > 0 }
+        influx.stop
+        eventually(5.seconds) { pipeline.points.size == 1 }
+      ensure
+        pipeline.stop
+      end
+    end
+
+    describe "with an MQTT broker that is not working" do
+      it "keeps writing to InfluxDB after the broker drops away" do
+        relay = BrokerRelay.new(test_broker.host, test_broker.port)
+        broker = relay.broker
+        mqtt = MqttBrokerManager.new
+        mqtt.process_resource(:created, broker).should eq Resource::Result::Success
+
+        influx = influx_manager
+        influx.start
+        # MQTT first, as the service registers them
+        pipeline = IngestPipeline.new([mqtt, influx] of PublisherManager)
+
+        begin
+          # the broker works to begin with
+          subscriber = PlaceOS::Source.subscriber
+          key = pipeline.topic("s_first")
+          received = PlaceOS::Source.subscribe_channel(subscriber, key)
+          sleep 200.milliseconds
+          pipeline.set "s_first", true
+          PlaceOS::Source.take(received, key)
+          subscriber.disconnect rescue nil
+
+          relay.cut
+
+          # comfortably more than a publisher's queue holds
+          count = StatusEvents::BATCH_SIZE * 3
+          count.times { |index| pipeline.set "s_#{index}", index }
+
+          eventually(20.seconds) { pipeline.points.size >= count + 1 }
+          pipeline.points.size.should eq count + 1
+        ensure
+          pipeline.stop
+          influx.stop
+          relay.cut
+          mqtt.process_resource(:deleted, broker) rescue nil
+        end
+      end
+
+      it "writes to InfluxDB when the broker never connects" do
+        relay = BrokerRelay.new(test_broker.host, test_broker.port)
+        relay.cut
+        broker = relay.broker
+        mqtt = MqttBrokerManager.new
+        mqtt.process_resource(:created, broker).should eq Resource::Result::Error
+
+        influx = influx_manager
+        influx.start
+        pipeline = IngestPipeline.new([mqtt, influx] of PublisherManager)
+
+        begin
+          count = StatusEvents::BATCH_SIZE * 3
+          count.times { |index| pipeline.set "s_#{index}", index }
+
+          eventually(20.seconds) { pipeline.points.size >= count }
+          pipeline.points.size.should eq count
+        ensure
+          pipeline.stop
+          influx.stop
+        end
+      end
+    end
+
+    # A resync republishes the current value of every status so MQTT brokers
+    # retain it. InfluxDB holds a time series: the same values again, stamped
+    # now, would record changes that never happened
+    it "does not write a broker resync to InfluxDB" do
+      mod = Model::Generator.module.save!
+      influx = influx_manager
+      influx.start
+      mqtt = MockManager.new
+      pipeline = IngestPipeline.new([mqtt, influx] of PublisherManager, mod.id.as(String))
+
+      begin
+        pipeline.set "power", true
+        eventually { pipeline.points.size == 1 }
+
+        mqtt.messages.clear
+        pipeline.status_events.resync_state
+
+        eventually(5.seconds) { mqtt.messages.any? { |message| message.data.as?(Mappings::Status).try(&.status) == "power" } }
+        sleep 1.second
+        pipeline.points.size.should eq 1
+      ensure
+        pipeline.stop
+        influx.stop
+        mod.destroy
+      end
+    end
+  end
+end
diff --git a/spec/publishing/influx_publisher_spec.cr b/spec/publishing/influx_publisher_spec.cr
index efe5c02..321be15 100644
--- a/spec/publishing/influx_publisher_spec.cr
+++ b/spec/publishing/influx_publisher_spec.cr
@@ -4,6 +4,20 @@ require "time"
 module PlaceOS::Source
   describe InfluxPublisher do
     describe "#transform" do
+      it "hashes values matching the default filters" do
+        status_event = Mappings.new(mock_state).status_events?("mod-1234", "state").not_nil!.first
+        message = Publisher::Message.new(status_event, {
+          email: "someone@example.com",
+          note:  "not an address",
+        }.to_json, timestamp: Time::UNIX_EPOCH)
+
+        point = InfluxPublisher.transform(message).first
+        email = point.fields["email"].as(String)
+        email.should_not eq "someone@example.com"
+        email.size.should eq 64
+        point.fields["note"].should eq "not an address"
+      end
+
       it "ignores PlaceOS Metadata events" do
         message = Publisher::Message.new(
           data: Mappings::Metadata.new("some-model-id"),
diff --git a/spec/reconcile_spec.cr b/spec/reconcile_spec.cr
index 2075777..417f08d 100644
--- a/spec/reconcile_spec.cr
+++ b/spec/reconcile_spec.cr
@@ -107,7 +107,7 @@ module PlaceOS::Source
       place(mappings, [] of {String, String})
       events.reconcile_now(Set{a})
 
-      keyed(manager.deletions).keys.sort.should eq ["Display/1/connected", "Display/1/power"]
+      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
diff --git a/src/source/publishing/influx_manager.cr b/src/source/publishing/influx_manager.cr
index 2187060..667a4b5 100644
--- a/src/source/publishing/influx_manager.cr
+++ b/src/source/publishing/influx_manager.cr
@@ -40,6 +40,12 @@ module PlaceOS::Source
       false
     end
 
+    # A resync is the current value again, stamped now: written here it would
+    # record changes that never happened, for every status at once
+    def retains_state? : Bool
+      false
+    end
+
     def broadcast(message : Publisher::Message)
       publisher.message_queue.send(message)
     end
diff --git a/src/source/publishing/influx_publisher.cr b/src/source/publishing/influx_publisher.cr
index 0a051a6..5c1720f 100644
--- a/src/source/publishing/influx_publisher.cr
+++ b/src/source/publishing/influx_publisher.cr
@@ -127,15 +127,37 @@ module PlaceOS::Source
     # Generate an InfluxDB Point from an mqtt key + payload
     #
     def self.transform(message : Publisher::Message) : Array(Flux::Point)
+      points = build_points(message)
+      points.each { |point| filter_values(point) }
+      points
+    end
+
+    # Hashes any field or tag that matches the default filters
+    #
+    # NOTE:: per value, once parsed. The filters are anchored, and were matched
+    # against the raw JSON — where a string value is still in quotes — so
+    # nothing ever matched and values such as email addresses were stored
+    protected def self.filter_values(point : Flux::Point) : Nil
+      point.fields.transform_values! do |value|
+        value.is_a?(String) ? filter_string(value) : value
+      end
+      point.tags.transform_values! { |value| filter_string(value) }
+    end
+
+    private FILTER = Regex.union(DEFAULT_FILTERS)
+
+    protected def self.filter_string(value : String) : String
+      value.gsub(FILTER) { |match_string, _| hmac_sha256(match_string) }
+    end
+
+    protected def self.build_points(message : Publisher::Message) : Array(Flux::Point)
       timestamp = message.timestamp
       data = message.data
 
       # Only Module status events are persisted
       return [] of Flux::Point unless data.is_a? Mappings::Status
 
-      payload = message.payload.presence.try &.gsub(Regex.union(DEFAULT_FILTERS)) do |match_string, _|
-        hmac_sha256(match_string)
-      end
+      payload = message.payload.presence
 
       # NOTE:: we should try and detect unsupport string values to prevent errors
       # Influx doesn't support `nil`
@@ -256,19 +278,12 @@ module PlaceOS::Source
         next if compacted.empty?
         measurement = default_measurement || data.module_name
 
-        override_timestamp = nil
-        if time_key = raw.ts_timestamp
-          if time = compacted.delete(time_key).as?(Float64)
-            override_timestamp = Time.unix time.to_i64
-          end
-        end
-
         # Must include a `pos_uniq` tag for seperating points
         # as per: https://docs.influxdata.com/influxdb/v2.0/write-data/best-practices/duplicate-points/#add-an-arbitrary-tag
         local_tags = tags.dup
         local_tags["pos_uniq"] = index.to_s
 
-        if point = build_custom_point(measurement, data, fields, local_tags, compacted, override_timestamp || timestamp, ts_map, raw.ts_tag_keys)
+        if point = build_custom_point(measurement, data, fields, local_tags, compacted, point_timestamp(compacted, raw.ts_timestamp) || timestamp, ts_map, raw.ts_tag_keys)
           points << point
         end
       end
@@ -276,6 +291,14 @@ module PlaceOS::Source
       points
     end
 
+    # The point's own timestamp, taken out of its values, when it has one
+    protected def self.point_timestamp(values, time_key : String?) : Time?
+      return unless time_key
+      if time = values.delete(time_key).as?(Float64)
+        Time.unix time.to_i64
+      end
+    end
+
     protected def self.build_custom_point(measurement, data, fields, local_tags, compacted, timestamp, ts_map, ts_tag_keys)
       measurement_value = measurement
       # Add the fields
diff --git a/src/source/publishing/mqtt_publisher.cr b/src/source/publishing/mqtt_publisher.cr
index cb13f89..d6b79bc 100644
--- a/src/source/publishing/mqtt_publisher.cr
+++ b/src/source/publishing/mqtt_publisher.cr
@@ -156,6 +156,12 @@ module PlaceOS::Source
             Log.warn(exception: e) { "MQTT publish failed, waiting for the client to reconnect" }
           }
         ) do
+          # NOTE:: nothing waits on a broker that is down. Retrying every
+          # message until it returned held this queue for half a minute each,
+          # which filled it and blocked `broadcast` — and with it every other
+          # sink, InfluxDB included. Reconnecting republishes all state
+          # (`on_reconnect`), which covers whatever is skipped here
+          return if client.closed?
           client.publish(topic: key, payload: payload, retain: true)
         end
       end
@@ -190,6 +196,8 @@ module PlaceOS::Source
           Log.warn(exception: e) { "MQTT delete failed, waiting for the client to reconnect" }
         }
       ) do
+        # down: the sweep that follows reconnecting removes it (see `publish`)
+        return if client.closed?
         client.publish(topic: key, payload: "", retain: true)
       end
     rescue e
@@ -226,7 +234,7 @@ module PlaceOS::Source
       last = Time.instant
 
       client.subscribe(filter) do |topic, payload, retained|
-        next unless retained && !payload.empty?
+        next if !retained || payload.empty?
         wanted = keep.call(topic)
         lock.synchronize do
           topics << topic if wanted && topics.size < max
diff --git a/src/source/publishing/publisher.cr b/src/source/publishing/publisher.cr
index 3c90b5c..0fab117 100644
--- a/src/source/publishing/publisher.cr
+++ b/src/source/publishing/publisher.cr
@@ -30,6 +30,9 @@ module PlaceOS::Source
     # same topic is, by construction, a value that belongs there
     getter message_queue : Channel(Message | Deletion) = Channel(Message | Deletion).new(StatusEvents::BATCH_SIZE)
 
+    # How long buffered messages may wait before they are committed
+    property commit_interval : Time::Span = 10.seconds
+
     getter processed : UInt64 = 0_u64
     getter deleted : UInt64 = 0_u64
 
@@ -83,6 +86,7 @@ module PlaceOS::Source
     end
 
     private def consume_messages
+      committed = Time.instant
       while !message_queue.closed?
         select
         when item = message_queue.receive?
@@ -98,11 +102,20 @@ module PlaceOS::Source
             handle_deletion(item)
           in Nil
           end
-        when timeout(10.seconds)
-          # commit any buffered messages that have not been published yet
+        when timeout(commit_interval)
+        end
+
+        # NOTE:: on age, not only once the queue goes quiet. A steady trickle
+        # never leaves it idle for a whole interval, so messages sat buffered
+        # until a full batch arrived
+        if Time.instant - committed >= commit_interval
           commit
+          committed = Time.instant
         end
       end
+    ensure
+      # whatever is still buffered when the publisher stops
+      commit
     end
 
     private def handle_deletion(deletion : Deletion) : Nil
diff --git a/src/source/publishing/publisher_manager.cr b/src/source/publishing/publisher_manager.cr
index 9b52bb7..6f8d490 100644
--- a/src/source/publishing/publisher_manager.cr
+++ b/src/source/publishing/publisher_manager.cr
@@ -14,5 +14,11 @@ module PlaceOS::Source
     abstract def stop
 
     abstract def stats : Hash(String, UInt64)
+
+    # Whether the sink holds the current value of each status, rather than a
+    # history of them. Only these are sent a resync
+    def retains_state? : Bool
+      true
+    end
   end
 end
diff --git a/src/source/router/control_system_router.cr b/src/source/router/control_system_router.cr
index 2c0f408..dc75763 100644
--- a/src/source/router/control_system_router.cr
+++ b/src/source/router/control_system_router.cr
@@ -86,7 +86,7 @@ module PlaceOS::Source::Router
           mappings.write { |state| state.system_zones[control_system_id] = zones }
           mappings.set_system_modules(control_system_id, modules)
         else
-          mappings.write { |state| state.system_zones.delete(control_system_id) }
+          mappings.write(&.system_zones.delete(control_system_id))
           mappings.remove_system_modules(control_system_id)
         end
 
diff --git a/src/source/status_events.cr b/src/source/status_events.cr
index 8fdfdf0..b14c6a7 100644
--- a/src/source/status_events.cr
+++ b/src/source/status_events.cr
@@ -160,7 +160,7 @@ module PlaceOS::Source
         store = PlaceOS::Driver::RedisStorage.new(module_id)
         store.each do |key, value|
           status_updated += 1_u64
-          add_event({source: :db, mod_id: module_id, status: key}, {pattern: pattern, payload: value, timestamp: Time.utc})
+          add_event({source: :resync, mod_id: module_id, status: key}, {pattern: pattern, payload: value, timestamp: Time.utc})
         end
 
         # Backpressure if event container is growing too fast
@@ -255,13 +255,14 @@ module PlaceOS::Source
 
     private def process_batch(batch)
       batch.each do |(key, value)|
-        process_pevent(value[:pattern], key[:mod_id], key[:status], value[:payload], value[:timestamp])
+        managers = key[:source] == :resync ? publisher_managers.select(&.retains_state?) : publisher_managers
+        process_pevent(value[:pattern], key[:mod_id], key[:status], value[:payload], value[:timestamp], managers)
       end
     rescue error
       Log.error(exception: error) { "Error processing event batch" }
     end
 
-    protected def process_pevent(pattern : String, module_id : String, status : String, payload : String, timestamp : Time)
+    protected def process_pevent(pattern : String, module_id : String, status : String, payload : String, timestamp : Time, managers : Array(PublisherManager) = publisher_managers)
       events = mappings.status_events?(module_id, status)
 
       Log.debug { {
@@ -280,7 +281,7 @@ module PlaceOS::Source
         end
 
         message = Publisher::Message.new(event, payload, timestamp)
-        publisher_managers.each do |manager|
+        managers.each do |manager|
           Log.trace { "broadcasting message to #{manager.class}" }
           manager.broadcast(message)
         end