Struct PostgresHistory 

pub struct PostgresHistory { /* private fields */ }

Available on crate features serve-history-postgres and serve only.

SQL-backed RunHistory. See crate::serve::history::sql for the shared schema + semantics.

Implementations

impl PostgresHistory

pub fn from_parts( pool: PgPool, idem_retention: Duration, lease_ttl: Duration, instance_id: String, stmts: Stmts, ) -> Self

Assemble from an already-connected pool (used by connect).

pub fn pool(&self) -> &PgPool

Borrow the underlying pool (tests close it to exercise fallback).

impl PostgresHistory

pub async fn connect( url: &str, idem_retention: Duration, lease_ttl: Duration, instance_id: String, ) -> Result<Self, HistoryError>

Connect, create the schema if absent, and return the backend. lease_ttl and instance_id drive instance-fenced orphan recovery (#146 H7).

Trait Implementations

impl RunHistory for PostgresHistory

fn claim_idempotency<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, key: &'life1 str, fingerprint: &'life2 str, run_id: &'life3 str, window: Duration, ) -> Pin<Box<dyn Future<Output = Result<Claim, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Atomically claim key for run_id (or report a replay/conflict). A prior claim older than window is treated as expired and re-claimable.

fn upsert<'life0, 'life1, 'async_trait>( &'life0 self, rec: &'life1 RunRecord, ) -> Pin<Box<dyn Future<Output = Result<(), HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Insert or replace a run record.

fn get<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<RunRecord>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn list<'life0, 'life1, 'async_trait>( &'life0 self, filter: &'life1 ListFilter, ) -> Pin<Box<dyn Future<Output = Result<ListPage, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn delete<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<DeleteOutcome, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Delete a terminal run. Non-terminal → StillRunning (caller maps to 409).

fn release_idempotency<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Release the idempotency claim(s) pointing at run_id. Called on the submit path when the run-record write that should immediately follow a Fresh claim fails — without it, a fallible (SQL) backend would orphan the claim, so every replay of the key 404s until the claim self-expires within the retention window (F21). Scoped by run_id, so a newer run that re-claimed the same key keeps its claim. Best-effort. Default: no-op — the in-memory backend’s upsert is infallible, so a Fresh claim is always paired with a record.

fn purge_expired<'life0, 'async_trait>( &'life0 self, retain_for: Duration, ) -> Pin<Box<dyn Future<Output = Result<usize, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Drop terminal records finished longer than retain_for ago. Returns the number removed.

fn recover_orphans<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<usize, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Mark non-terminal records whose owning instance’s lease has expired as failed (instance-fenced orphan recovery — never touches a live peer’s heartbeated runs, #146 H7). Returns the number recovered. The memory backend has nothing to recover (returns 0).

fn renew_leases<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<usize, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Heartbeat: extend the lease of this instance’s own non-terminal runs so a peer’s recover_orphans won’t reclaim them. Returns the number of leases renewed. The memory backend (single-process, unshared) is a no-op returning 0.

fn claim_pending<'life0, 'async_trait>( &'life0 self, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<RunRecord>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Atomically claim up to limit oldest Pending runs for this instance, moving them Pending → Running with a fresh lease, and return the claimed records (with config_body) for the caller to execute. Exclusive: a run claimed by one caller is never returned to another. Default: none (memory is single-process and never writes Pending).

fn reclaim_orphans<'life0, 'async_trait>( &'life0 self, max_attempts: u32, ) -> Pin<Box<dyn Future<Output = Result<ReclaimReport, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Cluster failover: expired-lease Running runs whose attempt < max_attempts go back to Pending (owner/lease cleared, attempt++); the rest are Failed (poison). Returns the counts. Default: nothing to reclaim.

fn finalize_owned<'life0, 'life1, 'async_trait>( &'life0 self, rec: &'life1 RunRecord, ) -> Pin<Box<dyn Future<Output = Result<bool, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Owner-fenced terminal write: persist rec only if this instance still owns the run. Returns true if the write landed, false if another instance reclaimed it (the caller should discard its result). Default: delegate to upsert (memory/single-process always owns its runs).

fn finalize_sharded_parent<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, status: RunStatus, finished_at: DateTime<Utc>, error: Option<String>, ) -> Pin<Box<dyn Future<Output = Result<bool, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Status-fenced finalize of a Sharded parent run: set the terminal status / finished_at / error only while the run is still Sharded, and — crucially — WITHOUT re-stamping owner / lease_expires_at. A terminal record must not re-arm a lease, and two shards finishing on two instances at once must not last-writer-wins overwrite each other via the owner-stamping upsert (F45). Returns true if this call performed the transition (the first finalizer wins; a concurrent second call is a no-op). Default: read-guard-write via upsert — correct for the single-process in-memory backend, which has no cross-instance race and no lease columns. The SQL backends override this with one conditional UPDATE.

fn cancel_pending<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<bool, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Cancel a still-Pending (unclaimed) run directly. Returns true if it was pending and is now Cancelled; false if it had already been claimed (the caller should fall back to request_cancel). Default: false.

fn request_cancel<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Flag a Running run for cross-instance cancellation; its owning instance fires the local cancel on its next claim-loop tick. Default: no-op.

fn pending_cancellations<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

This instance’s own Running runs that have a pending cancel request. Default: none.

fn heartbeat_instance<'life0, 'life1, 'async_trait>( &'life0 self, beat: &'life1 InstanceHeartbeat, ) -> Pin<Box<dyn Future<Output = Result<(), HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Membership heartbeat: upsert this instance’s liveness row. Default: no-op.

fn live_instances<'life0, 'async_trait>( &'life0 self, ttl: Duration, ) -> Pin<Box<dyn Future<Output = Result<Vec<InstanceRecord>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Live cluster members (last heartbeat within ttl). Default: none.

fn insert_shards<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, run_id: &'life1 str, shards: &'life2 [ShardInsert], ) -> Pin<Box<dyn Future<Output = Result<usize, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Idempotently insert the shard set for run_id (INSERT … ON CONFLICT (run_id, shard_id) DO NOTHING), so concurrent coordinators converge on the same set without a leader. Returns the number of rows newly inserted. Default: no-op.

fn claim_shards<'life0, 'async_trait>( &'life0 self, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<ClaimedShard>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Atomically claim up to limit pending shards for this instance (pending → running with a fresh lease), largest-estimated-size first for skew-aware balancing, returning each with its parent run record. Exclusive, like claim_pending. Default: none.

fn renew_shard_leases<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<usize, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Heartbeat: extend the lease of this instance’s own running shards so a peer’s reclaim_shards won’t reassign them. Returns the number renewed. Default: no-op.

fn reclaim_shards<'life0, 'async_trait>( &'life0 self, max_attempts: u32, ) -> Pin<Box<dyn Future<Output = Result<ReclaimReport, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Rebalance: expired-lease running shards whose attempt < max_attempts go back to pending (owner cleared, attempt++) for another worker to claim; the rest are failed (poison). Returns the counts. Default: none.

fn finalize_shard<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, run_id: &'life1 str, shard_id: &'life2 str, success: bool, ) -> Pin<Box<dyn Future<Output = Result<bool, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Owner-fenced terminal write for one shard (running → completed/failed), only if this instance still owns it. Returns true if the write landed. Default: false.

fn shard_progress<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<ShardProgress, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Aggregate shard status counts for a run (drives parent-run finalization). Default: empty.

fn pending_shard_cancellations<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Distinct run_ids for which THIS instance owns a running shard whose parent run has a pending cancellation request (F10). The claim loop fires each returned run’s local shard tokens via Registry::cancel_run_shards. Default: none (single-process / memory owns no cross-instance shards).

fn finalize_completed_sharded_parents<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<usize, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Sweep sharded parent runs whose shards are ALL terminal and finalize each to Completed (no failures) or Failed, stamping finished_at (F11). Recovers a parent that no shard task finalized inline (e.g. the coordinator crashed after the last shard completed elsewhere). Returns the number finalized. Status-fenced, so a concurrent inline finalize is a benign no-op and the run-finished metric is never double-counted. Default: nothing to finalize.

fn degraded(&self) -> bool

True when the backend is in fallback mode (drives /readyz). Always false for memory.