Struct PostgresTaskStore 

pub struct PostgresTaskStore { /* private fields */ }

PostgreSQL-backed TaskStore.

Stores tasks as JSONB blobs in a tasks table. Suitable for multi-node production deployments that need shared persistence and horizontal scaling.

Schema

The store auto-creates the following table on first use:

CREATE TABLE IF NOT EXISTS tasks (
    id         TEXT PRIMARY KEY,
    context_id TEXT NOT NULL,
    state      TEXT NOT NULL,
    data       JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

list() returns tasks most-recently-updated first (spec §3.1.4), ordered by (updated_at DESC, id DESC) with a composite row-value cursor. The cursor carries updated_at as a UTC-normalized microsecond string, so pagination is stable regardless of the connection’s session time zone.

Implementations

impl PostgresTaskStore

pub const fn with_max_page_size(self, max: u32) -> Self

Caps the page size list returns, however large a page is asked for.

Defaults to DEFAULT_MAX_PAGE_SIZE, which explains why this store needs its own knob rather than reading TaskStoreConfig.

pub async fn new(url: &str) -> Result<Self, Error>

Opens a PostgreSQL connection pool and initializes the schema.

Errors

Returns an error if the database cannot be opened or the schema migration fails.

pub async fn with_migrations(url: &str) -> Result<Self, Error>

Opens a PostgreSQL database with automatic schema migration.

Runs all pending migrations before returning the store. This is the recommended constructor for production deployments because it ensures the schema is always up to date without duplicating DDL statements.

Errors

Returns an error if the database cannot be opened or any migration fails.

pub async fn from_pool(pool: PgPool) -> Result<Self, Error>

Creates a store from an existing connection pool.

Errors

Returns an error if the schema migration fails.

pub async fn purge_expired( &self, policy: &RetentionPolicy, ) -> A2aResult<PurgeReport>

Deletes terminal tasks that have outlived policy.

Nothing calls this for you. A persistent store keeps every task until an operator says otherwise — see retention for why that is the default and why the in-memory store does the opposite — so this is the hook for whatever already schedules work: a cron entry, a Kubernetes CronJob, a tokio interval in your own binary.

Only Completed, Failed, Canceled and Rejected tasks are eligible. A task still Working, or parked in InputRequired waiting on a human, is never deleted however old it is.

Safe to run from several replicas at once: each batch is a single DELETE whose subquery picks the rows, so two sweeps racing delete disjoint sets rather than colliding.

Errors

Returns an error if a delete fails. A sweep that fails partway has still committed its earlier batches; the counts in the returned report are lost in that case, but the deletions are not undone and the next sweep simply continues.

Trait Implementations

impl Clone for PostgresTaskStore

fn clone(&self) -> PostgresTaskStore

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for PostgresTaskStore

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

Formats the value using the given formatter.

impl TaskStore for PostgresTaskStore

fn save_artifact_delta<'a>( &'a self, task: &'a Task, delta: ArtifactDelta, ) -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>>

Appends into the stored JSONB document instead of rewriting it.

save serializes the whole task in Rust and ships it as a bind parameter, so a streaming agent re-sends every artifact it has already persisted on every subsequent event. This sends only what changed and lets PostgreSQL splice it in with jsonb_set.

Unlike the SQLite implementation, which needs one path expression per appended part, jsonb’s || concatenates two arrays — so any number of parts lands in a single statement with constant SQL text.

What this does and does not remove

Removed: the Rust-side serde_json::to_value of the whole task and the transfer of the whole document. Both scale with the stream so far.

Not removed: PostgreSQL still rewrites the row. An UPDATE writes a new tuple version under MVCC, and a JSONB document past the TOAST threshold is rewritten out of line, so the statement stays linear in document size. Only a normalized artifacts table could avoid that, and the measurement in benches/benches/backpressure.rs puts the per-event round trip well above the document-size term — so that surgery would buy the smaller half. Recorded here rather than left implied.

updated_at is deliberately untouched: it carries the status timestamp that orders list (§3.1.4), and appending an artifact does not change a task’s status. Both other stores behave the same way, and a divergence here would be invisible until someone paginated.

Falls back to save when the delta cannot be applied exactly: no artifacts on the task, an index out of range, a Pushed that does not name the last position, fewer parts present than claimed, or a stored row whose document has no matching array. A store that is quietly wrong is worse than one that is slower.

fn save<'a>( &'a self, task: &'a Task, ) -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>>

Saves (creates or updates) a task.

fn get<'a>( &'a self, id: &'a TaskId, ) -> Pin<Box<dyn Future<Output = A2aResult<Option<Task>>> + Send + 'a>>

Retrieves a task by its ID, returning None if not found.

fn list<'a>( &'a self, params: &'a ListTasksParams, ) -> Pin<Box<dyn Future<Output = A2aResult<TaskListResponse>> + Send + 'a>>

Lists tasks matching the given filter parameters.

fn insert_if_absent<'a>( &'a self, task: &'a Task, ) -> Pin<Box<dyn Future<Output = A2aResult<bool>> + Send + 'a>>

Atomically inserts a task only if no task with the same ID exists.

fn delete<'a>( &'a self, id: &'a TaskId, ) -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>>

Deletes a task by its ID.

fn count<'a>( &'a self, ) -> Pin<Box<dyn Future<Output = A2aResult<u64>> + Send + 'a>>

Returns the total number of tasks in the store.