grind

Grind runs typed background jobs on PostgreSQL and Erlang/OTP.

import gleam/otp/static_supervisor as supervisor
import gleam/time/duration
import grind
import grind/job

pub fn children(pool: pog.Config, name: process.Name(grind.Message)) {
  let config =
    grind.new(pool)
    |> grind.with_worker(mailer())
    |> grind.with_startup_migration
  supervisor.new(supervisor.OneForOne)
  |> supervisor.add(grind.supervised(config, name))
}

pub fn send(name: process.Name(grind.Message), email: Email) {
  let jobs = grind.named(name)
  let assert Ok(admission) = grind.submit(jobs, job.new(mailer(), email))
  grind.await(jobs, grind.handle(admission), within: duration.seconds(5))
}

One child runs everything a node needs: the PostgreSQL pool built from the application’s pog.Config, an observation forwarder, one consumer per queue its workers use, and the pruner. grind.connection (and worker.connection inside a handler) returns that pool, so the application, Grind and other libraries share it; submit_in enqueues inside the application’s own transaction. A runtime that runs consumers starts only on a current schema: with_startup_migration applies the migrations first, or migrate at deploy time with migrate from a runtime without_consumers, or apply priv/migrations through cigogne.

Operation	Default	Setter
storage call	4 s; queued checkout may exceed it	`with_statement_deadline`
initial connect at start	15 s	`with_connect_timeout`
uniqueness lock wait	2 s	`with_unique_lock_wait`
migration step	30 s	`with_migration_deadline`
encoded input, output, error	1 MiB, then `PayloadTooLarge`	`with_max_payload_bytes`
job retention	pruner on, 7 days	`with_pruner`, `without_pruner`
observation forwarder	1,024 events in flight	`with_observation_capacity`
waiting for a result	`await(within:)` takes a bound

Worker and queue defaults are in grind/worker and grind/queue. Configuration is checked by start and supervised (check returns the same typed error up front). The errors and outcomes here may gain variants: branch on submit_error_kind, read_error_kind and cancel_error_kind where a new variant should not break your code.

Types

Admission

</>

What a submit admitted. A job without job.unique is always Inserted; handle returns the job either way, so a plain submit needs no case:

use admission <- result.try(grind.submit(jobs, job.new(mailer(), email)))
grind.await(jobs, grind.handle(admission), within: duration.seconds(5))

pub type Admission(input, output, error) {
  Inserted(job.JobHandle(input, output, error))
  Existing(Conflict(input, output, error))
  Rescheduled(Conflict(input, output, error))
}

Constructors

Inserted(job.JobHandle(input, output, error))

A new job.

```
Existing(Conflict(input, output, error))
```
A job already occupies the uniqueness key; nothing was inserted.
```
Rescheduled(Conflict(input, output, error))
```
The occupying scheduled job was moved to the policy’s time.

CancelError

</>

pub type CancelError {
  CancelJobNotFound
  CancelMismatch
  CancelCommitUnknown
  CancelUnavailable(pog.QueryError)
  CancelFromAnotherDatabase
  CancelNotRunning
}

Constructors

```
CancelJobNotFound
```
```
CancelMismatch
```
The stored job does not match the handle’s queue or worker.
```
CancelCommitUnknown
```
The cancellation may or may not have committed; cancelling again is safe.
```
CancelUnavailable(pog.QueryError)
```
```
CancelFromAnotherDatabase
```
```
CancelNotRunning
```

CancelResult

</>

What cancel did.

pub type CancelResult {
  CancelledBeforeRun
  CancellationRequested
  AlreadyCancelled
  AlreadyUncertain
  AlreadyFinished(job.State)
}

Constructors

```
CancelledBeforeRun
```
The job had not started; it is cancelled.
```
CancellationRequested
```
The job is running. Its handler’s worker.cancellation fires at the attempt’s next lease renewal, and its outcome becomes Cancelled unless it finishes first or reports Uncertain. Explicit uncertainty retains its evidence and cancellation intent.
```
AlreadyCancelled
```
```
AlreadyUncertain
```
The job remains uncertain. Cancellation intent is recorded and blocks replay; resolve it with an attributed terminal confirmation in grind/admin.
```
AlreadyFinished(job.State)
```

