mule/runtime/stager

The Stager promotes due scheduled/retryable jobs to available on a timer and wakes the queues’ producers. Because staging is an UPDATE (not an INSERT), no AFTER-INSERT trigger sees it, so the wake-up notification is published here explicitly.

Only the LEADER stages, in either mode (stager.ex stage_and_notify) — the leadership read is a total cell lookup. The mode, chosen each tick from Sonar’s notifier-health status, decides how the wake-up travels:

• Global — the leader wakes every queue that has available jobs (engine check_available), cluster-wide through the notifier; other nodes do nothing and rely on that fan-out. Used when the notifier reaches other nodes (Clustered), or when alone but leader (Solitary + leader).
• Local — every node wakes its OWN producers directly through the registry, bypassing the broken notifier (stager.ex notify_queues’s :local sends straight to registered pids). Used when the notifier is not delivering cross-node (Isolated).

A staging (or check_available) error falls back to forcing a wake-up anyway — safe_notify, dropping to the direct local sends when the global notify’s own database read also fails — so jobs keep processing through database hiccups instead of stranding until the next tick. Each tick reports through the [:mule, :plugin] span like every plugin (stager.ex spans stage_and_notify), so those errors stay observable.

Until Sonar has classified the notifier (Unknown, e.g. no Sonar running) the mode is left unchanged from its Local start, so a node with no cluster context still serves its own jobs.

The clustered Postgres engine (from_connection_clustered) additionally folds a pg_notify for the queues staged THIS tick into the promote UPDATE — one transaction, SQL lease-guarded. The check_available pass then re-wakes those queues (plus any with older available jobs); producers coalesce the duplicate through their dispatch-cooldown timer.

pub type Message {
  Stage
}

pub type Mode {
  Local
  Global
}

pub type State {
  State(
    config: config.Config,
    self_subject: process.Subject(Message),
    mode: Mode,
  )
}

pub fn start(
  config config: config.Config,
) -> Result(
  actor.Started(process.Subject(Message)),
  actor.StartError,
)