grind/unique

Uniqueness policies: which jobs count as duplicates of each other, for how long, and what a duplicate submission does.

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

let policy =
  unique.policy(unique.full_input(), unique.within(duration.hours(1), unique.FromInsertion))
grind.submit(jobs, job.new(mailer(), email) |> job.unique(policy))

A job submitted with a policy is admitted only when no job of the same worker and version with the same key occupies it. Otherwise the submit returns grind.Existing (or grind.Rescheduled with reschedule_to) with the occupying job. The policy is checked under an advisory lock in the admission transaction, so two concurrent submissions never both insert. See the “Admission and uniqueness” design section.

The defaults are: scope WithinQueue, states Incomplete, and a duplicate leaves the existing job unchanged.

Types

Key

</>

Which part of the input forms the key. Build it with full_input or selected.

pub type Key(input) =
  @internal Key(input)

Origin

</>

What a finite period is measured from.

pub type Origin {
  FromInsertion
  FromSchedule
}

Constructors

```
FromInsertion
```
The job’s insertion time.
```
FromSchedule
```
The job’s scheduled time.

Period

</>

How long a job occupies its key. Build it with within or while_retained.

pub type Period =
  @internal Period

Policy

</>

A complete uniqueness policy. Build it with policy.

pub type Policy(input) =
  @internal Uniqueness(input)

QueueScope

</>

Whether a key is shared by every queue in the schema.

pub type QueueScope {
  WithinQueue
  AcrossQueues
}

Constructors

```
WithinQueue
```
```
AcrossQueues
```

States

</>

Which stored states count as still occupying a key.

pub type States {
  Incomplete
  ScheduledOnly
  IncompleteOrSucceeded
  AllRetained
}

Constructors

```
Incomplete
```
queued, scheduled, retryable, executing and uncertain.
```
ScheduledOnly
```
scheduled only.
```
IncompleteOrSucceeded
```
Incomplete plus succeeded.
```
AllRetained
```
Every stored state.

Values

full_input

</>

pub fn full_input() -> Key(input)

The whole encoded input is the key.

policy

</>

pub fn policy(key: Key(input), period: Period) -> Policy(input)

A policy over key for period, scoped to the submitting queue, over the Incomplete states, that leaves a duplicate’s existing job unchanged.

reschedule_to

</>

pub fn reschedule_to(
  policy: Policy(input),
  at: timestamp.Timestamp,
) -> Policy(input)

A duplicate moves a scheduled existing job to at; an existing job in any other state is left unchanged. The submit then returns grind.Rescheduled.

selected

</>

pub fn selected(
  name: String,
  select: fn(input) -> key,
  codec: worker.Codec(key),
) -> Key(input)

A projection of the input, encoded with its own codec, is the key. The name and the codec’s version identify the key, so a changed projection never matches keys of the old one. If the codec rejects the projected value, the submit returns InvalidInput. Panics on an empty name.

while_retained

</>

pub fn while_retained() -> Period

A job occupies its key for as long as it is stored.

with_scope

</>

pub fn with_scope(
  policy: Policy(input),
  scope: QueueScope,
) -> Policy(input)

Sets whether the key is shared across queues.

with_states

</>

pub fn with_states(
  policy: Policy(input),
  states: States,
) -> Policy(input)

Sets which states occupy the key.

within

</>

pub fn within(
  period: duration.Duration,
  from from: Origin,
) -> Period

A job occupies its key for period, measured from from. Panics on a period that is not positive or above about 285 years.