Struct PostgresStorage 

pub struct PostgresStorage<Args, Codec = JsonCodec<CompactType>, Fetcher = PgFetcher<CompactType, Codec>> { /* private fields */ }

PostgreSQL storage backend implemented with Diesel.

Implementations

impl<Args> PostgresStorage<Args>

pub fn new(pool: &PgPool) -> Self

Create storage for the queue named after Args.

Do not share pool with HTTP request handlers or other unrelated workloads. Apalis holds long-lived connections (fetcher, lifecycle keep-alive, listener) and a backend that exhausts a shared pool will stall the worker, causing heartbeat loss and orphan reenqueue cascades. See the README section “Connection pool isolation” for the recommended sizing and the Self::push_with_conn outbox API for the supported way to enqueue from a backend transaction.

pub fn new_with_config(pool: &PgPool, config: &Config) -> Self

Create storage with an explicit Apalis SQL config.

Do not share pool with HTTP request handlers or other unrelated workloads — see Self::new for the rationale.

pub fn new_with_notify( pool: &PgPool, config: &Config, ) -> PostgresStorage<Args, JsonCodec<CompactType>, PgNotify>

Create storage that also listens for PostgreSQL notifications.

Notify mode uses a dedicated pooled connection for LISTEN "apalis::job::insert" while the polling stream is alive. Each new_with_notify storage spawns one listener thread and pins one pool connection. If you need notify-driven dequeue across many queues, prefer crate::SharedPostgresStorage — it spawns a single listener thread shared by all queues registered with it, so the thread/connection cost stays at one regardless of queue count.

Do not share pool with HTTP request handlers or other unrelated workloads — see Self::new for the rationale.

pub fn pool(&self) -> &PgPool

Return the underlying Diesel/r2d2 pool.

pub fn config(&self) -> &Config

Return the queue configuration.

impl<Args, Codec, Fetcher> PostgresStorage<Args, Codec, Fetcher>

pub fn with_codec<NewCodec>(self) -> PostgresStorage<Args, NewCodec, Fetcher>

Change the task codec while retaining pool, config, and fetcher.

impl<Args, EncodeCodec, Fetcher> PostgresStorage<Args, EncodeCodec, Fetcher>

where EncodeCodec: Codec<Args, Compact = CompactType>, EncodeCodec::Error: Error + Send + Sync + 'static,

Transactional enqueue on a caller-supplied connection — the outbox entry point. Use these methods when you want the task INSERT to share a transaction with your business-data writes.

pub fn push_with_conn( &self, conn: &mut PgConnection, args: Args, ) -> Result<PgTaskId, Error>

Enqueue a task using a caller-supplied PgConnection.

For transactional outbox semantics, call this inside conn.transaction(|c| ...) together with your business-data writes — the INSERT into apalis.jobs is committed only if the outer transaction commits. Without an outer transaction, Diesel auto-commits the INSERT (same behaviour as the pool-based Sink<Task> path).

NOTIFY is delivered when the (outer) transaction commits, so listeners only see tasks that were actually committed. No manual pg_notify is needed.

This is the only synchronous public method in the crate. From an async context, invoke it inside tokio::task::spawn_blocking together with your business-data writes so the entire transaction lives on one blocking task.

See Self::push_task_with_conn for the full-control variant that accepts a pre-built PgTask (custom idempotency_key, priority, run_at, max_attempts, metadata, or task_id).

Errors

• Error::Decode if the codec rejects args.
• Error::InvalidArgument if serialized metadata exceeds the byte cap, or for unreachable run_at.
• Error::Database for SQL/driver failures.

pub fn push_task_with_conn( &self, conn: &mut PgConnection, task: PgTask<Args>, ) -> Result<PgTaskId, Error>

Enqueue a fully-constructed PgTask<Args> using a caller-supplied connection. Use this when you need to set idempotency_key, priority, run_at, max_attempts, metadata, or a specific task_id.

Semantics are identical to Self::push_with_conn; see that method’s docs for the transaction/NOTIFY contract.

If task.parts.task_id is None, a fresh Ulid is generated and returned. If Some, that id is used as-is and echoed back.

Errors

• Error::Decode if the codec rejects the task’s args.
• Error::IdempotencyConflict on idempotency_key conflict — the savepoint for this batch is rolled back (the whole batch, not just the duplicate), but your outer transaction continues; decide whether to commit or roll back based on the error.
• Error::InvalidArgument if serialized metadata exceeds the byte cap.
• Error::Database for SQL/driver failures.