Config

opaque </>

A node’s Grind configuration. Build it with new and the with_* setters; start and supervised check it.

pub opaque type Config

ConfigError

</>

A configuration value start or supervised rejected.

pub type ConfigError {
  NotPositive(setting: String, value: Int)
  UniqueLockWaitTooCloseToDeadline(
    lock_wait_ms: Int,
    margin_ms: Int,
    deadline_ms: Int,
  )
  MigrationDeadlineTooCloseToLockTimeout(
    deadline_ms: Int,
    lock_timeout_ms: Int,
    margin_ms: Int,
  )
  InvalidSchema(schema: String)
  DuplicateWorker(id: String, version: String)
  QueueWithoutWorkers(queue: String)
  DuplicateQueue(queue: String)
  InvalidQueue(queue: String, setting: String, value: Int)
  LeaseTooShort(queue: String, lease_ms: Int, minimum_ms: Int)
}

Constructors

```
NotPositive(setting: String, value: Int)
```
A deadline, wait, timeout, limit, capacity or the pruner age is not positive. setting names it.

UniqueLockWaitTooCloseToDeadline(
  lock_wait_ms: Int,
  margin_ms: Int,
  deadline_ms: Int,
)

The uniqueness lock wait must end at least margin_ms before the statement deadline.

MigrationDeadlineTooCloseToLockTimeout(
  deadline_ms: Int,
  lock_timeout_ms: Int,
  margin_ms: Int,
)

The migration deadline must exceed the migration lock timeout by at least margin_ms.

```
InvalidSchema(schema: String)
```
The schema is empty, longer than 63 bytes, contains a NUL, is $user or starts with pg_.

DuplicateWorker(id: String, version: String)

Two registered workers share an id and version.

```
QueueWithoutWorkers(queue: String)
```
A configured queue has no registered worker.
```
DuplicateQueue(queue: String)
```
The same queue was configured twice.

InvalidQueue(queue: String, setting: String, value: Int)

A queue setting is out of range. setting names it.

LeaseTooShort(queue: String, lease_ms: Int, minimum_ms: Int)

The queue’s lease is shorter than four statement deadlines.

Conflict

</>

The stored job a uniqueness conflict found. state is its state when the decision was made. handle reads it with the submitted worker’s codecs: the occupying job has the same worker id and version, and a read still checks its stored codec versions.

pub type Conflict(input, output, error) {
  Conflict(
    job_id: Int,
    queue: String,
    state: job.State,
    handle: job.JobHandle(input, output, error),
  )
}

Constructors

Conflict(
  job_id: Int,
  queue: String,
  state: job.State,
  handle: job.JobHandle(input, output, error),
)

ErrorKind

</>

A stable classification of this module’s errors, for callers that should not match a union that may gain variants.

pub type ErrorKind {
  Invalid
  NotFound
  Mismatch
  Contended
  Unreachable
  Unknown
}

Constructors

```
Invalid
```
The request was wrong; retrying it unchanged fails the same way.
```
NotFound
```
```
Mismatch
```
The stored record does not match the request: another worker, queue, codec, id or database.
```
Contended
```
A lock was busy; retry.
```
Unreachable
```
The database or the runtime could not be reached; retry.
```
Unknown
```
The write may or may not have committed; reconcile, or retry under the same id.

Failure

</>

What kind of failure ended a job.

pub type Failure(error) {
  Business(error)
  BusinessUnrecorded
  RuntimeFailure
  ContractMismatch
}

Constructors

```
Business(error)
```
The handler’s typed error, stored by the worker’s error codec.
```
BusinessUnrecorded
```
A business failure with no stored error: the worker has no error codec, or the snooze limit ended the job.
```
RuntimeFailure
```
The input could not be decoded, or the output or error was rejected by its codec or exceeded the payload limit.
```
ContractMismatch
```
The stored codec versions no longer match the registered worker.

Grind

</>

A handle on a node’s Grind runtime, found by name. It stays valid before the runtime starts and across its restarts.

pub type Grind =
  @internal Grind

Message

</>

The runtime’s messages. Create the name start or supervised registers with process.new_name, once.

