grind/admin

Operator work: listing and counting jobs, resolving uncertainty, sweeping expired attempts, pruning, and reading acknowledgement receipts.

import grind/admin
import grind/job

// Find the jobs that wait for an operator.
let assert Ok(jobs) =
  admin.list(grind, admin.query(limit: 100) |> admin.in_state(job.Uncertain))

An uncertain job’s attempt may or may not have had its effect: its node died, its handler exceeded its timeout, or it reported Uncertain. Grind never runs it again on its own unless its worker opted into worker.ReplayAfterLeaseExpiry. An operator resolves it with resolve_uncertain: confirm the success or failure that happened, or authorize a replay. Every resolution is recorded with its author and is idempotent under its id.

Types

AcknowledgementReceipt

</>

The durable record of one acknowledged attempt.

pub type AcknowledgementReceipt {
  AcknowledgementReceipt(
    command_id: String,
    attempt_id: Int,
    attempt_epoch: Int,
    committed_state: job.State,
    cause: option.Option(job.TerminalCause),
    committed_at: timestamp.Timestamp,
  )
}

Constructors

AcknowledgementReceipt(
  command_id: String,
  attempt_id: Int,
  attempt_epoch: Int,
  committed_state: job.State,
  cause: option.Option(job.TerminalCause),
  committed_at: timestamp.Timestamp,
)

Decision

</>

An operator’s decision about an uncertain job.

pub type Decision(output, error) {
  ConfirmSuccess(output)
  ConfirmFailure(error)
  AuthorizeReplay
}

Constructors

```
ConfirmSuccess(output)
```
The attempt succeeded with this output.
```
ConfirmFailure(error)
```
The attempt failed with this error. Needs an error codec.
```
AuthorizeReplay
```
Run the job again.

Error

</>

pub type Error {
  InvalidLimit(limit: Int, maximum: Int)
  InvalidRetention(milliseconds: Int)
  EmptyResolutionField(field: String)
  NotUncertain
  ResolutionConflict
  RecordMismatch
  ResolutionNeedsErrorCodec
  ResolutionValueRejected(reason: String)
  CancellationPending
  ResolutionCommitUnknown(resolution_id: String)
  NotInTransaction
  TransactionIsolationUnsupported(String)
  JobNotFound
  ReceiptNotFound
  WrongDatabase
  Unavailable(pog.QueryError)
  NotRunning
}

Constructors

```
InvalidLimit(limit: Int, maximum: Int)
```
A limit is below 1 or above maximum.
```
InvalidRetention(milliseconds: Int)
```
A retention age is not positive.
```
EmptyResolutionField(field: String)
```
A resolution’s field is empty.
```
NotUncertain
```
The job is not uncertain, so there is nothing to resolve.
```
ResolutionConflict
```
The resolution id was already used for another job or a different command in this installation (the same PostgreSQL database and Grind schema).
```
RecordMismatch
```
The stored job does not match the handle’s queue, worker or codecs.
```
ResolutionNeedsErrorCodec
```
A ConfirmFailure needs a worker with an error codec.
```
ResolutionValueRejected(reason: String)
```
The confirmed value was rejected by its codec.
```
CancellationPending
```
Cancellation intent forbids replay. Investigate the effect and use an attributed terminal confirmation to settle the job.
```
ResolutionCommitUnknown(resolution_id: String)
```
The resolution may or may not have committed; repeat it with the same id.
```
NotInTransaction
```
Borrowed resolution requires the connection passed to a pog.transaction callback.
```
TransactionIsolationUnsupported(String)
```
Borrowed resolution requires READ COMMITTED; isolation is never changed.
```
JobNotFound
```
```
ReceiptNotFound
```
```
WrongDatabase
```
The handle belongs to another installation, or the transaction uses another database.
```
Unavailable(pog.QueryError)
```
```
NotRunning
```
No runtime is running under this name on this node.

JobSummary

</>

One stored job, without its typed payloads.

pub type JobSummary {
  JobSummary(
    id: Int,
    queue: String,
    worker_id: String,
    worker_version: String,
    state: job.State,
    attempt: Int,
    max_attempts: Int,
    snooze_count: Int,
    replay_count: Int,
    inserted_at: timestamp.Timestamp,
    available_at: timestamp.Timestamp,
    finished_at: option.Option(timestamp.Timestamp),
    description: option.Option(String),
    correlation: option.Option(correlation.Correlation),
  )
}

Constructors

JobSummary(
  id: Int,
  queue: String,
  worker_id: String,
  worker_version: String,
  state: job.State,
  attempt: Int,
  max_attempts: Int,
  snooze_count: Int,
  replay_count: Int,
  inserted_at: timestamp.Timestamp,
  available_at: timestamp.Timestamp,
  finished_at: option.Option(timestamp.Timestamp),
  description: option.Option(String),
  correlation: option.Option(correlation.Correlation),
)

Query

opaque </>

A filter for list. Build it with query and the in_* and after setters.

pub opaque type Query

Resolution

opaque </>

An attributed decision. Build it with resolution.

pub opaque type Resolution(output, error)

Resolved

</>

What a resolution did.

pub type Resolved {
  Applied(job.State)
  AlreadyApplied(job.State)
}

Constructors

```
Applied(job.State)
```
The decision was applied; the job is now in this state.
```
AlreadyApplied(job.State)
```
The same decision was already applied under this id.

StagedResolution

</>

The result of resolution statements inside a caller-owned transaction. Even AlreadyApplied does not prove that the surrounding application writes committed.

