Crate agent_effects_postgres 

PostgreSQL store for agent-effects.

For production services: many workers on many hosts sharing one database.

let store = agent_effects_postgres::PostgresStore::connect("postgres://localhost/effects").await?;

Sharing an application’s database

PostgresOptions fits the store into a database an application already uses, including through PgBouncer:

• schema keeps the store’s tables and its migration history in a schema of their own, named in every query. The application’s own sqlx migrations and search_path are left alone.
• migrate and attach are separate steps: migrate once over a direct connection, then attach to the application’s pool without migrating.
• transaction_pooling makes every query safe through a pooler in transaction mode.

use agent_effects_postgres::PostgresOptions;
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};

let options = PostgresOptions::new()
    .schema("agent_effects")
    .transaction_pooling(true);
// Once, at deploy or startup, over a direct connection to Postgres.
options.migrate(&direct).await?;
// The application's pool, which goes through PgBouncer. sqlx sends the
// `extra_float_digits` startup parameter unless told not to, and a
// PgBouncer that does not ignore it refuses the connection.
let pooled = PgPoolOptions::new()
    .connect_with(
        "postgres://app@pgbouncer:6432/app"
            .parse::<PgConnectOptions>()?
            .extra_float_digits(None),
    )
    .await?;
let store = options.attach(pooled).await?;

The database’s clock

By default (ClockSource::Database), every time the store uses comes from Postgres’s clock_timestamp(), not from the calling worker:

• lease liveness, lease expiry and takeover;
• the times recorded on transitions;
• the “no live lease” filter of recovery scans.

Workers whose clocks disagree therefore still agree on who holds a lease. This removes the clock-skew limit that leases otherwise have. Retry schedules (next_attempt_at) are still computed by the worker. ClockSource::Caller uses the times the runtime passes in instead, as the in-memory and SQLite stores do; the conformance suite needs it to drive time.

Concurrency

Every change is “load the row FOR UPDATE, apply the pure EffectRecord operation, save” in one transaction, so concurrent writers serialize on the row. The update also checks the version it read. Recovery and pending scans read with FOR UPDATE SKIP LOCKED: rows another worker is changing right now are skipped, not waited on.

Structs

PostgresOptions

Where the store keeps its tables and how it talks to the database.

PostgresStore

An EffectStore in a PostgreSQL database.

Enums

ClockSource

Whose clock the store trusts for leases and recorded times.