howdy/jobs

Background jobs: work queued in the app’s database and done outside the request, tried again when it fails, and run on a schedule.

import howdy/jobs

// A job is defined once, in code, with how its arguments are stored.
pub fn welcome_email() -> jobs.Job(Int) {
  jobs.define("welcome_email", encode: json.int, decoder: decode.int)
  |> jobs.on_queue("mailers")
}

// It is queued inside the transaction that makes it necessary: if the
// transaction rolls back, the job was never queued.
use conn <- database.transaction(repo)
use user <- result.try(users.insert(conn, form))
use _ <- result.try(jobs.new(welcome_email(), user.id) |> jobs.insert(conn))
Ok(user)

// And done by a runner the app keeps under its supervisor.
jobs.runner(repo)
|> jobs.queue("mailers", concurrency: 5)
|> jobs.handle(welcome_email(), fn(user_id, _attempt) {
  case send_welcome(mailer, user_id) {
    Ok(_) -> jobs.Done
    Error(error) -> jobs.Retry(mail.error_to_string(error))
  }
})
|> jobs.supervised

Jobs are kept in the howdy_jobs table (apply schema with howdy/migration), on PostgreSQL or SQLite. Every node running a runner takes due jobs from the queues it has, so the work spreads across them, and a job is only ever taken by one node at a time.

At least once

A job is done at least once. While a node runs a job it keeps renewing a lease on it; if the node stops without finishing, the lease runs out and another node runs the job again. A job that sends an email, then crashes before saying it is done, sends it twice. Write jobs so that doing them twice is harmless, such as by checking first whether the work is done.

Arguments last

A job’s arguments are stored as JSON and decoded when it runs, which may be after a deploy has changed the decoder. Keep them small, such as IDs rather than whole records, and make decoders accept what older code stored. A job whose arguments cannot be decoded fails without retrying, and can be retried by hand once a fixed decoder is deployed.

Types

Attempt

</>

The attempt a handler is asked to make. Times are unix milliseconds.

pub type Attempt {
  Attempt(
    id: Int,
    name: String,
    queue: String,
    number: Int,
    max_attempts: Int,
    inserted_at: Int,
    scheduled_at: Int,
    errors: List(Failure),
  )
}

Constructors

Attempt(
  id: Int,
  name: String,
  queue: String,
  number: Int,
  max_attempts: Int,
  inserted_at: Int,
  scheduled_at: Int,
  errors: List(Failure),
)

Arguments

number: 1 for the first attempt.
errors: What went wrong in earlier attempts, oldest first.

Count

</>

How many jobs a queue has in a state.

pub type Count {
  Count(queue: String, state: State, count: Int)
}

Constructors

Count(queue: String, state: State, count: Int)

Drained

</>

What drain did.

pub type Drained {
  Drained(
    completed: Int,
    retried: Int,
    snoozed: Int,
    cancelled: Int,
    failed: Int,
  )
}

Constructors

Drained(
  completed: Int,
  retried: Int,
  snoozed: Int,
  cancelled: Int,
  failed: Int,
)

Failure

</>

Why an attempt did not succeed.

pub type Failure {
  Failure(attempt: Int, at: Int, reason: String)
}

Constructors

Failure(attempt: Int, at: Int, reason: String)

Insert

opaque </>

A job about to be queued. Build one with new.

pub opaque type Insert

Inserted

</>

What insert did.

pub type Inserted {
  Inserted(id: Int)
  Duplicate(id: Int)
}

Constructors

```
Inserted(id: Int)
```
```
Duplicate(id: Int)
```
A job holding the same unique key was waiting, or running, already.

Job

opaque </>

A kind of job, taking arguments of type args. Define each once, as a function, and use it both where the job is queued and where the runner is told how to do it.

pub opaque type Job(args)

Listing

opaque </>

Which stored jobs list returns. Build one with listing.

pub opaque type Listing

Outcome

</>

What became of an attempt at a job.

pub type Outcome {
  Done
  Retry(reason: String)
  RetryIn(milliseconds: Int, reason: String)
  Snooze(milliseconds: Int)
  Cancel(reason: String)
  GiveUp(reason: String)
}

Constructors

```
Done
```
```
Retry(reason: String)
```
It may work later, such as when a service it needs is down. It is tried again after the job’s backoff, unless this was its last attempt.