Trait Implementations

impl<Args, Decode, Fetcher> Backend for PostgresStorage<Args, Decode, Fetcher>

where Args: Send + 'static + Unpin, Decode: Codec<Args, Compact = CompactType> + Send + 'static, Decode::Error: Error + Send + Sync + 'static, Fetcher: PgFetcherSource,

Single generic Backend impl covering every Fetcher: PgFetcherSource. Heartbeat/middleware are identical for all three modes; the per-mode pipeline is delegated through PgFetcherSource::into_compact_stream.

type Args = Args

The type of arguments the backend handles.

type IdType = Ulid

The type used to uniquely identify tasks.

type Context = SqlContext<Pool<ConnectionManager<PgConnection>>>

Context associated with each task.

type Error = Error

The error type returned by backend operations

type Stream = Pin<Box<dyn Stream<Item = Result<Option<Task<Args, SqlContext<Pool<ConnectionManager<PgConnection>>>, Ulid>>, Error>> + Send>>

A stream of tasks provided by the backend.

type Beat = Pin<Box<dyn Stream<Item = Result<(), Error>> + Send>>

A stream representing heartbeat signals.

type Layer = PgMiddleware

The type representing backend middleware layer.

fn heartbeat(&self, worker: &WorkerContext) -> Self::Beat

Returns a heartbeat stream for the given worker.

fn middleware(&self) -> Self::Layer

Returns the backend’s middleware layer.

fn poll(self, worker: &WorkerContext) -> Self::Stream

Polls the backend for tasks for the given worker.

impl<Args, Decode, Fetcher> BackendExt for PostgresStorage<Args, Decode, Fetcher>

where Args: Send + 'static + Unpin, Decode: Codec<Args, Compact = CompactType> + Send + 'static, Decode::Error: Error + Send + Sync + 'static, Fetcher: PgFetcherSource,

type Compact = Vec<u8>

The compact representation of task arguments.

type Codec = Decode

The codec used for serialization/deserialization of tasks.

type CompactStream = Pin<Box<dyn Stream<Item = Result<Option<Task<Vec<u8>, SqlContext<Pool<ConnectionManager<PgConnection>>>, Ulid>>, <PostgresStorage<Args, Decode, Fetcher> as Backend>::Error>> + Send>>

A stream of encoded tasks provided by the backend.

fn get_queue(&self) -> Queue

Returns the queue associated with the backend.

fn poll_compact(self, worker: &WorkerContext) -> Self::CompactStream

Polls the backend for encoded tasks for the given worker.

impl<Args, Codec, Fetcher: Clone> Clone for PostgresStorage<Args, Codec, Fetcher>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<Args, Codec, Fetcher: Debug> Debug for PostgresStorage<Args, Codec, Fetcher>

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

Formats the value using the given formatter.

impl<Args, D, F> FetchById<Args> for PostgresStorage<Args, D, F>

where PostgresStorage<Args, D, F>: BackendExt<Context = PgContext, Compact = CompactType, IdType = Ulid, Error = Error>, D: Codec<Args, Compact = CompactType>, D::Error: Error + Send + Sync + 'static, Args: 'static,

fn fetch_by_id( &mut self, task_id: &PgTaskId, ) -> impl Future<Output = Result<Option<PgTask<Args>>, Self::Error>> + Send

Fetch a task by its unique identifier

impl<Args, D, F> ListAllTasks for PostgresStorage<Args, D, F>

where PostgresStorage<Args, D, F>: BackendExt<Context = PgContext, Compact = CompactType, IdType = Ulid, Error = Error>,

fn list_all_tasks( &self, filter: &Filter, ) -> impl Future<Output = Result<Vec<Task<Self::Compact, Self::Context, Self::IdType>>, Self::Error>> + Send

List tasks matching the given filter in all queues

impl<Args, D, F> ListQueues for PostgresStorage<Args, D, F>

where PostgresStorage<Args, D, F>: BackendExt<Context = PgContext, Compact = CompactType, IdType = Ulid, Error = Error>,

fn list_queues( &self, ) -> impl Future<Output = Result<Vec<QueueInfo>, Self::Error>> + Send

List all available queues in the backend

impl<Args, D, F> ListTasks<Args> for PostgresStorage<Args, D, F>