pub type Message =
  @internal Message

MigrateError

</>

Why migrate failed.

pub type MigrateError {
  MigrateNotRunning
  IncompatibleSchema
  UnsupportedSchemaVersion(version: Int)
  MigrationStepFailed(version: Int, reason: pog.QueryError)
  MigrationLockUnavailable(version: Int)
  MigrationCommitUnknown(version: Int)
  MigrationUnavailable(pog.QueryError)
}

Constructors

```
MigrateNotRunning
```
The runtime is not running.
```
IncompatibleSchema
```
The schema belongs to an unknown or newer Grind, or was altered.
```
UnsupportedSchemaVersion(version: Int)
```
The schema is at a version this Grind does not know.
```
MigrationStepFailed(version: Int, reason: pog.QueryError)
```
A migration step failed and rolled back; earlier steps stay applied.
```
MigrationLockUnavailable(version: Int)
```
A step’s lock wait expired; retry.
```
MigrationCommitUnknown(version: Int)
```
A step may or may not have committed; retrying is always safe.
```
MigrationUnavailable(pog.QueryError)
```
The database could not be reached, or the schema could not be created.

Outcome

</>

A job’s last committed result.

pub type Outcome(output, error) {
  Pending(state: job.State)
  Succeeded(output: output)
  Failed(
    failure: Failure(error),
    cause: option.Option(job.TerminalCause),
    description: String,
  )
  Discarded(reason: String)
  Cancelled(reason: String)
  Uncertain(evidence: String)
}

Constructors

```
Pending(state: job.State)
```
Not finished yet.
```
Succeeded(output: output)
```

Failed(
  failure: Failure(error),
  cause: option.Option(job.TerminalCause),
  description: String,
)

Failed terminally. cause says why a business failure is terminal.

```
Discarded(reason: String)
```
```
Cancelled(reason: String)
```
```
Uncertain(evidence: String)
```
The attempt’s effect is unknown; resolve it with admin.resolve_uncertain.

PendingSubmission

</>

A submit whose commit could not be confirmed. Pass it to reconcile_submission.

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

ReadError

</>

pub type ReadError {
  JobNotFound
  QueueMismatch(expected: String, actual: String)
  WorkerMismatch(
    expected_id: String,
    expected_version: String,
    actual_id: String,
    actual_version: String,
  )
  CodecMismatch(
    kind: job.CodecKind,
    expected: String,
    actual: String,
  )
  UndecodableValue(reason: String)
  CorruptRecord(value: String)
  HandleFromAnotherDatabase
  ReadUnavailable(pog.QueryError)
  ReadNotRunning
}

Constructors

```
JobNotFound
```
No stored job has this id.
```
QueueMismatch(expected: String, actual: String)
```
The stored job is in another queue than the handle says.

WorkerMismatch(
  expected_id: String,
  expected_version: String,
  actual_id: String,
  actual_version: String,
)

The stored job belongs to another worker id or version.

CodecMismatch(
  kind: job.CodecKind,
  expected: String,
  actual: String,
)

A stored codec version differs from the worker’s.

```
UndecodableValue(reason: String)
```
A stored value did not decode with its codec.
```
CorruptRecord(value: String)
```
A stored column holds a value Grind does not recognize.
```
HandleFromAnotherDatabase
```
The handle belongs to another database or schema.
```
ReadUnavailable(pog.QueryError)
```
```
ReadNotRunning
```
No runtime is running under this name on this node.

StartError

</>

Why start failed.

pub type StartError {
  InvalidConfig(ConfigError)
  Unavailable(pog.QueryError)
  SchemaNotMigrated(found: option.Option(Int), required: Int)
  StartupMigrationFailed(MigrateError)
  StartFailed(description: String)
}

Constructors

```
InvalidConfig(ConfigError)
```
```
Unavailable(pog.QueryError)
```
No connection could be made within the connect timeout, or the first query failed.
```
SchemaNotMigrated(found: option.Option(Int), required: Int)
```
The runtime would start consumers, but the schema is at found (None: not installed), below the required version. Configure with_startup_migration, or run migrate from a runtime started without_consumers (or apply priv/migrations) first.
```
StartupMigrationFailed(MigrateError)
```
with_startup_migration was set and the migration failed.
```
StartFailed(description: String)
```
The runtime’s processes could not start, for example because another runtime already uses this name or pool name. The description never includes an exit reason, which could carry the pool configuration.

