grind/worker

Defines typed workers: what a job’s input, output and error are, how they persist as JSON, which queue runs them, and how failures, snoozes, timeouts and abandoned attempts are handled.

import gleam/dynamic/decode
import gleam/json
import grind/worker

pub fn mailer() -> worker.Worker(Email, String, Nil) {
  worker.new(
    "mailer.send",
    input: worker.codec(worker.infallible(encode_email), email_decoder()),
    output: worker.codec(worker.infallible(json.string), decode.string),
    perform: fn(email) { Ok("msg:" <> email.to) },
  )
  |> worker.with_queue("mailers")
}

A worker is a definition written in source code, so its constructors and setters are total: an empty id, version or queue, or an out-of-range limit, is a programming error and panics with the worker’s id. The worker id and version, and each codec version, are stored with every job, so a consumer runs a job only with the exact definition it was submitted under. Versions default to "1"; change one with with_version or with_codec_version when a stored shape changes.

A handler built with new returns Result(output, error). One built with responding receives the job’s Context (job id, attempt, snooze count, correlation, cancellation, deadline and the runtime’s database connection) and returns a Response that may also snooze, discard, cancel or report an uncertain effect.

Policy	Default	Setter
queue	`"default"`	`with_queue`
version	`"1"`	`with_version`
attempts	20	`with_max_attempts`
retry delay	15 s doubling to 1 day, plus 0–10% jitter	`with_retry_policy`
handler timeout	15 minutes	`with_timeout(Infinity)`
snoozes per job	100, then `SnoozeLimitReached`	`with_max_snoozes`
abandoned attempts	`HoldUncertain`	`with_abandonment`

Types

Abandonment

</>

What happens to an attempt whose node died, whose handler exceeded its timeout, or whose lease otherwise expired before an acknowledgement.

pub type Abandonment {
  HoldUncertain
  ReplayAfterLeaseExpiry(max_replays: Int)
}

Constructors

```
HoldUncertain
```
Hold the job uncertain until an operator resolves it. The default: a handler that may have acted is never run twice without a decision.
```
ReplayAfterLeaseExpiry(max_replays: Int)
```
Requeue the job after its lease expires, at most max_replays times, then hold it uncertain. Choose it for handlers that are idempotent by construction. A replay redelivers the same business attempt: it does not count against with_max_attempts, and attempt reads the same number again. [grind, job, quarantined] reports it with replayed: True.

Codec

</>

A versioned JSON boundary for one value type. Build it with codec.

pub type Codec(value) =
  @internal Codec(value)

Context

</>

What a running handler knows about its job. Read it with job_id, attempt, max_attempts, snooze_count, queue, correlation, cancellation, deadline and connection. grind/testing.context builds one for a test.

pub type Context =
  @internal Context

Response

</>

What a responding handler proposes. Grind commits it through the attempt’s fenced acknowledgement; the job’s committed outcome is read back with grind.outcome or grind.await.

pub type Response(output, error) {
  Succeeded(output)
  Failed(error)
  Snoozed(after: duration.Duration, reason: String)
  Discarded(reason: String)
  Cancelled(reason: String)
  Uncertain(evidence: String)
}

Constructors

```
Succeeded(output)
```
The job succeeded with this output.
```
Failed(error)
```
The job failed with this error. Grind retries it while attempts and the retry policy allow, then ends it Failed.
```
Snoozed(after: duration.Duration, reason: String)
```
Run the job again after after, without using an attempt. The 101st snooze (by default) ends the job with cause SnoozeLimitReached.
```
Discarded(reason: String)
```
End the job as discarded, without retrying.
```
Cancelled(reason: String)
```
End the job as cancelled.
```
Uncertain(evidence: String)
```
The handler cannot tell whether its effect happened. The job is held uncertain until an operator resolves it with grind/admin.

RetryDecision

</>

A retry policy’s decision for one failed attempt.

pub type RetryDecision {
  RetryAfter(duration.Duration)
  DoNotRetry
}

Constructors

```
RetryAfter(duration.Duration)
```
```
DoNotRetry
```

Timeout

</>

How long a handler may run before Grind stops it.

pub type Timeout {
  After(duration.Duration)
  Infinity
}

Constructors

```
After(duration.Duration)
```
```
Infinity
```
No limit: the attempt’s lease is renewed for as long as it runs.

Worker

</>

A typed worker definition. Build it with new or responding.

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

Values

attempt

</>

pub fn attempt(context: Context) -> Int

The current business attempt, from 1.

cancellation

</>

pub fn cancellation(context: Context) -> process.Selector(Nil)

A selector that yields once a caller’s grind.cancel of this job has committed. Grind sees the cancellation at the attempt’s next lease renewal, within a third of the queue’s lease. A handler that stops early returns Cancelled.

codec

</>

pub fn codec(
  encode: fn(value) -> Result(json.Json, String),
  decoder: decode.Decoder(value),
) -> Codec(value)

A codec with version "1". encode may reject a value with a reason: grind.submit then returns InvalidInput and writes nothing, and a rejected output or error ends the job as a runtime failure. Wrap an encoder that cannot fail with infallible.

