Struct PostgresStore 

pub struct PostgresStore { /* private fields */ }

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 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 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 create_api_key( &self, key_hash: &str, prefix: &str, label: Option<&str>, created_at: f64, ) -> Result<()>

async fn validate_api_key(&self, key_hash: &str) -> Result<bool>

async fn list_api_keys(&self) -> Result<Vec<ApiKeyRecord>>

async fn revoke_api_key(&self, prefix: &str) -> Result<bool>

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

Return true iff the api_keys table has no rows. Used by the HTTP layer to identify the first-ever key-creation window (where the POST /api/v1/api-keys endpoint is callable without authentication).

async fn get_api_key_by_label( &self, label: &str, ) -> Result<Option<ApiKeyRecord>>

Find an existing API key by its label. Returns None if no key has this label (or label is NULL). Used to implement idempotent key creation: a second POST /api/v1/api-keys { label, idempotent: true } call hits this method and returns the existing record’s metadata (without a plaintext, which is only ever retrievable at generation time).

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.

fn subscribe_runnable<'a>( &'a self, namespace: &'a str, ) -> impl Future<Output = BoxStream<'a, String>> + Send + 'a

Emits workflow IDs as they become runnable in the given namespace. The subscription primitive (e.g. PG LISTEN) is registered inside the returned future so the stream is guaranteed active by the time the caller receives it — later pg_notifys cannot race ahead of setup. PG implements this via LISTEN/NOTIFY; SQLite returns an empty stream (the scheduler falls back to its local timer heap on backends without a native push primitive).

fn subscribe_tasks<'a>( &'a self, queue_names: &'a [&'a str], ) -> impl Future<Output = BoxStream<'a, String>> + Send + 'a

Emits workflow task IDs as new tasks arrive on any of the listed queues. Subscription is registered before the stream is handed back (see subscribe_runnable for why). Empty-stream fallback for backends without native push.