mule

Types

A running instance’s handle: the assembled Config, which carries everything the public API consumes (engine, notifier, queues). The root supervisor is deliberately not carried — nothing reads it, and whereis answers pid questions by instance name — which is what lets named and handle build an equivalent handle without having started the instance.

pub type Mule {
  Mule(config: config.Config)
}

Constructors

Everything needed to start a Mule instance. The caller wires the driver directly: engine + notifier are the records the runtime consumes (the Gleam analog of Oban’s engine: + notifier: options), built at the call site from a concrete driver (e.g. engines/in_memory.new paired with notifiers/isolated.new). mule supervises notifier_child; the engine’s backing resource (the in-memory actor or the Postgres pool) is started by the caller’s own supervision tree — mirroring how Oban references your already-running Repo rather than starting it.

pub type MuleSpec {
  MuleSpec(
    name: String,
    engine: engine.Engine,
    notifier: notifier.Notifier,
    notifier_child: supervision.ChildSpecification(Nil),
    peer: peer.Peer,
    peer_child: option.Option(supervision.ChildSpecification(Nil)),
    registry: worker_registry.Registry,
    queues: List(#(String, Int)),
    stage_interval: duration.Duration,
    producer_poll_interval: duration.Duration,
    dispatch_cooldown: duration.Duration,
    shutdown_grace_period: duration.Duration,
    plugins: List(plugin.Plugin),
    testing: TestingMode,
  )
}

Constructors

Scoping options shared by pause/resume/scale/start/stop — mule.ex’s local_only / node options. LocalOnly takes precedence over OnNode regardless of order (mule.ex’s scope_signal cond checks local_only first); duplicate OnNodes keep the first, like opts[:node].

pub type SignalOption {
  LocalOnly
  OnNode(node: String)
}

Constructors

  • LocalOnly
  • OnNode(node: String)

The spec-side testing switch (config.ex testing: values). Manual and Inline clear queues + plugins and skip the Stager (the typed stage_interval: :infinity analog) — nothing dispatches on its own. TestingInline additionally routes every public-API engine call through the inline engine, so inserts execute synchronously in the inserting process and return a TERMINAL job without persistence. The Testing* prefix avoids ambiguity against config.Disabled/Manual/Inline in files importing both.

pub type TestingMode {
  TestingDisabled
  TestingManual
  TestingInline
}

Constructors

  • TestingDisabled
  • TestingManual
  • TestingInline

Values

pub fn all_jobs(
  instance: Mule,
  query: job.Query,
) -> Result(List(job.Job), engine.EngineError)

Fetch all jobs matching the filter (Oban.all_jobs/2), ordered by id (deterministic where Elixir’s Repo.all is unordered).

pub fn cancel_all_jobs(
  instance: Mule,
  query: job.Query,
) -> Result(Int, engine.EngineError)

Cancel all matching cancellable jobs (Oban.cancel_all_jobs/2), then broadcast ONE Pkill carrying every id that was executing when cancelled — database-then-broadcast, using the pre-transition state the engine returns, batched exactly like mule.ex’s single Notifier.notify payload list. Returns how many were cancelled.

pub fn cancel_job(
  instance: Mule,
  job_id: Int,
) -> Result(Nil, engine.EngineError)

Cancel an executing, available, scheduled or retryable job and mark it cancelled so it will not run. Delegates through the bulk cancel with an id-only filter, exactly like Oban.cancel_job/2 (mule.ex delegates to cancel_all_jobs): database-then-broadcast, with a Pkill published ONLY for a row that was executing when cancelled — a scheduled/available/retryable or already-cancelled job has no process to kill and broadcasts nothing. One deliberate divergence: an unknown id surfaces as Error(JobNotFound) where Elixir silently returns :ok.

pub fn check_all_queues(
  instance: Mule,
) -> List(engine.QueueState)

Check every locally running producer, sorted by queue name (Oban.check_all_queues/1, which sorts too). Producers that die mid-call are skipped, like safe_check. Checks run sequentially where mule.ex uses Task.async_stream — fine at introspection cadence (parity §9).

pub fn check_queue(
  instance: Mule,
  queue: String,
) -> option.Option(engine.QueueState)

The state of a locally running queue — limit, paused, executing job ids, started/updated timestamps (Oban.check_queue/2). None when the queue is not running locally or its producer died mid-call; the rescue around the producer call is mule.ex’s safe_check catch :exit.

pub fn config(
  instance_name instance_name: String,
) -> Result(config.Config, Nil)

The running instance’s assembled Config, for holders of only the instance name (Oban.config/1 via Oban.Registry.config). Result instead of Elixir’s raise — parity §9’s typed-accessor note.

pub fn delete_all_jobs(
  instance: Mule,
  query: job.Query,
) -> Result(Int, engine.EngineError)

Delete all matching non-executing jobs (Oban.delete_all_jobs/2). Returns how many were deleted.

pub fn delete_job(
  instance: Mule,
  job_id: Int,
) -> Result(Nil, engine.EngineError)

Delete a job that is not currently executing (Oban.delete_job/2). An unknown id is a silent Ok(Nil), like retry_job.

pub fn drain_queue(
  instance: Mule,
  queue: String,
) -> Result(drainer.DrainResult, engine.EngineError)

Synchronously execute a queue’s available jobs in the calling process, counting terminal states (Oban.drain_queue/2 with its defaults: no limit, no recursion, safe, no scheduled staging). Uses the same executor as regular dispatch — failures are rescued and acked as retryable with real backoff — and fetches through the engine directly, so the queue need not be running (manual testing mode’s whole point).

pub fn drain_queue_with(
  instance: Mule,
  queue: String,
  options: List(drainer.DrainOption),
) -> Result(drainer.DrainResult, engine.EngineError)

drain_queue with drainer options: WithLimit caps each fetch, WithScheduled/WithScheduledBefore promote the queue’s scheduled/retryable rows first, WithRecursion keeps draining while the counts change (so jobs enqueued by drained jobs run too), and WithSafety(False) re-raises a drained job’s crash in the calling process AFTER acking it. Errors surface as a Result where Elixir raises — consistent with every other public function.

pub fn handle(spec: MuleSpec) -> Mule

The handle for spec, built without starting anything — the value-threading counterpart to named, for when a handle must exist before (or above) the supervision tree that starts the instance. assemble_config is deterministic, so this equals the running instance’s handle whenever an instance started from the same spec value is up on this VM. Liveness is NOT checked: calls through a handle whose instance is not running fail at call time.

pub fn insert(
  instance: Mule,
  worker: worker.Worker(a),
  args: a,
) -> Result(job.Job, engine.EngineError)

Enqueue a job with the worker’s defaults. Equivalent to insert_with(instance, worker, args, []).

pub fn insert_all(
  instance: Mule,
  new_jobs: List(job.NewJob),
) -> Result(List(job.Job), engine.EngineError)

Bulk-insert pre-encoded jobs in one engine round-trip (Oban.insert_all/2). Build each entry with worker.new_job and resolve the Results first, so validation surfaces before anything is inserted (the typed analog of Elixir raising InvalidChangesetError mid-batch): result.all the entries, then call this. Unique options on entries are IGNORED (basic.ex parity: bulk unique is a Pro Smart-engine feature — use insert for unique jobs). Publishes one InsertNotification per distinct queue that received an Available job (engine.ex notify_trigger, uniq: true); scheduled rows are left for the Stager.

pub fn insert_with(
  instance: Mule,
  worker: worker.Worker(a),
  args: a,
  options: List(job.Option),
) -> Result(job.Job, engine.EngineError)

Enqueue a job with Oban.Job.new/2-style options (queue, priority, tags, meta, max_attempts, scheduled_at / schedule_in, unique, replace). Inserts via the engine and, when the new job lands Available, publishes InsertNotification(queue) so the queue’s producer wakes and dispatches it.

Mirrors engine.ex’s notify_trigger, which runs on whatever row the engine returned: a job inserted as Scheduled is left for the Stager to promote (and notify) later, while a unique Conflict whose existing row is still Available publishes a wake-up just like a fresh insert — in both cases the existing/scheduled job is returned.

A job.Meta value must be a JSON object (uniqueness keys matching runs jsonb_each over it). An option that fails Job.new/2’s validations (priority 0–9, max_attempts > 0, queue length 1–128) surfaces as an EngineFailure — Elixir’s {:error, changeset}.

pub fn named(
  instance_name instance_name: String,
) -> Result(Mule, Nil)

The running instance’s handle, for holders of only the instance name — the Oban.Registry lookup behind Elixir’s name-based Oban.insert/2. Confirms at lookup time that the instance is up: Error(Nil) when nothing by this name is running.

pub fn pause_all_queues(instance: Mule) -> Nil

Pause every running queue (Oban.pause_all_queues/2) — the wildcard "*" queue target every producer matches.

pub fn pause_all_queues_with(
  instance: Mule,
  options: List(SignalOption),
) -> Nil
pub fn pause_queue(instance: Mule, queue: String) -> Nil

Stop a queue’s producers (cluster-wide) from dispatching new jobs. Running jobs continue (Oban.pause_queue/2).

pub fn pause_queue_with(
  instance: Mule,
  queue: String,
  options: List(SignalOption),
) -> Nil
pub fn resume_all_queues(instance: Mule) -> Nil

Resume every paused queue (Oban.resume_all_queues/2).

pub fn resume_all_queues_with(
  instance: Mule,
  options: List(SignalOption),
) -> Nil
pub fn resume_queue(instance: Mule, queue: String) -> Nil

Resume a paused queue’s producers (Oban.resume_queue/2).

pub fn resume_queue_with(
  instance: Mule,
  queue: String,
  options: List(SignalOption),
) -> Nil
pub fn retry_all_jobs(
  instance: Mule,
  query: job.Query,
) -> Result(Int, engine.EngineError)

Retry all matching jobs not currently available/executing (Oban.retry_all_jobs/2). Returns how many were retried. No wake-up is published (as in Elixir); the Stager’s global-mode check_available pass and the producers’ poll interval are the backstops.

pub fn retry_job(
  instance: Mule,
  job_id: Int,
) -> Result(Nil, engine.EngineError)

Make a job available again, adding an attempt if already maxed out (Oban.retry_job/2). available/executing jobs are ignored, and — unlike this port’s cancel_job — an unknown id is a silent Ok(Nil), mirroring Elixir’s :ok (retry never observes the resulting row, so there is no divergence to buy).

pub fn scale_queue(
  instance: Mule,
  queue: String,
  limit: Int,
) -> Result(Nil, engine.EngineError)

Change a queue’s concurrency limit (Oban.scale_queue/2). A limit below 1 is rejected before anything is published — Elixir’s validate_queue_opts! raises ArgumentError there; the typed analog is an EngineFailure (nothing validates on the receiving side, so an unchecked non-positive limit would silently stall every matching producer).

pub fn scale_queue_with(
  instance: Mule,
  queue: String,
  limit: Int,
  options: List(SignalOption),
) -> Result(Nil, engine.EngineError)
pub fn start(spec: MuleSpec) -> Result(Mule, actor.StartError)

Build the Config, assemble the supervision tree, and start it.

Tree (mirrors mule.ex init/1’s Notifier + Nursery + Peer + Sonar + Harbor shape below the root):

root (RestForOne)
├── registry  (owner actor; creates the per-instance ETS table)
├── notifier  (named worker)
├── [peer]    (optional; actor-backed peers only)
├── nursery (RestForOne)               — nursery.ex
│   ├── foreman: factory of queue supervisors
│   │   └── queue_supervisor[q] (OneForAll)   — queue/supervisor.ex
│   │       ├── foreman[q]  (job task factory)
│   │       ├── producer[q]
│   │       └── watchman[q] (drain-waits on shutdown)
│   └── midwife  (starts queues; handles start/stop signals)
└── runtime (OneForOne)                — the Harbor analog
     ├── sonar
     ├── stager
     └── plugin…

The engine’s backing resource (in-memory actor / Postgres pool) is not in this tree — the caller starts it in their own supervision tree, like Oban’s Repo. Deliberate divergence at the root: Elixir uses one_for_one plus per-consumer notifier monitors/resubscribe loops (midwife.ex:100-133); the port keeps RestForOne-with-notifier-first so a notifier crash restarts the subscribers instead. The registry sits FIRST so it survives notifier crashes; a registry crash restarts everything after it, which re-creates the table and re-registers every process. Termination runs in reverse start order, so the runtime and the queues drain before the notifier goes down — the drain-time Pkill path stays live through shutdown.

pub fn start_queue(
  instance: Mule,
  queue: String,
  limit: Int,
) -> Result(Nil, engine.EngineError)

Start a queue’s supervision tree at runtime, cluster-wide (Oban.start_queue/2). A node already running the queue ignores the signal. The queue starts unpaused; use start_queue_with for a paused start or scoping. A limit below 1 is rejected up front like scale_queue — the midwife’s signal path has no way to report a failed queue start back to the caller, so it must never be published.

pub fn start_queue_with(
  instance: Mule,
  queue: String,
  limit: Int,
  paused: Bool,
  options: List(SignalOption),
) -> Result(Nil, engine.EngineError)
pub fn stop_queue(instance: Mule, queue: String) -> Nil

Shut a queue’s supervision tree down, cluster-wide (Oban.stop_queue/2). The queue pauses first and running jobs get the shutdown grace period to finish (the watchman drain). A node not running the queue ignores the signal.

pub fn stop_queue_with(
  instance: Mule,
  queue: String,
  options: List(SignalOption),
) -> Nil
pub fn supervised(
  spec: MuleSpec,
) -> supervision.ChildSpecification(Mule)

The instance as a child of a supervision tree — Oban’s {Oban, opts} child spec. A supervisor-type child, so shutdown time is unlimited and the watchman drain gets the full shutdown_grace_period (a worker child’s 5s default would cut the drain short). The entire start path — already-running guard, migration check, overlay sweep, config assembly — runs inside the child start closure, so a parent-driven restart of the instance repeats all of it: a whole-instance restart is a fresh start (see the overlay comment in do_start for the two-level §9 contract). A very fast restart can transiently clash with the dying generation’s registry table and fail the first attempt; the parent’s restart tolerance should allow a retry.

pub fn update_job(
  instance: Mule,
  job_id: Int,
  updates: List(job.Update),
) -> Result(job.Job, engine.EngineError)

Update a job’s updatable fields — args, max_attempts, meta, priority, queue, scheduled_at, tags (Oban.update_job/3; job.ex @updatable_params minus worker, whose name the typed model pairs with an args codec). The row is re-read under FOR UPDATE SKIP LOCKED in a transaction, so a row another transaction currently holds — or an unknown id — is Error(JobLockedOrNotFound). Updating scheduled_at also moves the job to scheduled (job.ex normalize_state). As in Elixir, prefer this over hand-rolled updates, but an update racing the job’s own ack can still be overwritten by it.

pub fn whereis(
  instance_name instance_name: String,
) -> option.Option(process.Pid)

The root supervisor pid of a named running instance (Oban.whereis/1).

Search Document