grind/job

Builds the jobs grind.submit admits, and names the states a stored job moves through.

import gleam/time/duration
import grind
import grind/job

let request =
  job.new(receipt_worker(), order.id)
  |> job.with_id("receipt:" <> order.id)
  |> job.after(duration.minutes(5))
grind.submit(jobs, request)

A job built with new runs now, in its worker’s queue, with its worker’s attempt limit and a correlation Grind generates. Every submit records a receipt under the job’s id (with_id) or under an id Grind generates, so a submit whose reply was lost is reconcilable. Give a job your own id when the submit may be retried: resubmitting the same job under the same id returns the original admission instead of a second job.

State may gain variants; branch on is_finished or state_name where a new state should not break your code.

Types

CodecKind

</>

Which of a worker’s codecs a contract check concerns.

pub type CodecKind {
  InputCodec
  OutputCodec
  ErrorCodec
}

Constructors

```
InputCodec
```
```
OutputCodec
```
```
ErrorCodec
```

Job

</>

A job to submit. Build it with new and the with_*, at, after and unique setters.

pub type Job(input, output, error) =
  @internal Job(input, output, error)

JobHandle

</>

A typed reference to a stored job. It keeps the worker’s codecs, so grind.state, grind.outcome and grind.await return typed values. Store id(handle) to find the job again with grind.bind.

pub type JobHandle(input, output, error) =
  @internal JobHandle(input, output, error)

State

</>

The stored lifecycle states.

pub type State {
  Queued
  Scheduled
  Retryable
  Executing
  Succeeded
  BusinessFailed
  RuntimeFailed
  ContractMismatch
  Uncertain
  Discarded
  Cancelled
}

Constructors

```
Queued
```
Waiting to run.
```
Scheduled
```
Waiting for its scheduled time, or snoozed.
```
Retryable
```
Failed and waiting to retry.
```
Executing
```
Claimed by a consumer and running.
```
Succeeded
```
```
BusinessFailed
```
Failed terminally: attempts ran out, the retry policy declined, or the snooze limit was reached.
```
RuntimeFailed
```
The input could not be decoded, or the handler’s output or error was rejected by its codec or exceeded the payload limit.
```
ContractMismatch
```
The stored codec versions no longer match the registered worker.
```
Uncertain
```
The attempt’s effect is unknown; waits for an audited resolution.
```
Discarded
```
```
Cancelled
```

TerminalCause

</>

Why a failure is terminal.

pub type TerminalCause {
  BudgetExhausted
  RetryDeclined
  SnoozeLimitReached
}

Constructors

```
BudgetExhausted
```
The last attempt failed.
```
RetryDeclined
```
The worker’s retry policy declined to retry.
```
SnoozeLimitReached
```
The job snoozed more often than its worker allows.

Values

after

</>

pub fn after(
  job: Job(input, output, error),
  delay: duration.Duration,
) -> Job(input, output, error)

Runs the job delay after it is admitted, measured on the database’s clock. A resubmit under the same id converges on the first admission.

at

</>

pub fn at(
  job: Job(input, output, error),
  at: timestamp.Timestamp,
) -> Job(input, output, error)

Runs the job no earlier than at. A time in the past runs it now.

id

</>

pub fn id(handle: JobHandle(input, output, error)) -> Int

The stored job’s durable id.

is_finished

</>

pub fn is_finished(state: State) -> Bool

Whether the state is terminal. Uncertain is not: it waits for an audited resolution, which may replay the job.

new

</>

pub fn new(
  worker: worker.Worker(input, output, error),
  input: input,
) -> Job(input, output, error)

A job for worker with input, to run now in the worker’s queue.

queue

</>

pub fn queue(handle: JobHandle(input, output, error)) -> String

The queue the stored job runs in.

state_from_name

</>

pub fn state_from_name(name: String) -> Result(State, Nil)

The state with this stored name.

state_name

</>

pub fn state_name(state: State) -> String

The state’s stored name, such as "business_failed", for logs, metrics and records.

terminal_cause_name

</>

pub fn terminal_cause_name(cause: TerminalCause) -> String

The cause’s stored name, such as "snooze_limit_reached".

unique

</>

pub fn unique(
  job: Job(input, output, error),
  policy: unique.Policy(input),
) -> Job(input, output, error)

Admits the job only if no job occupies its uniqueness key.

with_correlation

</>

pub fn with_correlation(
  job: Job(input, output, error),
  correlation: correlation.Correlation,
) -> Job(input, output, error)

The correlation the job carries into its worker’s context and every event Grind emits about it. Without one, Grind generates one at submit.

with_id

</>

pub fn with_id(
  job: Job(input, output, error),
  id: String,
) -> Job(input, output, error)

The job’s idempotency key. A second submit with the same id and the same job returns the first admission; with a different job it fails with grind.IdConflict. An empty id fails at submit with grind.EmptyJobId. The key lasts as long as the job is stored.

with_max_attempts

</>

pub fn with_max_attempts(
  job: Job(input, output, error),
  max_attempts: Int,
) -> Job(input, output, error)

This one job’s attempt limit instead of its worker’s. Panics below 1.

with_queue

</>

pub fn with_queue(
  job: Job(input, output, error),
  queue: String,
) -> Job(input, output, error)

Submits this one job to queue instead of its worker’s queue. A consumer must run that queue with this worker registered. Panics on an empty queue.