Struct Queue 

pub struct Queue { /* private fields */ }

A Postgres-backed job queue.

Implementations

impl Queue

pub fn from_pool(pool: PgPool, options: QueueOptions) -> Self

A queue on pool, checking and changing nothing. Call Queue::check at boot. Give the pool at least workers + 2 connections.

pub async fn connect_existing(uri: &str, options: QueueOptions) -> Result<Self>

Connect to tables that a migration made, and change no schema, for a database role that may not. A migration runs SETUP_STATEMENTS.

Errors

Fails when the database cannot be reached, a table is missing or out of date, or a trigger of axa_job_acts is missing or not set to fire always, so a service finds out at boot rather than at its first job. There is no way to run without the triggers.

pub async fn connect(uri: &str, options: QueueOptions) -> Result<Self>

Connect and create the tables if needed, which runs DDL at boot. Use Queue::connect_existing where a service’s role may not.

Errors

Fails when the database cannot be reached or the tables not created.

pub async fn check(&self) -> Result<()>

Check that the tables exist and are up to date, and that the triggers of axa_job_acts are there and set to fire always.

Errors

Fails when the database cannot be reached, a table is missing or out of date, or a trigger is missing or turned off.

pub const fn pool(&self) -> &PgPool

The pool, for a service’s own queries or enqueue_on outside a transaction.

pub const fn options(&self) -> &QueueOptions

The options this queue runs with.

pub fn handle<F, Fut, E>(&mut self, name: impl Into<String>, handler: F)

where F: Fn(Value) -> Fut + Send + Sync + 'static, Fut: Future<Output = Result<(), E>> + Send + 'static, E: Display + 'static,

Run handler for jobs named name. Its error text is kept on the job’s row. Registering a name again replaces its handler.

pub fn limit(&mut self, name: impl Into<String>, jobs: usize)

Run at most jobs jobs named name at once in this process, and at least one. Other jobs keep the remaining loops busy.

pub fn schedule(&mut self, schedule: Schedule)

Queue a job on every due time of schedule.

Panics

When a schedule of that name was added already.

pub async fn enqueue(&self, name: &str, args: Value) -> Result<Uuid>

Add a job and return its id.

Errors

Fails on a database error.

pub async fn enqueue_limited( &self, name: &str, args: Value, max_attempts: Option<i32>, ) -> Result<Uuid>

Add a job with its own limit on tries, or the queue’s when None.

Errors

Fails on a database error.

pub async fn enqueue_job(&self, job: &NewJob) -> Result<Uuid>

Add job and return its id.

Errors

Fails on a database error.

pub async fn enqueue_all(&self, jobs: &[NewJob]) -> Result<Vec<Uuid>>

Add every job in jobs in one transaction, so all are queued or none.

Errors

Fails on a database error, and then none was queued.

pub async fn job(&self, id: Uuid) -> Result<Option<JobInfo>>

What the queue knows of the job id, or None when there is no such job, as after retention deleted it.

Errors

Fails on a database error.

pub async fn acts(&self, id: Uuid) -> Result<Vec<Act>>

Every act a person did to the job id, oldest first, for a service’s own review screens. Acts are kept after retention deletes the job’s row. A person acts with cancel() and retry(), in the caller’s transaction.

Errors

Fails on a database error.

pub async fn counts(&self) -> Result<Counts>

How many jobs are in each state.

Errors

Fails on a database error.

pub async fn ping(&self) -> Result<()>

Check the database connection.

Errors

Fails when the database does not answer.

pub async fn requeue_stale(&self) -> Result<u64>

Recover jobs abandoned by a worker that died: queue again those with attempts left and fail the rest, returning how many were queued again.

Errors

Fails on a database error.

pub async fn purge_finished(&self) -> Result<u64>

Delete finished and cancelled rows older than the retention period, returning how many went.

Errors

Fails on a database error.

pub async fn tick_schedules(&self) -> Result<usize>

Queue a job for each schedule that is due, returning how many were queued. Queue::run calls this on every poll. Any number of processes may call it at once: each due time is queued once.

Errors

Fails on a database error, which leaves that schedule due for the next tick.

pub async fn archive_schedules(&self) -> Result<u64>

Archive the schedules that no live process has in code, returning how many were archived. Rows are kept, for audit. Housekeeping runs this.

Errors

Fails on a database error.

pub async fn schedules(&self) -> Result<Vec<ScheduleInfo>>

Every schedule the table holds, archived or not, with its last job’s state.

Errors

Fails on a database error.

pub async fn run_one(&self) -> Result<bool>

Claim and run one job, returning whether there was one. A job’s own failure is recorded on its row, not returned.

Errors

Fails on a database error.

pub async fn run(self: Arc<Self>, shutdown: Receiver<bool>)

Run job loops, housekeeping, and schedules until shutdown changes to true. Jobs in progress finish first, so a caller that must stop by a time limit drops this future at it, which aborts those jobs and leaves them for stale recovery. Housekeeping runs straight away, to recover anything a previous process left behind, and then on a timer.

pub fn handled(&self) -> HashSet<&str>

The names of the job types this queue handles.

Trait Implementations

impl Debug for Queue

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.