Struct PostgresStore 

pub struct PostgresStore { /* private fields */ }

Clone is derived because the underlying PgPool is itself Clone (it’s Arc<PoolInner> internally) — cloning the store hands back a new wrapper around the same connection pool. Required so engine composition (EngineState<S>) can derive Clone and pass through axum with_state.

Implementations

impl PostgresStore

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

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

Build a store from an existing pool. Runs migrations on the target database. Useful when the engine owns the pool (shared with other modules) and hands a clone to the workflow module, or for tests that point many stores at different databases in the same Postgres server.

pub fn pool(&self) -> &PgPool

Expose the underlying pool (used by the engine to build a PgEngineEventBus that shares the same connection pool).

pub async fn try_acquire_leader_lock(&self) -> Result<bool>

Try to acquire pg_advisory_lock for leader election. Returns true if this instance is the leader (scheduler should run).

Trait Implementations

impl Clone for PostgresStore

fn clone(&self) -> PostgresStore

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl WorkflowStore for PostgresStore

async fn create_namespace(&self, name: &str) -> Result<()>

async fn list_namespaces(&self) -> Result<Vec<NamespaceRecord>>

async fn delete_namespace(&self, name: &str) -> Result<bool>

async fn get_namespace_stats(&self, namespace: &str) -> Result<NamespaceStats>

async fn create_workflow(&self, wf: &WorkflowRecord) -> Result<()>

async fn get_workflow(&self, id: &str) -> Result<Option<WorkflowRecord>>

async fn list_workflows( &self, namespace: &str, status: Option<WorkflowStatus>, workflow_type: Option<&str>, search_attrs_filter: Option<&str>, limit: i64, offset: i64, ) -> Result<Vec<WorkflowRecord>>

async fn update_workflow_status( &self, id: &str, status: WorkflowStatus, result: Option<&str>, error: Option<&str>, ) -> Result<()>

async fn claim_workflow(&self, id: &str, worker_id: &str) -> Result<bool>

async fn mark_workflow_dispatchable(&self, workflow_id: &str) -> Result<()>

Mark a workflow as having new events that need a worker to replay it. Idempotent — calling repeatedly is fine. Cleared by claim_workflow_task.

async fn claim_workflow_task( &self, task_queue: &str, worker_id: &str, ) -> Result<Option<WorkflowRecord>>

Atomically claim the oldest dispatchable workflow on a queue. Sets dispatch_claimed_by and dispatch_last_heartbeat, clears needs_dispatch. Returns the workflow record or None if nothing is available.

async fn release_workflow_task( &self, workflow_id: &str, worker_id: &str, ) -> Result<()>

Release a workflow task’s dispatch lease (called when the worker submits its commands batch). Only succeeds if dispatch_claimed_by matches the calling worker.

async fn release_stale_dispatch_leases( &self, now: f64, timeout_secs: f64, ) -> Result<u64>

Forcibly release dispatch leases whose worker hasn’t heartbeat’d within timeout_secs. Used by the engine’s background poller to recover from worker crashes. Returns how many leases were released (each becomes claimable again, with needs_dispatch=true).

async fn append_event(&self, ev: &WorkflowEvent) -> Result<i64>

async fn list_events(&self, workflow_id: &str) -> Result<Vec<WorkflowEvent>>

async fn get_event_count(&self, workflow_id: &str) -> Result<i64>

async fn create_activity(&self, act: &WorkflowActivity) -> Result<i64>

async fn get_activity(&self, id: i64) -> Result<Option<WorkflowActivity>>

Look up an activity by its primary key.

async fn get_activity_by_workflow_seq( &self, workflow_id: &str, seq: i32, ) -> Result<Option<WorkflowActivity>>

Look up an activity by its workflow-relative sequence number. Used for idempotent scheduling: the engine checks if (workflow_id, seq) already exists before creating a new row.

async fn claim_activity( &self, task_queue: &str, worker_id: &str, ) -> Result<Option<WorkflowActivity>>

async fn requeue_activity_for_retry( &self, id: i64, next_attempt: i32, next_scheduled_at: f64, ) -> Result<()>

Re-queue an activity for retry: clears the running state (status→PENDING, claimed_by/started_at cleared), bumps attempt, and sets scheduled_at = now + backoff so the next claim_activity won’t pick it up before the backoff elapses.

async fn complete_activity( &self, id: i64, result: Option<&str>, error: Option<&str>, failed: bool, ) -> Result<()>