StopError

</>

pub type StopError {
  NotStarted
  OwnedBySupervisor
  StopTimedOut
}

Constructors

```
NotStarted
```
No runtime is running under this name.
```
OwnedBySupervisor
```
The runtime belongs to an application supervisor; stop it there.
```
StopTimedOut
```
The runtime’s processes did not confirm their shutdown in time.

StopOutcome

</>

How a stop ended.

pub type StopOutcome {
  StoppedCleanly
  StoppedWithActiveWork(active_attempts: Int)
}

Constructors

```
StoppedCleanly
```
Every queue finished its running jobs within its grace.
```
StoppedWithActiveWork(active_attempts: Int)
```
These attempts were still running when their grace expired. Their claims are recovered when their leases expire.

SubmitError

</>

pub type SubmitError(input, output, error) {
  InvalidInput(reason: String)
  PayloadTooLarge(bytes: Int, limit: Int)
  EmptyJobId
  IdConflict
  UniquenessContended
  NotCommitted(reason: pog.QueryError)
  CommitUnknown(PendingSubmission(input, output, error))
  NotInTransaction
  TransactionIsolationUnsupported(isolation: String)
  WrongDatabase
  SubmitNotRunning
}

Constructors

```
InvalidInput(reason: String)
```
The input codec, or a uniqueness key’s codec, rejected the value. Nothing was written.
```
PayloadTooLarge(bytes: Int, limit: Int)
```
The encoded input is larger than the payload limit. Nothing was written.
```
EmptyJobId
```
job.with_id was given an empty id. Nothing was written.
```
IdConflict
```
The job’s id was already used for a different job.
```
UniquenessContended
```
The uniqueness lock wait expired. Nothing was written; retry.
```
NotCommitted(reason: pog.QueryError)
```
The admission did not commit. Retry, under the same id when it has one.
```
CommitUnknown(PendingSubmission(input, output, error))
```
The admission may or may not have committed. Retry the same job, or pass the pending submission to reconcile_submission.
```
NotInTransaction
```
submit_in was given a pool, not a transaction.

TransactionIsolationUnsupported(isolation: String)

submit_in’s transaction is not READ COMMITTED.

```
WrongDatabase
```
The transaction or pending submission belongs to another database or schema.
```
SubmitNotRunning
```
No runtime is running under this name on this node.

Values

arguments

</>

pub fn arguments(
  grind: Grind,
  handle: job.JobHandle(input, output, error),
) -> Result(input, ReadError)

The job’s typed input.

await

</>

pub fn await(
  grind: Grind,
  handle: job.JobHandle(input, output, error),
  within within: duration.Duration,
) -> Result(Outcome(output, error), ReadError)

Waits up to within for the job to leave the pending states, and returns its outcome then: the finished result, Uncertain, or Pending(state) when the time ran out. The budget is checked after each storage read, so the last read can finish after within.

bind

</>

pub fn bind(
  grind: Grind,
  worker: worker.Worker(input, output, error),
  id: Int,
) -> Result(job.JobHandle(input, output, error), ReadError)

Rebinds a stored job by its id to worker, checking that the stored worker and codec versions match. Bind against the same database and schema the id came from.

cancel

</>

pub fn cancel(
  grind: Grind,
  handle: job.JobHandle(input, output, error),
) -> Result(CancelResult, CancelError)

Cancels a job. A job that has not started is cancelled at once; a running job’s handler is told through worker.cancellation. An uncertain job retains its evidence and records intent, including on repeated calls. Cancellation does not prove that an external effect was prevented.

cancel_error_kind

</>

pub fn cancel_error_kind(error: CancelError) -> ErrorKind

check

</>

pub fn check(config: Config) -> Result(Nil, ConfigError)

Checks a configuration without starting anything. start and supervised run the same checks.

connection

</>

pub fn connection(grind: Grind) -> pog.Connection

