Struct PostgresStorage 

pub struct PostgresStorage<Args> { /* private fields */ }

A backend for persisting and consuming jobs behind a postgres database

Features

Feature	Status	Description
Backend	✅	Supports storage and retrieval of tasks
TaskSink	✅	Ability to push new tasks
Serialization	✅	Serialization support for arguments
Workflow	✅	Flexible enough to support workflows
Web Interface	✅	Expose a web interface for monitoring tasks
FetchById	✅	Allow fetching a task by its ID
RegisterWorker	✅	Allow registering a worker with the backend
MakeShared	✅	Share one connection across multiple workers via PostgresStorageFactory
WaitForCompletion	✅	Wait for tasks to complete without blocking
ResumeById	✅	Resume a task by its ID
ResumeAbandoned	✅	Resume abandoned tasks
ListWorkers	✅	List all workers registered with the backend
ListTasks	✅	List all tasks in the backend

Key: ✅ : Supported | ⚠️ : Not implemented | ❌ : Not Supported | ❗ Limited support

Tests:

Backend

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    async fn task(task: u32, worker: WorkerContext) {
        // Do some work 
    }
    let worker = WorkerBuilder::new("backend-test")
        .backend(backend)
        .build(task);
    let _ = worker.stream().take(1).collect::<Vec<_>>().await;
}

TaskSink

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    backend.push(42).await.unwrap();

    async fn task(task: u32, worker: WorkerContext) {
        worker.stop().unwrap();
    }
    let worker = WorkerBuilder::new("task-sink-test")
        .backend(backend)
        .build(task);
    worker.run().await.unwrap();
}

Serialization

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
   fn assert_codec<B: BackendConfig + WireFormatBackend>(backend: B) 
   where
       B::Codec: Codec<B::Args, Compact=Vec<u8>>,
   {
   }
   assert_codec(backend);
}

Workflow

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    backend.push(42).await.unwrap();

    async fn task1(task: u32, worker: WorkerContext) -> u32 {
        task + 99 
    }
    async fn task2(task: u32, worker: WorkerContext) -> u32 {
        task + 1 
    }
    async fn task3(task: u32, worker: WorkerContext) {
        assert_eq!(task, 142);
        worker.stop().unwrap();
    }
    let workflow = SteppedFlow::new("test-workflow")
       .and_then(task1)
       .and_then(task2)
       .and_then(task3);
    let worker = WorkerBuilder::new("workflow-test")
        .backend(backend)
        .build(workflow);
    worker.run().await.unwrap();
}

WebUI

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    fn assert_web_ui<B: Expose<u32, Durable>>(backend: B) {};
    assert_web_ui(backend);
}

WaitForCompletion

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    fn assert_wait_for_completion<B: WaitForCompletion<u32> + Backend>(backend: B) {};
    assert_wait_for_completion(backend);
}

Implementations

impl PostgresStorage<()>

pub async fn setup(pool: &PgPool) -> Result<(), Error>

Runs the PostgreSQL storage migrations.

Fresh databases

No manual setup is required. Calling setup() will create the required tables and migration history.

Upgrading to 1.0

⚠️ Important: Existing databases created by a pre-1.0 version require a one-time manual migration before calling setup().

The 1.0 migration history is no longer relocated automatically by setup(). Follow the “Upgrading to 1.0” section in the README to perform the required transition.

After the transition has been completed, setup() can be used normally for subsequent migrations.

Example

use apalis_postgres::PostgresStorage;
use sqlx::PgPool;

PostgresStorage::<()>::setup(&pool).await?;

Errors

Returns an error if the migrations cannot be applied to the database.

pub fn migrations() -> Migrator

Get postgres migrations without running them

impl<Args> PostgresStorage<Args>

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

Creates a new PostgresStorage instance.

pub fn with_pubsub(self) -> PollWith<Self, StreamStrategy<Pubsub>>

Mount a standalone Pubsub which uses its own connection under the hood

pub fn with_config(self, config: Config) -> Self

Configure a new PostgresStorage instance.

pub fn pool(&self) -> &PgPool

Returns a reference to the pool.

pub fn config(&self) -> &Config

Returns a reference to the config.

Trait Implementations

impl<Args> Backend for PostgresStorage<Args>

type Task = Task<Vec<u8>>

The type of task the backend emits.

type Error = Error

The error type returned by backend operations

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

Polls whether the backend is ready for the worker to request more work.

fn poll_next( &mut self, cx: &mut Context<'_>, worker: &WorkerContext, ) -> Poll<Option<Result<PgTask, Self::Error>>>

Polls the backend for the next available task for this worker.

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

Flushes/releases any resources the backend holds (pending acks, open subscriptions, connections) before the worker fully shuts down.

impl<Args> BackendConfig for PostgresStorage<Args>

type Args = Args

The type of argument this backend emits

type Kind = Durable

This defines the kind of Backend

type Id = Ulid

The internal type of this Backend’s TaskId

type Config = Config

The config for the backend

type Layer = TaskPersistLayer<JsonCodec<Value>, Value>

The type representing backend middleware layer.

fn config(&self) -> &Self::Config

Returns the config associated with the backend.

fn middleware(&mut self, _: &mut WorkerContext) -> Self::Layer

Returns the backend’s middleware layer.

impl<Args> Clone for PostgresStorage<Args>

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<Args> FetchById for PostgresStorage<Args>

where Self: Backend<Error = Error, Task = PgTask>, Args: 'static,

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

Fetch a task by its unique identifier

impl<Args> ListAllTasks for PostgresStorage<Args>

where PostgresStorage<Args>: Backend<Error = Error>,

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

List tasks matching the given filter in all queues

impl<Args> ListQueues for PostgresStorage<Args>

where PostgresStorage<Args>: Backend<Error = Error>,

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

List all available queues in the backend

impl<Args> ListTasks for PostgresStorage<Args>

where PostgresStorage<Args>: Backend<Error = Error>, Args: 'static,

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

List tasks matching the given filter in the current queue

impl<Args: Sync> ListWorkers for PostgresStorage<Args>

where PostgresStorage<Args>: Backend<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> Metrics for PostgresStorage<Args>

where PostgresStorage<Args>: Backend<Error = Error>,

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

Collects and returns global statistics from the backend

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

Collects and returns statistics for a specific queue

impl<Args> Sink<Task<Vec<u8>>> for PostgresStorage<Args>

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

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) -> 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<'pin, Args> Unpin for PostgresStorage<Args>

where PinnedFieldsOf<__PostgresStorage<'pin, Args>>: Unpin,

impl<O: 'static + Send, Args> WaitForCompletion<O> for PostgresStorage<Args>

where PostgresStorage<Args>: Backend<Error = Error>, Result<O, String>: DeserializeOwned,

type ResultStream = Pin<Box<dyn Stream<Item = Result<TaskResult<O>, <PostgresStorage<Args> as Backend>::Error>> + Send>>

The result stream type yielding task results

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

Wait for multiple tasks to complete, yielding results as they become available

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

Check current status of tasks without waiting

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

Wait for a single task to complete, yielding its result

impl<Args> WireFormatBackend for PostgresStorage<Args>

type Codec = JsonCodec

The codec used to encode and decode tasks.

type Compact = Vec<u8>

The compact representation of task arguments.

fn codec(&self) -> &Self::Codec

Returns a reference to the codec used by the backend.