RetryIn(milliseconds: Int, reason: String)

Like Retry, after this many milliseconds.

```
Snooze(milliseconds: Int)
```
Not now, and it does not count as an attempt: try it again after this many milliseconds, such as when a rate limit says to wait.
```
Cancel(reason: String)
```
It is no longer needed. It is not tried again.
```
GiveUp(reason: String)
```
It never will work, such as when the record it is about is gone. It fails without being tried again.

Record

</>

A job as it is stored. Times are unix milliseconds; args is the JSON the job’s arguments were stored as.

pub type Record {
  Record(
    id: Int,
    name: String,
    queue: String,
    args: String,
    state: State,
    priority: Int,
    attempt: Int,
    max_attempts: Int,
    scheduled_at: Int,
    inserted_at: Int,
    attempted_at: option.Option(Int),
    finished_at: option.Option(Int),
    attempted_by: option.Option(String),
    unique_key: option.Option(String),
    errors: List(Failure),
  )
}

Constructors

Record(
  id: Int,
  name: String,
  queue: String,
  args: String,
  state: State,
  priority: Int,
  attempt: Int,
  max_attempts: Int,
  scheduled_at: Int,
  inserted_at: Int,
  attempted_at: option.Option(Int),
  finished_at: option.Option(Int),
  attempted_by: option.Option(String),
  unique_key: option.Option(String),
  errors: List(Failure),
)

Runner

opaque </>

What a node does jobs with: its queues, the handler for each kind of job, and the schedules it keeps. Build one with runner and run it with supervised. Every node can run the same runner.

pub opaque type Runner

Schedule

</>

A schedule a runner keeps.

pub type Schedule {
  Schedule(
    name: String,
    cron: String,
    queue: String,
    last: option.Option(Int),
    next: option.Option(Int),
  )
}

Constructors

Schedule(
  name: String,
  cron: String,
  queue: String,
  last: option.Option(Int),
  next: option.Option(Int),
)

Arguments

last: The time it last queued a job for, in unix milliseconds.
next: When it is next due, in unix milliseconds.

State

</>

Where a job is. Available and Scheduled jobs are both waiting; an available one is due now, a scheduled one later, including one waiting to be tried again.

pub type State {
  Available
  Scheduled
  Executing
  Completed
  Cancelled
  Failed
}

Constructors

```
Available
```
```
Scheduled
```
```
Executing
```
```
Completed
```
```
Cancelled
```
```
Failed
```

Values

after

</>

pub fn after(insert: Insert, milliseconds: Int) -> Insert

Do it no sooner than milliseconds from now.

at

</>

pub fn at(insert: Insert, unix_milliseconds: Int) -> Insert

Do it no sooner than this time, in unix milliseconds.

backoff

</>

pub fn backoff(
  job: Job(args),
  delay: fn(Int) -> Int,
) -> Job(args)

How long to wait before trying again after attempt number n failed, in milliseconds. The default doubles from 15 seconds up to six hours. A tenth either way is added at random, so jobs that failed together are not all tried again together.

before

</>

pub fn before(listing: Listing, id: Int) -> Listing

Only jobs older than the one with this id: the next page after it.

cancel

</>

pub fn cancel(
  repo: repo.Repo,
  id: Int,
) -> Result(Nil, service.Error)

Cancel a job that has not finished. A job that is running carries on, but what it returns is ignored.

check_every

</>

pub fn check_every(runner: Runner, milliseconds: Int) -> Runner

How often producers look for due jobs, in milliseconds, besides when a job is queued. Defaults to a second.

counts

</>

pub fn counts(
  repo: repo.Repo,
) -> Result(List(Count), service.Error)

How many jobs each queue has in each state, leaving out the states with none.

default_queue

</>

pub const default_queue: String

The queue jobs go to unless their definition says otherwise.

define

</>

pub fn define(
  name: String,
  encode encode: fn(args) -> json.Json,
  decoder decoder: decode.Decoder(args),
) -> Job(args)

A job called name, whose arguments are stored with encode and read back with decoder. The name is how stored jobs find their handler, so it must not change once jobs have been queued under it. Names and queue names are letters, digits, _, -, . and :, up to 100 characters.

By default a job goes to the default queue at priority 0, is tried up to 20 times with a growing delay between attempts, and is given five minutes for each attempt.

delete

</>