The runtime’s pool, shared with the application: run your own queries on it, or open the transaction that submit_in enqueues inside. It is pog.named_connection of the configuration’s pool name, and its search_path is the application’s: Grind sets its own schema for its own statements only. A handler reaches the same pool with worker.connection(context). Panics when no runtime was configured under this name on this node with start or supervised.

describe_cancel_error

</>

pub fn describe_cancel_error(error: CancelError) -> String

describe_config_error

</>

pub fn describe_config_error(error: ConfigError) -> String

describe_read_error

</>

pub fn describe_read_error(error: ReadError) -> String

describe_start_error

</>

pub fn describe_start_error(error: StartError) -> String

describe_submit_error

</>

pub fn describe_submit_error(
  error: SubmitError(input, output, error),
) -> String

handle

</>

pub fn handle(
  admission: Admission(input, output, error),
) -> job.JobHandle(input, output, error)

The admitted job: the new one, or the one occupying the uniqueness key.

migrate

</>

pub fn migrate(grind: Grind) -> Result(Nil, MigrateError)

Applies every missing schema migration, each in its own transaction under an advisory lock. Running it again, or on many nodes at once, is safe. A runtime that runs consumers needs a current schema to start, so call this from a runtime started without_consumers (a deploy step), or configure with_startup_migration instead.

named

</>

pub fn named(name: process.Name(Message)) -> Grind

The handle for the runtime registered under name.

new

</>

pub fn new(pool: pog.Config) -> Config

A configuration that builds Grind’s pool from the application’s pog.Config, keeping its pool name, size and search_path. Grind adds two connection parameters: READ COMMITTED isolation, and an idle-in-transaction timeout of twice the statement deadline. Grind’s own statements run under its schema (with_schema), set for each storage call and restored before the connection returns to the pool, so the application’s unqualified tables keep resolving through its own search_path. The configuration is held in a closure, so its password does not print.

outcome

</>

pub fn outcome(
  grind: Grind,
  handle: job.JobHandle(input, output, error),
) -> Result(Outcome(output, error), ReadError)

The job’s last committed result.

read_error_kind

</>

pub fn read_error_kind(error: ReadError) -> ErrorKind

reconcile_submission

</>

pub fn reconcile_submission(
  grind: Grind,
  pending: PendingSubmission(input, output, error),
) -> Result(
  Admission(input, output, error),
  SubmitError(input, output, error),
)

Settles a CommitUnknown by reading the receipt the submission would have written. A matching receipt returns the original admission; no receipt yet is another CommitUnknown.

start

</>

pub fn start(
  config: Config,
  name: process.Name(Message),
) -> Result(Grind, StartError)

Starts a runtime that no supervisor owns, registered under name. Use supervised in an application; start suits scripts and tests. Stop it with stop.

state

</>

pub fn state(
  grind: Grind,
  handle: job.JobHandle(input, output, error),
) -> Result(job.State, ReadError)

The job’s stored state.

stop

</>

pub fn stop(grind: Grind) -> Result(StopOutcome, StopError)

Stops a runtime started with start, from any process: each queue stops claiming and waits up to its grace for running jobs, then every process stops, the pool last.

submit

</>

pub fn submit(
  grind: Grind,
  job: job.Job(input, output, error),
) -> Result(
  Admission(input, output, error),
  SubmitError(input, output, error),
)

Admits a job in its own transaction. Every admission records a receipt, under the caller’s submission key or a generated key. Retain the pending command after CommitUnknown; reconciliation depends on receipt retention. Emits [grind, job, admitted] once the commit is proven.

submit_error_kind

</>

pub fn submit_error_kind(
  error: SubmitError(input, output, error),
) -> ErrorKind

submit_in

</>

pub fn submit_in(
  grind: Grind,
  tx: pog.Connection,
  job: job.Job(input, output, error),
) -> Result(
  Admission(input, output, error),
  SubmitError(input, output, error),
)

Admits a job inside the application’s open transaction tx, so it commits or rolls back with the application’s own writes. Grind sends no BEGIN or COMMIT. The transaction must be READ COMMITTED and on Grind’s database; Grind sets search_path and lock_timeout for its own statements, then restores yours. SERIALIZABLE and REPEATABLE READ are refused; retain the application isolation and use an application outbox when needed. A lost outer commit reply requires business-command reconciliation: an older matching admission receipt cannot prove this transaction’s business writes committed. Retry job admission under the same id (job.with_id) only as part of that application recovery protocol.

