A cluster-wide advisory lock, for the one thing that still needs one.

Propagation does not. A cascade takes no lock and a resumption takes no lock: at-least-once delivery over idempotent work is a stronger position than exactly-once scheduling, because it degrades gracefully rather than depending on job-state bookkeeping staying correct across a node death.

What still needs a lock is EXTERNAL I/O. Two nodes sweeping the same upstreams in the same minute is duplicated fetching — someone else's server, someone else's rate limit — and no amount of idempotence downstream makes those requests not happen. That concern is unrelated to propagation, which is why this survived the queue it used to live in.

A Postgres advisory lock because the requirement is exactly what they are for: cluster-wide, held on a connection, and released automatically if that connection dies — a node crashing mid-sweep does not leave the graph locked, which a lock table would.

Returns {:ok, result}, or :busy when another node holds it. Busy is not an error: the other node is doing this node's work.

case Lock.with_lock(fn -> sweep(plan) end, scope: tenant) do
  {:ok, result} -> result
  :busy -> :already_sweeping
end