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

[stakach]

Everything source publishes is retained, so a topic holds the current value of a status indefinitely. Nothing cleared those values when a topic stopped being valid, which meant a renamed, reindexed or destroyed Module left its last value sitting at the old path forever — consumers went on reading state for something that no longer publishes there.

54 examples, 0 failures, over repeated full runs against the broker.

What happens now

When a Module is renamed, reindexed or destroyed — or its Driver is destroyed — the keys it currently publishes are collected and queued for removal, and the queue is drained ahead of new messages.

Order matters here. A rename removes the old topic and republishes under the new one; if a publish won that race the deletion would take the new value straight back out again. select makes a non-blocking pass over its branches in order, so listing the delete queue first is what gives deletions priority. A state resync also waits for the queue to drain, so it cannot republish a key that is about to be removed.

Clearing is 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 and consumers would keep seeing a value.

Collection is best effort by design, as agreed. The statuses come from the Module's stored state, the same source resync_state reads, and another service may already have cleared it by the time a delete reaches us. Whatever is still there gets removed; anything already gone was never ours to find.

The keys are collected before the mapping is rewritten, while the old topics can still be resolved. That ordering is the whole trick — afterwards there is no way to know what the old topic was.

InfluxDB gets nothing, deliberately. A time series has no current value to remove, and a renamed Module's history belongs under the tags it was recorded with.

Two bugs found on the way

• 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 counted now.
• The first attempt drained the queue manually with a non-blocking select. Reordering the main select does the same job without the extra construct.

The first commit's message says the branch does not work. It was fixed in cc96efb; the message is left as it was written rather than rewritten after the fact.

Testing

• Rename and reindex round trips against the broker, each asserting both ends: the old topic cleared and the value present at the new one. Clearing without republishing loses the state, republishing without clearing leaves a ghost, and a spec checking one end would pass for either failure.
• Mutation tested — disabling remove_retained_state fails both, on the old topic still holding its value. Worth doing, since two specs earlier in this work passed for the wrong reason.
• Queue priority and drain reporting are deterministic unit specs rather than broker round trips, so they are fast and cannot flake.
• MockBrokerManager lets a router's deletions reach a real publisher; shared broker and subscriber helpers moved to spec_helper.

Not covered

• The best effort destroy path where storage has already been cleared by another service. The code rescues and treats an empty status list as nothing to do, but no spec drives that race.
• driver_router builds a second Router::Module purely to reuse remove_retained_state. It works, but the method would sit better somewhere neutral.

The branch is still named wip, which it no longer is.

🤖 Generated with Claude Code

[github-actions]

Hey, your PR is titled Clear retained state when a topic moves or goes away.
This is not a valid conventional commit.
Since we use well-formed PR titles for our changelogs, not fixing this can introduce problems :'(