pog.transaction(grind.connection(jobs), fn(tx) {
  use _ <- result.try(orders.confirm(tx, order))
  grind.submit_in(jobs, tx, job.new(receipt(), order.id))
  |> result.map_error(ReceiptNotQueued)
})

Grind emits no [grind, job, admitted] event here, because it cannot see whether your commit succeeded; the job’s first event is claimed. To trace the admission, record it yourself once pog.transaction returns Ok: the admission’s job.id(grind.handle(admission)), and the correlation you gave the job with job.with_correlation, join it to the job’s later events.

supervised

</>

pub fn supervised(
  config: Config,
  name: process.Name(Message),
) -> supervision.ChildSpecification(Grind)

The runtime as one child of the application’s supervision tree, registered under name. A configuration error, an unreachable database, or a schema behind this Grind’s when the node runs consumers (see with_startup_migration) fails the child’s start with a description; call check first for a typed configuration error. On shutdown each queue stops claiming and waits up to its grace for running jobs.

with_connect_timeout

</>

pub fn with_connect_timeout(
  config: Config,
  timeout: duration.Duration,
) -> Config

How long start waits for a first connection before it fails with Unavailable.

with_max_payload_bytes

</>

pub fn with_max_payload_bytes(
  config: Config,
  bytes: Int,
) -> Config

The largest encoded input, output or error, in bytes. A larger input fails the submit with PayloadTooLarge; a larger output or error ends the job as a runtime failure.

with_migration_deadline

</>

pub fn with_migration_deadline(
  config: Config,
  deadline: duration.Duration,
) -> Config

Bounds each migration step.

with_observation_capacity

</>

pub fn with_observation_capacity(
  config: Config,
  capacity: Int,
) -> Config

How many events the observation forwarder holds in flight before it drops and counts them.

with_pruner

</>

pub fn with_pruner(
  config: Config,
  max_age max_age: duration.Duration,
) -> Config

Deletes finished jobs, with their receipts, once they are older than max_age: every 30 seconds, up to 10,000 at a time. It is on by default with a max_age of 7 days, and safe to run on every node at once.

with_queue

</>

pub fn with_queue(config: Config, queue: queue.Queue) -> Config

Tunes one queue. The queue must be one a registered worker uses.

with_schema

</>

pub fn with_schema(config: Config, schema: String) -> Config

The PostgreSQL schema Grind reads and writes, public by default. The schema is the unit of isolation between Grind installations in one database. It scopes only Grind’s own statements: queries the application runs on connection keep the application’s search_path.

with_startup_migration

</>

pub fn with_startup_migration(config: Config) -> Config

Applies missing schema migrations when the runtime starts, before any consumer polls, exactly as migrate does; a failure fails the start with StartupMigrationFailed. Without it, a runtime that would start consumers refuses a schema behind this Grind’s with SchemaNotMigrated. Leave it off when the runtime’s role may not run DDL, and migrate at deploy time instead.

with_statement_deadline

</>

pub fn with_statement_deadline(
  config: Config,
  deadline: duration.Duration,
) -> Config

Sets the absolute storage deadline before checkout. A connection obtained after expiry sends nothing; the pinned pool can wait past it while queued. See docs/adr/0008-state-driver-deadline-and-installation-limits.md.

with_unique_lock_wait

</>

pub fn with_unique_lock_wait(
  config: Config,
  wait: duration.Duration,
) -> Config

Bounds the wait for a uniqueness admission’s lock; it must end at least one second before the statement deadline.

with_worker

</>

pub fn with_worker(
  config: Config,
  worker: worker.Worker(input, output, error),
) -> Config

Registers a worker. Its queue runs on this node unless consumers are off; a submit-only node registers no workers.

without_consumers

</>

pub fn without_consumers(config: Config) -> Config

Runs no consumers on this node: it submits and reads jobs only, like Oban’s queues: false.

without_pruner

</>

pub fn without_pruner(config: Config) -> Config

Keeps finished jobs forever.