Jobs / Ids

On this page

Four calls, one table
Debounce and throttle
Once per hour
Pages of jobs
See also

Jobs / Ids

Ids

An id names one job of a queue, so that a program can find it, change it, cancel it, or add it once however many times it is asked. Most jobs need none. A message scheduled for later, a reminder a user edits, a check that moves each time a device pings: those get an id.

const later = store.queue<Draft>('later')

const added = await later.add(draft, { id: `chat:${chatId}:${draft.id}`, at: nineAm }) // false: a second tap on "send"
const edited = await later.update(`chat:${chatId}:${draft.id}`, draft, { at: tenAm }) // false: too late, it ran
const cancelled = await later.cancel(`chat:${chatId}:${draft.id}`) // false: there was none
const job = await later.get(`chat:${chatId}:${draft.id}`) // { id, value, state, at, attempt, … } | undefined

later = store.queue("later", Draft)

added = await later.add(draft, id=f"chat:{chat_id}:{draft.id}", at=nine_am)  # False: a second tap on "send"
edited = await later.update(f"chat:{chat_id}:{draft.id}", draft, at=ten_am)  # False: too late, it ran
cancelled = await later.cancel(f"chat:{chat_id}:{draft.id}")  # False: there was none
job = await later.get(f"chat:{chat_id}:{draft.id}")  # Job(id, value, state, at, attempt, …) | None

later, err := jobs.OpenQueue[Draft](store, "later")

id := fmt.Sprintf("chat:%d:%s", chatID, draft.ID)
added, err := later.Add(ctx, draft, jobs.ID(id), jobs.At(nineAM)) // false: a second tap on "send"
edited, err := later.Update(ctx, id, draft, jobs.At(tenAM))      // false: too late, it ran
cancelled, err := later.Cancel(ctx, id)                           // false: there was none
job, found, err := later.Get(ctx, id)

let later = store.queue::<Draft>("later").open()?;

let id = format!("chat:{chat_id}:{}", draft.id);
let added: bool = later.id(&id).at(nine_am).add(&draft)?; // false: a second tap on "send"
let edited: bool = later.id(&id).at(ten_am).update(&draft)?; // false: too late, it ran
let cancelled: bool = later.cancel(&id)?; // false: there was none
let job = later.get(&id)?;

Four calls, one table

An id is taken while its job is scheduled, waiting, running or failed, and while dedupe keeps it once the job is done.

The id's job is	`add`	`set`	`update`	`cancel`
none	adds it: `true`	adds it	nothing: `false`	`false`
scheduled, or waiting	nothing: `false`	replaces its value and time	changes it: `true`	removes it: `true`
running	nothing: `false`	one run more after this one, with this value	nothing: `false`	tells its handler to stop: `true`
failed, kept for `keep`	nothing: `false`	starts it again, its attempts from one	nothing: `false`	removes it: `true`
done, while `dedupe` keeps it	nothing: `false`	adds it again	nothing: `false`	`false`

add, update and cancel answer rather than throw: a message being sent says it is too late to edit, and the program shows that.

Debounce and throttle

await indexing.set(`note:${id}`, { noteId: id }, { delay: '5s' }) // debounce: each edit moves it 5 s on
await reports.add({ team }, { id: `report:${team}`, delay: '1m' }) // throttle: the first call stands

await indexing.set(f"note:{id}", Index(note_id=id), delay="5s")  # debounce: each edit moves it 5 s on
await reports.add(Report(team=team), id=f"report:{team}", delay="1m")  # throttle: the first call stands

err = indexing.Set(ctx, fmt.Sprint("note:", id), Index{NoteID: id}, jobs.Delay(5*time.Second))           // debounce
_, err = reports.Add(ctx, Report{Team: team}, jobs.ID("report:"+team), jobs.Delay(time.Minute)) // throttle

indexing.id(&format!("note:{id}")).delay(Duration::from_secs(5)).set(&Index { note_id: id })?; // debounce
reports.id(&format!("report:{team}")).delay(Duration::from_mins(1)).add(&Report { team })?; // throttle

Once per hour

const pushes = store.queue<Push>('pushes', { dedupe: '1h' })
await pushes.add(push, { id: `digest:${userId}` }) // a second add within the hour after it ran adds nothing

pushes = store.queue("pushes", Push, dedupe="1h")
await pushes.add(push, id=f"digest:{user_id}")

pushes, err := jobs.OpenQueue[Push](store, "pushes", jobs.Dedupe(time.Hour))
_, err = pushes.Add(ctx, push, jobs.ID(fmt.Sprint("digest:", userID)))

let pushes = store.queue::<Push>("pushes").dedupe(Duration::from_hours(1)).open()?;
pushes.id(&format!("digest:{user_id}")).add(&push)?;

With dedupe, a done job keeps its id taken for that long, so add of the same id adds nothing; set arms it again.

Pages of jobs

const page = await later.list({ prefix: `chat:${chatId}:`, limit: 100 }) // { jobs, next }
const failed = await emails.list({ state: 'failed' }) // the latest failures first

page = await later.list(prefix=f"chat:{chat_id}:", limit=100)  # Page(jobs, next)
failed = await emails.list(state="failed")

page, err := later.List(ctx, jobs.Prefix(fmt.Sprintf("chat:%d:", chatID)), jobs.Limit(100)) // jobs.Page[Draft]{Jobs, Next}
failed, err := emails.List(ctx, jobs.Failed)

A prefix is text: chat:4 matches chat 42 too, so end a prefix with a separator.

What	Limit
an id	1 to 1,024 bytes
a page	1,000 jobs or 4 MiB, 100 unless `limit` says

See also

Jobs: the engine's overview.
Watching a job: a job's state as it changes.

Edit this page on GitHub Updated 19 October 2026