async fn heartbeat_activity( &self, id: i64, _details: Option<&str>, ) -> Result<()>

async fn get_timed_out_activities( &self, now: f64, ) -> Result<Vec<WorkflowActivity>>

async fn create_timer(&self, timer: &WorkflowTimer) -> Result<i64>

async fn cancel_pending_activities(&self, workflow_id: &str) -> Result<u64>

Mark all PENDING activities of a workflow as CANCELLED so workers that haven’t claimed them yet won’t pick them up. Returns the number of rows affected. Does NOT touch RUNNING activities — those will see the cancellation when they next heartbeat or complete.

async fn cancel_pending_timers(&self, workflow_id: &str) -> Result<u64>

Mark all unfired timers of a workflow as fired without firing (effectively removing them from the timer poller). Returns the number of rows affected.

async fn get_timer_by_workflow_seq( &self, workflow_id: &str, seq: i32, ) -> Result<Option<WorkflowTimer>>

Look up an existing timer by its workflow-relative seq. Used by the engine for idempotent ScheduleTimer (deterministic replay can call schedule_timer for the same seq more than once on retries).

async fn fire_due_timers(&self, now: f64) -> Result<Vec<WorkflowTimer>>

async fn send_signal(&self, sig: &WorkflowSignal) -> Result<i64>

async fn consume_signals( &self, workflow_id: &str, name: &str, ) -> Result<Vec<WorkflowSignal>>

async fn create_schedule(&self, sched: &WorkflowSchedule) -> Result<()>

async fn get_schedule( &self, namespace: &str, name: &str, ) -> Result<Option<WorkflowSchedule>>

async fn list_schedules(&self, namespace: &str) -> Result<Vec<WorkflowSchedule>>

async fn update_schedule_last_run( &self, namespace: &str, name: &str, last_run_at: f64, next_run_at: f64, workflow_id: &str, ) -> Result<()>

async fn delete_schedule(&self, namespace: &str, name: &str) -> Result<bool>

async fn list_archivable_workflows( &self, cutoff: f64, limit: i64, ) -> Result<Vec<WorkflowRecord>>

List workflows in terminal states whose completed_at is older than cutoff and which haven’t been archived yet. Used by the optional S3 archival background task to batch candidates.

async fn mark_archived_and_purge( &self, workflow_id: &str, archive_uri: &str, archived_at: f64, ) -> Result<()>

Mark a workflow as archived (records archived_at + archive_uri) and purge its events, activities, timers, signals, and snapshots. The workflow record itself is preserved so GET /workflows/{id} still resolves with an archive pointer.

async fn upsert_search_attributes( &self, workflow_id: &str, patch_json: &str, ) -> Result<()>

Merge a JSON object patch into the workflow’s search_attributes. Keys in the patch overwrite existing keys; keys already present but not in the patch are preserved. If the current column is NULL, the patch becomes the new value.

async fn update_schedule( &self, namespace: &str, name: &str, patch: &SchedulePatch, ) -> Result<Option<WorkflowSchedule>>

Apply an in-place patch to a schedule. Only fields present on patch are updated; the rest keep their current values. Returns the updated record, or None if the schedule doesn’t exist.

async fn set_schedule_paused( &self, namespace: &str, name: &str, paused: bool, ) -> Result<Option<WorkflowSchedule>>

Flip a schedule’s paused flag. Returns the updated record, or None if the schedule doesn’t exist.

async fn register_worker(&self, w: &WorkflowWorker) -> Result<()>

async fn heartbeat_worker(&self, id: &str, now: f64) -> Result<()>

async fn list_workers(&self, namespace: &str) -> Result<Vec<WorkflowWorker>>

async fn remove_dead_workers(&self, cutoff: f64) -> Result<Vec<String>>

async fn list_child_workflows( &self, parent_id: &str, ) -> Result<Vec<WorkflowRecord>>

async fn create_snapshot( &self, workflow_id: &str, event_seq: i32, state_json: &str, ) -> Result<()>

async fn get_latest_snapshot( &self, workflow_id: &str, ) -> Result<Option<WorkflowSnapshot>>

async fn get_queue_stats(&self, namespace: &str) -> Result<Vec<QueueStats>>

async fn try_acquire_scheduler_lock(&self) -> Result<bool>

Try to acquire the scheduler lock for leader election. Returns true if this instance should run the cron scheduler.