mule - v0.1.0

mule/engine

Types

AffectedJob

</>

The slim row a bulk transition returns — exactly the {id, queue, state} partial select basic.ex makes. For cancel/retry/delete the select rides a subquery, so state is the row’s state WHEN MATCHED (before the transition) — how mule.cancel_all_jobs knows which cancelled rows were executing and need a Pkill broadcast. For rescue_jobs the select rides the update’s RETURNING, so state is the POST-rescue state instead.

pub type AffectedJob {
  AffectedJob(id: Int, queue: String, state: states.State)
}

Constructors

Engine

</> 
pub type Engine {
  Engine(
    all_jobs: fn(job.Query) -> Result(List(job.Job), EngineError),
    cancel_all_jobs: fn(job.Query) -> Result(
      List(AffectedJob),
      EngineError,
    ),
    cancel_job: fn(Int, option.Option(error.Error)) -> Result(
      job.Job,
      EngineError,
    ),
    check_available: fn() -> Result(List(String), EngineError),
    check_meta: fn(QueueMeta, List(Int)) -> QueueState,
    complete_job: fn(Int) -> Result(job.Job, EngineError),
    delete_all_jobs: fn(job.Query) -> Result(
      List(AffectedJob),
      EngineError,
    ),
    delete_job: fn(Int) -> Result(Nil, EngineError),
    discard_job: fn(Int, error.Error) -> Result(
      job.Job,
      EngineError,
    ),
    error_job: fn(Int, error.Error, duration.Duration) -> Result(
      job.Job,
      EngineError,
    ),
    fetch_jobs: fn(QueueMeta, List(Int)) -> Result(
      #(QueueMeta, List(job.Job)),
      EngineError,
    ),
    get_job: fn(Int) -> Result(job.Job, EngineError),
    init_queue: fn(QueueInit) -> Result(QueueMeta, EngineError),
    insert_all_jobs: fn(List(job.NewJob)) -> Result(
      List(job.Job),
      EngineError,
    ),
    insert_job: fn(job.NewJob) -> Result(
      InsertOutcome,
      EngineError,
    ),
    prune_jobs: fn(duration.Duration, Int) -> Result(
      Int,
      EngineError,
    ),
    put_meta: fn(QueueMeta, MetaUpdate) -> QueueMeta,
    refresh: fn(QueueMeta) -> QueueMeta,
    rescue_jobs: fn(duration.Duration) -> Result(
      List(AffectedJob),
      EngineError,
    ),
    reindex: fn(List(String), Int) -> Result(Nil, EngineError),
    retry_all_jobs: fn(job.Query) -> Result(
      List(AffectedJob),
      EngineError,
    ),
    retry_job: fn(Int) -> Result(Nil, EngineError),
    shutdown_meta: fn(QueueMeta) -> QueueMeta,
    snooze_job: fn(Int, duration.Duration) -> Result(
      job.Job,
      EngineError,
    ),
    stage_scheduled: fn(
      String,
      option.Option(timestamp.Timestamp),
    ) -> Result(Int, EngineError),
    stage_jobs: fn(Int) -> Result(List(String), EngineError),
    update_job: fn(Int, List(job.Update)) -> Result(
      job.Job,
      EngineError,
    ),
    verify_migrated: fn() -> Result(Nil, EngineError),
    elect_leader: fn(PeerLease, Bool) -> Result(Bool, EngineError),
    read_leader: fn(String) -> Result(
      option.Option(String),
      EngineError,
    ),
    delete_leader: fn(String, String) -> Result(Int, EngineError),
  )
}

Constructors

EngineError

</> 
pub type EngineError {
  JobNotFound(id: Int)
  JobLockedOrNotFound(id: Int)
  EngineFailure(message: String)
}

Constructors 

  • JobNotFound(id: Int)
  • JobLockedOrNotFound(id: Int)
  • EngineFailure(message: String)

InsertOutcome

</> 
pub type InsertOutcome {
  Inserted(job: job.Job)
  Conflict(existing: job.Job)
  ConflictLocked
}

Constructors

MetaUpdate

</>

The typed analog of put_meta’s (atom, term) pair — only the keys the pause/resume/scale signals mutate.

pub type MetaUpdate {
  SetLimit(limit: Int)
  SetPaused(paused: Bool)
}

Constructors 

  • SetLimit(limit: Int)
  • SetPaused(paused: Bool)

PeerLease

</>

One leadership-lease attempt, passed to elect_leader. name is the Mule instance name (the mule_peers conflict key), node this node’s identity, and interval_seconds the lease TTL — expires_at becomes now() + interval_seconds on the database clock.

pub type PeerLease {
  PeerLease(name: String, node: String, interval_seconds: Int)
}

Constructors 

  • PeerLease(name: String, node: String, interval_seconds: Int)

QueueInit

</>

The options init_queue receives — engine.ex init/2’s conf + opts, flattened into the typed record the producer builds at startup.

pub type QueueInit {
  QueueInit(
    name: String,
    node: String,
    queue: String,
    limit: Int,
    paused: Bool,
  )
}

Constructors 

  • QueueInit(
  name: String,
  node: String,
  queue: String,
  limit: Int,
  paused: Bool,
)

QueueMeta

</>

The queue metadata a producer threads through every engine call — the typed analog of the Basic engine’s meta map (basic.ex init, minus the per-producer uuid the port does not carry yet).

pub type QueueMeta {
  QueueMeta(
    name: String,
    node: String,
    queue: String,
    limit: Int,
    paused: Bool,
    refresh_interval: duration.Duration,
    started_at: timestamp.Timestamp,
    updated_at: timestamp.Timestamp,
    shutdown_started_at: option.Option(timestamp.Timestamp),
  )
}

Constructors

QueueState

</>

check_meta’s projection — Elixir’s Oban.queue_state(). running is the executing job ids on this node’s producer.

pub type QueueState {
  QueueState(
    queue: String,
    node: String,
    limit: Int,
    paused: Bool,
    running: List(Int),
    started_at: timestamp.Timestamp,
    updated_at: timestamp.Timestamp,
    shutdown_started_at: option.Option(timestamp.Timestamp),
  )
}

Constructors

Values

panic_engine

</> 
pub fn panic_engine(message message: String) -> Engine

An engine whose every operation panics with message — the canary the no-persistence executor paths (the inline engine, testing.perform_job) hand their ack: False executors: any engine call out of one is a broken contract, and the panic names it. No Elixir analog (behaviour dispatch needs no placeholder record) — this is port scaffolding, kept in one place so the record stays in lockstep with Engine.

✨ Search Document