pub type StagedResolution {
  Staged(Resolved)
}

Constructors

```
Staged(Resolved)
```

StateStatistics

</>

Ages are elapsed milliseconds, clamped at zero. An absent age means the corresponding count is zero. Insertion age is not time in the current state. Due counts only waiting jobs whose availability has arrived.

pub type StateStatistics {
  StateStatistics(
    state: job.State,
    count: Int,
    oldest_job_age_ms: option.Option(Int),
    due_count: Int,
    oldest_due_age_ms: option.Option(Int),
  )
}

Constructors

StateStatistics(
  state: job.State,
  count: Int,
  oldest_job_age_ms: option.Option(Int),
  due_count: Int,
  oldest_due_age_ms: option.Option(Int),
)

Statistics

</>

Committed counts for one queue at one database snapshot. State order is unspecified; all known states are present, including empty groups.

pub type Statistics {
  Statistics(sampled_at_ms: Int, states: List(StateStatistics))
}

Constructors

Statistics(sampled_at_ms: Int, states: List(StateStatistics))

Values

after

</>

pub fn after(query: Query, id: Int) -> Query

Only jobs with an id above id: pass the last id of one page to read the next.

describe_error

</>

pub fn describe_error(error: Error) -> String

error_kind

</>

pub fn error_kind(error: Error) -> grind.ErrorKind

in_queue

</>

pub fn in_queue(query: Query, queue: String) -> Query

Only jobs in queue.

in_state

</>

pub fn in_state(query: Query, state: job.State) -> Query

Only jobs in state.

list

</>

pub fn list(
  grind: grind.Grind,
  query: Query,
) -> Result(List(JobSummary), Error)

Lists stored jobs that match query. Uncertain jobs are served by an index; other states are read in id order.

prune_finished

</>

pub fn prune_finished(
  grind: grind.Grind,
  older_than older_than: duration.Duration,
  limit limit: Int,
) -> Result(Int, Error)

Deletes up to limit (at most 10,000) jobs that finished more than older_than ago, with their receipts, and returns how many. The configured pruner does this on a timer.

quarantine_expired

</>

pub fn quarantine_expired(
  grind: grind.Grind,
  limit limit: Int,
) -> Result(Int, Error)

Moves up to limit (at most 10,000) expired executing attempts, in every queue, to their abandonment outcome: uncertain, or queued again for a worker that replays. Consumers already do this for the queues they poll; run it for queues no node polls. Returns how many rows moved.

query

</>

pub fn query(limit limit: Int) -> Query

Up to limit jobs (at most 10,000), oldest id first.

reconcile_acknowledgement

</>

pub fn reconcile_acknowledgement(
  grind: grind.Grind,
  handle: job.JobHandle(input, output, error),
  command_id: String,
) -> Result(AcknowledgementReceipt, Error)

Reads the receipt of the acknowledgement command_id, from a [grind, job, acknowledged] event or a diagnostic, to settle whether it committed.

resolution

</>

pub fn resolution(
  decision: Decision(output, error),
  id id: String,
  by by: String,
  details details: String,
) -> Resolution(output, error)

A decision, identified by id so a repeated call after a lost reply applies it once, and attributed to by with details.

IDs are unique across one installation (the same PostgreSQL database and Grind schema), not just within a job, queue or worker. Retain one ID for one exact command. Retry it with the same job, decision, value, author and details; another job or changed command returns ResolutionConflict. A later decision needs a new ID. A durable application decision ID, qualified by the job ID when necessary, can supply this identity. Do not generate a new ID merely because the original reply was lost.

The receipt is removed when its job is pruned. These IDs and attribution fields neither authenticate the operator nor replace business idempotency.

resolve_uncertain

</>

pub fn resolve_uncertain(
  grind: grind.Grind,
  handle: job.JobHandle(input, output, error),
  resolution: Resolution(output, error),
) -> Result(Resolved, Error)

Applies an audited decision to an uncertain job.

resolve_uncertain_in

</>

pub fn resolve_uncertain_in(
  grind: grind.Grind,
  transaction: pog.Connection,
  handle: job.JobHandle(input, output, error),
  resolution: Resolution(output, error),
) -> Result(StagedResolution, Error)

Resolves an uncertain job together with application writes in the caller’s open READ COMMITTED transaction on the same PostgreSQL database. Pass the connection received by a pog.transaction callback, not a pool. The caller owns commit, rollback, checkout and the outer transaction lifetime.

Grind scopes search_path to its schema and bounds lock_timeout and statement_timeout by with_statement_deadline, preserving stricter caller limits. Successful statements restore these settings before returning. A database statement error may abort the transaction; propagate every error to its owner so all staged writes roll back. There is no nested transaction.

Keep investigation and external calls outside the transaction: row locks remain held until the caller finishes it. No committed resolved event is emitted. If the outer commit reply is lost, read the application’s durable acknowledgment; a staged result is not commit evidence. Exact-command retry remains valid while the job and its receipt are retained.

statistics

</>

pub fn statistics(
  grind: grind.Grind,
  queue queue: String,
) -> Result(Statistics, Error)

Reads content-free statistics for one explicitly selected queue. Uses one statement and one database clock sample; it does not mutate or claim jobs.

Due means Queued, Scheduled or Retryable with availability reached. It does not establish worker compatibility, retry budget, capacity or authority to execute. The caller owns authorization, metric selection and sampling rate.

Storage errors are not empty queues. This uses the existing storage deadline and its pool-checkout limitations. Result size is bounded by the state set; read cost grows with retained rows in the selected queue.