where PostgresStorage<Args, D, F>: BackendExt<Context = PgContext, Compact = CompactType, IdType = Ulid, Error = Error>, D: Codec<Args, Compact = CompactType>, D::Error: Error + Send + Sync + 'static, Args: 'static,

fn list_tasks( &self, filter: &Filter, ) -> impl Future<Output = Result<Vec<PgTask<Args>>, Self::Error>> + Send

List tasks matching the given filter in the current queue

impl<Args, D, F> ListWorkers for PostgresStorage<Args, D, F>

where Args: Sync, PostgresStorage<Args, D, F>: BackendExt<Context = PgContext, Compact = CompactType, IdType = Ulid, Error = Error>,

fn list_workers( &self, ) -> impl Future<Output = Result<Vec<RunningWorker>, Self::Error>> + Send

List all registered workers in the current queue

fn list_all_workers( &self, ) -> impl Future<Output = Result<Vec<RunningWorker>, Self::Error>> + Send

List all registered workers in all queues

impl<Args, D, F> Metrics for PostgresStorage<Args, D, F>

where PostgresStorage<Args, D, F>: BackendExt<Context = PgContext, Compact = CompactType, IdType = Ulid, Error = Error>,

fn global( &self, ) -> impl Future<Output = Result<Vec<Statistic>, Self::Error>> + Send

Scans apalis.jobs to compute global statistics. Each call evaluates 20+ FILTER aggregates over every row; cost grows linearly with the table size. Treat as a slow admin call.

fn fetch_by_queue( &self, ) -> impl Future<Output = Result<Vec<Statistic>, Self::Error>> + Send

Same shape as Self::global, scoped to the configured queue. Cost still depends on the number of jobs in that job_type.

impl<Args, D, F> RegisterWorker for PostgresStorage<Args, D, F>

where PostgresStorage<Args, D, F>: BackendExt<Context = PgContext, Compact = CompactType, IdType = Ulid, Error = Error>,

fn register_worker( &mut self, worker_id: String, ) -> impl Future<Output = Result<(), Self::Error>> + Send

Registers a worker

impl<Args, Encode, Fetcher> Sink<Task<Vec<u8>, SqlContext<Pool<ConnectionManager<PgConnection>>>, Ulid>> for PostgresStorage<Args, Encode, Fetcher>

where Args: Send + Sync + 'static, Fetcher: Unpin,

type Error = Error

The type of value produced by the sink when an error occurs.

fn poll_ready( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>

Attempts to prepare the Sink to receive a value.

fn start_send( self: Pin<&mut Self>, item: PgTask<CompactType>, ) -> Result<(), Self::Error>

Begin the process of sending a value to the sink. Each call to this function must be preceded by a successful call to poll_ready which returned Poll::Ready(Ok(())).

fn poll_flush( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>

Flush any remaining output from this sink.

fn poll_close( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>

Flush any remaining output and close this sink, if necessary.

impl<O, Args, F, Decode> WaitForCompletion<O> for PostgresStorage<Args, Decode, F>

where O: 'static + Send, PostgresStorage<Args, Decode, F>: BackendExt<Context = PgContext, Compact = CompactType, IdType = Ulid, Error = Error>, Result<O, String>: DeserializeOwned,

fn wait_for( &self, task_ids: impl IntoIterator<Item = TaskId<Self::IdType>>, ) -> Self::ResultStream

Wait for the given tasks to complete, yielding each result as it lands.

Error handling

A transient database error during polling does not abandon the batch: the poll is retried with backoff. The stream yields an Err and ends only once the failures persist across several consecutive polls. Because completed results are durable in apalis.jobs, a surfaced error is always safe to recover from by calling wait_for again with the ids that have not yet yielded a result.

type ResultStream = Pin<Box<dyn Stream<Item = Result<TaskResult<O, Ulid>, Error>> + Send>>

The result stream type yielding task results

fn check_status( &self, task_ids: impl IntoIterator<Item = TaskId<Self::IdType>> + Send, ) -> impl Future<Output = Result<Vec<TaskResult<O, Ulid>>, Self::Error>> + Send

Check current status of tasks without waiting

fn wait_for_single(&self, task_id: TaskId<Self::IdType>) -> Self::ResultStream

Wait for a single task to complete, yielding its result

impl<Args, Codec, Fetcher: Unpin> Unpin for PostgresStorage<Args, Codec, Fetcher>