worker.codec(worker.infallible(json.string), decode.string)

connection

</>

pub fn connection(context: Context) -> pog.Connection

The runtime’s pool, the one grind.connection returns, for the handler’s own queries. A worker is defined before the runtime starts, so this is how its handler reaches the shared pool. Queries on it run under the application’s search_path. It is not the attempt’s acknowledgement: a write commits on its own, so a handler that may run twice writes idempotently.

correlation

</>

pub fn correlation(context: Context) -> correlation.Correlation

The job’s correlation: the one given with job.with_correlation, or one Grind generated at submit. Pass it on to the packages the handler calls.

deadline

</>

pub fn deadline(
  context: Context,
) -> option.Option(timestamp.Timestamp)

When the handler will be stopped, if the worker has a timeout.

default_backoff

</>

pub fn default_backoff(attempt: Int) -> duration.Duration

The default delay before retrying after attempt failed, before jitter: 15 seconds after the first attempt, doubling, capped at one day.

from_result

</>

pub fn from_result(
  result: Result(output, error),
) -> Response(output, error)

Converts a perform-style result to a Response.

id

</>

pub fn id(worker: Worker(input, output, error)) -> String

The worker’s id.

infallible

</>

pub fn infallible(
  encode: fn(value) -> json.Json,
) -> fn(value) -> Result(json.Json, String)

Adapts an encoder that cannot fail, such as json.string.

job_id

</>

pub fn job_id(context: Context) -> Int

The job’s id, the durable integer job.id returns at submit.

max_attempts

</>

pub fn max_attempts(context: Context) -> Int

The job’s attempt limit.

new

</>

pub fn new(
  id: String,
  input input: Codec(input),
  output output: Codec(output),
  perform perform: fn(input) -> Result(output, error),
) -> Worker(input, output, error)

A worker whose handler returns Result(output, error). An Error is retried with the default backoff until the attempts run out. Panics on an empty id.

queue

</>

pub fn queue(context: Context) -> String

The queue the job runs in.

responding

</>

pub fn responding(
  id: String,
  input input: Codec(input),
  output output: Codec(output),
  handle handle: fn(Context, input) -> Response(output, error),
) -> Worker(input, output, error)

A worker whose handler receives the job’s Context and returns a Response. Panics on an empty id.

snooze_count

</>

pub fn snooze_count(context: Context) -> Int

How many times this job has snoozed so far.

version

</>

pub fn version(worker: Worker(input, output, error)) -> String

The worker’s version.

with_abandonment

</>

pub fn with_abandonment(
  worker: Worker(input, output, error),
  abandonment: Abandonment,
) -> Worker(input, output, error)

What happens to an abandoned attempt; HoldUncertain by default. Panics on ReplayAfterLeaseExpiry with fewer than 1 replay.

with_codec_version

</>

pub fn with_codec_version(
  codec: Codec(value),
  version: String,
) -> Codec(value)

Sets the codec’s stored version. Change it when the encoded shape changes, so jobs stored under the old shape are not decoded with the new decoder. Panics on an empty version.

with_error_codec

</>

pub fn with_error_codec(
  worker: Worker(input, output, error),
  codec: Codec(error),
) -> Worker(input, output, error)

Stores the handler’s typed error, so grind.outcome returns Failed(Business(error), ..). Without it, a failure is stored as its description only (BusinessUnrecorded).

with_max_attempts

</>

pub fn with_max_attempts(
  worker: Worker(input, output, error),
  max_attempts: Int,
) -> Worker(input, output, error)

The number of business attempts, 20 by default. A snooze or a replayed lease does not use one. Panics below 1.

with_max_snoozes

</>

pub fn with_max_snoozes(
  worker: Worker(input, output, error),
  max_snoozes: Int,
) -> Worker(input, output, error)

How many times one job may snooze, 100 by default. The next snooze ends the job as failed with cause SnoozeLimitReached. Panics below 0.

with_queue

</>

pub fn with_queue(
  worker: Worker(input, output, error),
  queue: String,
) -> Worker(input, output, error)

The queue this worker’s jobs are submitted to and claimed from. The default is "default". grind.with_worker derives the queues a node runs from its workers. Panics on an empty queue.

with_retry_policy

</>

pub fn with_retry_policy(
  worker: Worker(input, output, error),
  policy: fn(error, Int) -> RetryDecision,
) -> Worker(input, output, error)

Chooses whether and when a failed attempt is retried. The callback gets the typed error and the attempt that just failed (from 1). It is not consulted after the last attempt. The default retries after default_backoff(attempt) plus up to 10% jitter.

with_timeout

</>

pub fn with_timeout(
  worker: Worker(input, output, error),
  timeout: Timeout,
) -> Worker(input, output, error)

How long the handler may run, 15 minutes by default. When it expires the handler is stopped and with_abandonment decides what happens next. Infinity lifts the limit. Panics on a duration that is not positive.

with_version

</>

pub fn with_version(
  worker: Worker(input, output, error),
  version: String,
) -> Worker(input, output, error)

The worker’s stored version, "1" by default. A consumer runs only jobs submitted under the exact id and version it registers. Panics on an empty version.