Crate axa_postgres_job_queue

Expand description

A Postgres job queue, worker pool, and recurring schedules.

Jobs live in one axa_queue table. Each job loop claims the ready job with the highest priority, oldest first, in a single statement using FOR UPDATE SKIP LOCKED, so two loops can never claim the same row and no broker is needed.

use std::time::Duration;
use axa_postgres_job_queue::{NewJob, Queue, QueueOptions, Schedule};
use tokio::sync::watch;

let mut queue = Queue::connect(uri, QueueOptions::default()).await?;
queue.handle("Report", |args| async move {
    let id = args["id"].as_i64().ok_or("no id")?;
    println!("report {id}");
    Ok::<(), &str>(())
});
queue.limit("Report", 1); // at most one at a time in this process
queue.schedule(Schedule::every(
    "nightly-report",
    Duration::from_hours(24),
    NewJob::new("Report", serde_json::json!({ "id": 0 })),
));

let id = queue
    .enqueue_job(&NewJob::new("Report", serde_json::json!({ "id": 7 })).run_in(Duration::from_mins(5)))
    .await?;
let queue = std::sync::Arc::new(queue);
let (stop, stopped) = watch::channel(false);
let jobs = tokio::spawn(queue.clone().run(stopped));
// ... serve until told to stop, then:
let _ = stop.send(true);
let _ = jobs.await;

A job moves from queued to running to completed or failed, with the failure text kept on the row. A queued job can be cancelled.

§A person’s acts

A job row is state: it changes as the job runs, and retention deletes it. What a person does to a job is a record. cancel() stops a queued job, and retry() queues a failed one again or releases one from review. Each takes the caller’s transaction and a by and reason, and appends the act to axa_job_acts in the same statement as the change, so the act, the change, and the service’s own audit row commit together or not at all. There is no form that takes a pool. An act that does not apply, such as cancelling a running job, changes nothing and writes no act.

let mut transaction = queue.pool().begin().await?;
let cancelled = axa_postgres_job_queue::cancel(&mut transaction, id, "ops:alice", "customer_asked").await?;
// ... write the service's own audit row on the same transaction ...
transaction.commit().await?;

by and reason are codes, not content: at most MAX_REASON_CHARS characters with no line breaks, and the table refuses anything else. axa_job_acts is only ever inserted into, and its triggers refuse UPDATE, DELETE, and TRUNCATE, whoever sends them. An act names its job by id, with no foreign key, so it outlives the job’s row. Queue::acts reads a job’s acts.

§Attempts

A job gets QueueOptions::max_attempts tries, unless it was queued with its own limit. The default is one, because a retry repeats the job’s work, and only its author knows that is safe. Each retry waits twice as long as the last, starting from QueueOptions::retry_backoff. A job that panics counts as failed, and its loop carries on.

§Lost jobs

A running job reports in every QueueOptions::heartbeat_interval, so a job may run for as long as it needs. One not heard from for QueueOptions::stale_minutes is assumed lost with its worker. If it has attempts left it is queued again, and otherwise it is marked failed, so a crash can never loop for ever. Each claim is identified by its attempt, so a run that was given up on cannot then overwrite the result of its replacement.

§How many at once

A process runs QueueOptions::workers job loops, so at most that many jobs at once. Queue::limit caps one job type below that, in this process. A limit is per process and not across replicas: three processes with a limit of one may run three. Work that must never overlap anywhere needs a lock of its own.

§When a job runs

NewJob::run_in and NewJob::run_at delay a job, and NewJob::priority puts it ahead of others that are ready. Delay is measured on the database’s clock. A delayed job is claimed only once it is due, and then about as soon as a loop next polls.

§Schedules

A Schedule queues a job every interval. Each due time becomes one ordinary job, so its retries, status, limit, and recovery are the queue’s. The schedule’s row in axa_schedule is locked while its job is queued, in the same transaction, so across any number of replicas each due time is queued once.

No drift. The next time is counted from when the last was due, not from when it ran.
Missed times run once. After every process was down, one job is queued, and the schedule keeps its place: the next time is still a whole number of intervals from the first.
No stacking. When the schedule’s last job is still queued or running at a due time, that time is skipped.
The first time is Schedule::first_at, or as soon as the row is made. It applies only when the row is made, so a schedule every 24 hours first due at 02:00 runs at 02:00 each day, on the database’s clock in UTC.

Queue::run ticks the schedules on every poll, and a service with no job loops can call Queue::tick_schedules itself. A schedule’s job type must be handled by some worker, or its jobs wait. A schedule removed from code leaves a row that nothing reads.

§Queuing inside a transaction

enqueue_on takes the caller’s transaction, so a job and the record it is about commit together or not at all. Queue::enqueue uses its own connection, so a job queued that way can outlive a rollback.

§Personal data

The args column is plain JSONB, and finished rows are deleted after QueueOptions::retention_days, so a job table does not become a second copy of the data. Prefer an id to a snapshot, and never put a secret in arguments. JobInfo leaves the arguments out. Its error text is whatever the job’s error said, so it is for operators, and is not to be sent to a client. An act keeps no arguments either, and nothing logs its by or reason.

Structs§

Act: One act of a person on a job, as axa_job_acts keeps it. It outlives the job’s row, which retention deletes.
Counts: How many jobs are in each state, for metrics.
DateTime: ISO 8601 combined date and time with time zone.
JobInfo: What the queue knows of one job. Its arguments are left out.
NewJob: A job to queue: its type, its arguments, and when and how it runs.
Queue: A Postgres-backed job queue.
QueueOptions: Queue tuning.
Schedule: A job queued every interval. See Schedules.
ScheduleInfo: What the table holds of one schedule.
Utc: The UTC time zone. This is the most efficient time zone when you don’t need the local time. It is also used as an offset (which is also a dummy type).

Enums§

ActKind: What a person did to a job.
Error: What went wrong.
JobState: Where a job is.

Constants§

MAX_REASON_CHARS: The longest by or reason kept on an act, in characters.
SETUP_STATEMENTS: The statements that make the queue’s tables, in order, each safe to repeat. A service whose role may not change the schema runs these in a migration, and calls Queue::connect_existing. They need Postgres 14 or later.

Functions§

backoff: How long to wait before the attempt after attempts, doubling each time.
cancel: Cancel the job id if it is still queued, as an act of by, for reason, kept in axa_job_acts in the caller’s transaction, so that the caller’s own audit row commits with it. Answers whether it was cancelled. A job already claimed runs on: cancelling cannot stop work in progress. A job in any other state, or no such job, changes nothing and writes no act.
enqueue_all_on: Add every job in jobs on connection, returning their ids in order. On a transaction they commit or roll back together; Queue::enqueue_all opens one.
enqueue_on: Add job on executor, usually the caller’s transaction, and return its id. The job then commits or rolls back with the transaction.
retry: Queue the job id again, now, for one more try, if it failed or waits for review, as an act of by, for reason, kept in axa_job_acts in the caller’s transaction. The act is ActKind::Retried for a failed job and ActKind::Released for one in review. Answers whether it was queued again. A job in any other state, or no such job, changes nothing and writes no act. Its last error stays until it next runs.

Type Aliases§

Result: What this crate’s functions return.