pub fn delete(
  repo: repo.Repo,
  id: Int,
) -> Result(Nil, service.Error)

Delete a job that is not running.

drain

</>

pub fn drain(runner: Runner) -> Result(Drained, service.Error)

Run every job that is due, in the calling process, one after another, until none are left, including any the jobs queue. For tests: no runner need be started, and the jobs’ work is done by the time it returns. Jobs that are tried again later are not waited for.

every

</>

pub fn every(
  runner: Runner,
  cron: cron.Cron,
  run job: Job(args),
  with args: args,
) -> Runner

Queue the job with these arguments whenever cron comes round. Each time is queued once, whichever nodes run the runner. A time missed while no node was running is queued once when one starts, however many were missed. A schedule new to the database starts from the next time.

get

</>

pub fn get(
  repo: repo.Repo,
  id: Int,
) -> Result(Record, service.Error)

handle

</>

pub fn handle(
  runner: Runner,
  job: Job(args),
  perform: fn(args, Attempt) -> Outcome,
) -> Runner

Do jobs of this kind with perform, which is given the job’s arguments and the attempt being made, and says what became of it. A crash is an attempt that failed and is tried again.

handled

</>

pub fn handled(runner: Runner) -> List(String)

The names of the jobs the runner has handlers for.

in_queue

</>

pub fn in_queue(listing: Listing, queue: String) -> Listing

in_state

</>

pub fn in_state(listing: Listing, state: State) -> Listing

insert

</>

pub fn insert(
  insert: Insert,
  into conn: repo.Repo,
) -> Result(Inserted, service.Error)

Queue the job on conn. Pass the connection of the transaction that makes the job necessary, and the job is only queued if that commits.

insert_all

</>

pub fn insert_all(
  inserts: List(Insert),
  into conn: repo.Repo,
) -> Result(List(Inserted), service.Error)

Queue several jobs, in order, on conn.

is_last

</>

pub fn is_last(attempt: Attempt) -> Bool

Whether this is the job’s last attempt, so a handler can do something different, such as tell someone, before it fails for good.

keep_finished_for

</>

pub fn keep_finished_for(
  runner: Runner,
  milliseconds: Int,
) -> Runner

How long finished jobs are kept, in milliseconds, before the runner deletes them. Defaults to seven days.

lease

</>

pub fn lease(runner: Runner, milliseconds: Int) -> Runner

How long a node’s claim on a running job lasts without being renewed, in milliseconds. A node renews it every third of this while the job runs, and when a node stops without finishing a job, others run it again after this long. Defaults to a minute.

limit

</>

pub fn limit(listing: Listing, limit: Int) -> Listing

list

</>

pub fn list(
  repo: repo.Repo,
  listing: Listing,
) -> Result(List(Record), service.Error)

listen_via

</>

pub fn listen_via(runner: Runner, config: pog.Config) -> Runner

On PostgreSQL, a runner listens for jobs queued by other nodes, using the settings of the Repo’s pool. Give the settings here when the pool was not opened through howdy/database/postgres or pog.

listing

</>

pub fn listing() -> Listing

Every job, newest first, 50 at a time.

max_attempts

</>

pub fn max_attempts(job: Job(args), attempts: Int) -> Job(args)

How many times the job is tried, counting the first, before it fails for good. At least 1.

name

</>

pub fn name(job: Job(args)) -> String

named

</>

pub fn named(listing: Listing, name: String) -> Listing

new

</>

pub fn new(job: Job(args), args: args) -> Insert

The job, to be done with these arguments as soon as a runner can.

on_queue

</>

pub fn on_queue(job: Job(args), queue: String) -> Job(args)

pause

</>

pub fn pause(
  repo: repo.Repo,
  queue: String,
) -> Result(Nil, service.Error)

Stop every node taking jobs from queue until it is resumed. Jobs already running finish.

paused

</>

pub fn paused(
  repo: repo.Repo,
) -> Result(List(String), service.Error)

The queues that are paused.

priority

</>

pub fn priority(job: Job(args), priority: Int) -> Job(args)

Which due jobs of a queue go first: 0, the default, before 1, and so on to 9. Jobs of the same priority go in the order they became due.

prune

</>

pub fn prune(
  repo: repo.Repo,
  older_than milliseconds: Int,
) -> Result(Int, service.Error)

