feat(api): clear and resync the MQTT brokers on request

[stakach]

Why

When a broker's retained state has drifted, there was no way to put it right short of restarting source, and even a restart can't remove what nothing maps to on a large broker. The startup sweep sees only part of a large broker because mosquitto drops what a subscriber's queue can't hold (max_queued_messages). At 3,000 systems and 15,000 modules with 2,500 deleted while source was down (about 27,500 stale topics), the sweep removed 385 with the default queue and 2,090 at 10,000.

What

• POST /api/source/v1/resync?clear= publishes every module's state and every model's metadata to the brokers again. With clear=true, everything source publishes (all placeos/+/state/# and placeos/+/metadata/#) is removed first. It runs in the background, one at a time: 202 once started, 409 while one is running.
• GET /api/source/v1/resync reports where the last one got to: state (idle, clearing, resyncing, complete or failed), started_at, finished_at, cleared, metadata and error.

Source's API is internal and unauthenticated. PlaceOS/rest-api exposes these to system admins as POST/GET /api/engine/v2/source/resync (that PR is linked below).

How clearing works:

• Event processing is held back throughout (a lock around each processing step), so nothing is published again mid-clear. Events that arrive meanwhile coalesce and are processed afterwards.
• It works in rounds. Each round takes up to 5,000 retained topics, removes them, and waits for the removals to reach the broker. It ends when a round finds nothing, so it's complete even when the broker drops part of each subscription.
• The registry of what's published is cleared with it. The resync then rebuilds it, waiting for the initial sync if that hasn't finished.

Metadata (zones, systems, drivers) is republished page by page, by key. Scope for systems and drivers comes from the mappings, so there are no per-row queries.

Results

At 3,000 systems, 15,000 modules and about 166,000 retained topics, including about 25,000 stale ones the sweep had missed:

• Clear: 165,915 topics removed in about 24s. A second request during it got 409.
• Whole run: complete in 37s.
• Broker afterwards: exactly what a fresh boot publishes (137,812 values + 3,005 metadata − 261 values of modules in unscoped systems = 140,556). Topics under the deleted systems: 0.

Testing

• ./test: 71 examples, 0 failures. New spec/clear_spec.cr clears 50 topics in rounds of 20 against the test broker and leaves a topic outside placeos/ alone. A manager spec checks one resync runs at a time.
• e2e (PlaceOS/user-interfaces stagehand local suite): 15/15 on an image of this branch, and again after a clear and resync of the stack.

🤖 Generated with Claude Code