Delete jobs that finished (completed, failed or were cancelled) more than older_than milliseconds ago, and say how many. Runners do this themselves, with keep_finished_for.

queue

</>

pub fn queue(
  runner: Runner,
  name: String,
  concurrency concurrency: Int,
) -> Runner

Take jobs from queue, running at most concurrency of them at once on this node. A queue a handler’s job uses is taken from anyway, ten at a time, unless it is given here.

queue_of

</>

pub fn queue_of(job: Job(args)) -> String

queues

</>

pub fn queues(runner: Runner) -> List(#(String, Int))

The queues the runner takes from, with their concurrency.

record_to_json

</>

pub fn record_to_json(record: Record) -> json.Json

The job as JSON, with its arguments as the JSON they were stored as.

repo

</>

pub fn repo(runner: Runner) -> repo.Repo

resume

</>

pub fn resume(
  repo: repo.Repo,
  queue: String,
) -> Result(Nil, service.Error)

retry

</>

pub fn retry(
  repo: repo.Repo,
  id: Int,
) -> Result(Nil, service.Error)

Run a waiting, failed or cancelled job as soon as possible, allowing it one more attempt if it has used them all. NotFound for a job that is running, has completed or does not exist.

runner

</>

pub fn runner(repo: repo.Repo) -> Runner

A runner for the jobs in repo, which must have had schema applied.

schedules

</>

pub fn schedules(
  runner: Runner,
) -> Result(List(Schedule), service.Error)

The runner’s schedules, with when each last ran and runs next.

schema

</>

pub fn schema() -> migration.Package

The migrations for the package’s tables, all named howdy_jobs.

shutdown_grace

</>

pub fn shutdown_grace(
  runner: Runner,
  milliseconds: Int,
) -> Runner

How long jobs that are running when the runner is stopped get to finish, in milliseconds. Those still running then are stopped and handed back, without counting the attempt. Defaults to 15 seconds.

start

</>

pub fn start(
  runner: Runner,
) -> Result(
  actor.Started(static_supervisor.Supervisor),
  actor.StartError,
)

Start the runner linked to the caller. Prefer supervised.

state_from_string

</>

pub fn state_from_string(text: String) -> Result(State, Nil)

state_to_string

</>

pub fn state_to_string(state: State) -> String

states

</>

pub const states: List(State)

stop

</>

pub fn stop(runner: Runner) -> Nil

Stop a runner started with start, giving running jobs its shutdown_grace to finish.

stored_names

</>

pub fn stored_names(
  repo: repo.Repo,
) -> Result(List(String), service.Error)

The names of all the jobs stored, of any state.

supervised

</>

pub fn supervised(
  runner: Runner,
) -> supervision.ChildSpecification(static_supervisor.Supervisor)

The runner as a child of the app’s supervisor, so it starts and stops with the app:

supervisor.new(supervisor.OneForOne)
|> supervisor.add(jobs.supervised(runner))
|> supervisor.add(howdy.supervised(app))

When it stops, running jobs get shutdown_grace to finish.

timeout

</>

pub fn timeout(job: Job(args), milliseconds: Int) -> Job(args)

How long an attempt may take, in milliseconds, before it is stopped and counted as failed.

unique

</>

pub fn unique(insert: Insert, key: String) -> Insert

Queue it only if no unfinished job has the same key, such as "reindex:" <> id. While one is waiting or running, insert gives Duplicate with that job’s id instead. Once it has finished, the key is free again.

unique_while_waiting

</>

pub fn unique_while_waiting(
  insert: Insert,
  key: String,
) -> Insert

Queue it only if no job with the same key is waiting to start; one that is running does not count. Work that arrives while a job runs then gets a job of its own, which a job holding the key the whole time would miss: with after, a burst of changes is gathered into one job, and a change made while it runs is not lost. insert gives Duplicate with the waiting job’s id when there is one. Keys of unique jobs and of these are separate.

A job like this that fails, snoozes or is handed back while another with its key waits is cancelled rather than waiting again: the other does the work, brought forward to when this one would have run again if it was due later.

without_args

</>

pub fn without_args(name: String) -> Job(Nil)

A job that takes no arguments, such as a nightly clean-up.

without_listening

</>

pub fn without_listening(runner: Runner) -> Runner

Do not listen on PostgreSQL: rely on check_every for jobs other nodes queue.