Struct PostgresWorkflowStore 

pub struct PostgresWorkflowStore { /* private fields */ }

PostgreSQL-backed distributed workflow task store.

Claim, heartbeat, and finish operations use the database clock. Every ownership mutation compares worker identity, fencing token, and lease expiration.

Implementations

impl PostgresWorkflowStore

pub async fn connect( connection: &str, table: &str, ) -> Result<Self, PostgresWorkflowStoreError>

Connects without creating or changing schema.

Errors

Rejects unsafe table identifiers and propagates connection failures.

pub async fn ensure_schema(&self) -> Result<(), PostgresWorkflowStoreError>

Explicitly creates the workflow task table and claim index.

Runtime queue operations never perform hidden migrations.

Errors

Propagates PostgreSQL DDL failures.

Trait Implementations

impl Clone for PostgresWorkflowStore

fn clone(&self) -> PostgresWorkflowStore

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for PostgresWorkflowStore

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

Formats the value using the given formatter.

impl WorkflowStore for PostgresWorkflowStore

fn current_time_ms( &self, ) -> WorkflowStoreFuture<'_, Result<u64, WorkflowStoreError>>

Returns the store-authoritative Unix time in milliseconds.

fn set_tenant_policy( &self, tenant_id: WorkflowTenantId, policy: WorkflowTenantPolicy, ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>

Creates or replaces one tenant’s admission policy.

fn set_tenant_budget_policy( &self, tenant_id: WorkflowTenantId, policy: WorkflowTenantBudgetPolicy, ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>

Creates or replaces one tenant’s persistent aggregate budget policy.

fn list_tenant_budgets( &self, after: Option<WorkflowTenantId>, limit: WorkflowTenantListLimit, ) -> WorkflowStoreFuture<'_, Result<Vec<WorkflowTenantId>, WorkflowStoreError>>

Discovers budget-enabled tenants in stable identity order.

fn inspect_tenant_budget( &self, tenant_id: WorkflowTenantId, ) -> WorkflowStoreFuture<'_, Result<WorkflowTenantBudgetSnapshot, WorkflowStoreError>>

Reads a tenant budget after reclaiming expired reservations.

fn list_tenant_budget_audit( &self, tenant_id: WorkflowTenantId, after: Option<WorkflowBudgetAuditCursor>, limit: WorkflowBudgetAuditLimit, ) -> WorkflowStoreFuture<'_, Result<Vec<WorkflowBudgetAuditEvent>, WorkflowStoreError>>

Reads durable budget decisions strictly after an optional cursor.

fn compact_tenant_budget_audit( &self, tenant_id: WorkflowTenantId, through: WorkflowBudgetAuditCursor, ) -> WorkflowStoreFuture<'_, Result<u64, WorkflowStoreError>>

Deletes tenant audit facts at or before an explicitly acknowledged cursor.

fn load_or_create_tenant_budget_audit_projection( &self, tenant_id: WorkflowTenantId, projection_id: WorkflowBudgetAuditProjectionId, ) -> WorkflowStoreFuture<'_, Result<WorkflowBudgetAuditCursor, WorkflowStoreError>>

Loads or atomically registers one named consumer at cursor zero.

fn advance_tenant_budget_audit_projection( &self, tenant_id: WorkflowTenantId, projection_id: WorkflowBudgetAuditProjectionId, expected: WorkflowBudgetAuditCursor, next: WorkflowBudgetAuditCursor, ) -> WorkflowStoreFuture<'_, Result<bool, WorkflowStoreError>>

Monotonically advances a projection cursor using compare-and-set.

fn claim_tenant_budget_audit_projection( &self, tenant_id: WorkflowTenantId, projection_id: WorkflowBudgetAuditProjectionId, owner: WorkerId, lease: LeaseDuration, ) -> WorkflowStoreFuture<'_, Result<Option<WorkflowBudgetAuditProjectionLease>, WorkflowStoreError>>

Exclusively claims an idle or expired named audit projection.

fn heartbeat_tenant_budget_audit_projection( &self, lease: WorkflowBudgetAuditProjectionLease, extension: LeaseDuration, ) -> WorkflowStoreFuture<'_, Result<WorkflowBudgetAuditProjectionLease, WorkflowStoreError>>

Extends an active projection lease under its current fencing token.

fn advance_tenant_budget_audit_projection_lease( &self, lease: WorkflowBudgetAuditProjectionLease, next: WorkflowBudgetAuditCursor, ) -> WorkflowStoreFuture<'_, Result<WorkflowBudgetAuditProjectionLease, WorkflowStoreError>>

Advances a projection cursor only while its fenced lease remains active.

fn release_tenant_budget_audit_projection( &self, lease: WorkflowBudgetAuditProjectionLease, ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>

Releases a currently fenced projection without changing its cursor.

fn reserve_budget( &self, lease: WorkflowLease, workflow_limit: Budget, baseline: Usage, ) -> WorkflowStoreFuture<'_, Result<WorkflowBudgetReservationOutcome, WorkflowStoreError>>

Idempotently reserves the remaining workflow envelope under a lease.

fn settle_budget( &self, lease: WorkflowLease, cumulative: Usage, ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>

Commits observed cumulative usage and releases unused reservation.

fn enqueue( &self, task: WorkflowTask, ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>

Enqueues a task exactly once.

fn claim( &self, worker: WorkerId, lease: LeaseDuration, ) -> WorkflowStoreFuture<'_, Result<Option<ClaimedWorkflow>, WorkflowStoreError>>

Atomically claims the highest-priority eligible task.

fn heartbeat( &self, lease: WorkflowLease, extension: LeaseDuration, ) -> WorkflowStoreFuture<'_, Result<WorkflowLease, WorkflowStoreError>>

Extends a currently owned, unexpired lease.

fn finish( &self, lease: WorkflowLease, disposition: WorkflowDisposition, ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>

Applies a terminal or retry disposition under the current lease.

fn publish_signal( &self, tenant_id: WorkflowTenantId, signal: WorkflowSignal, ) -> WorkflowStoreFuture<'_, Result<WorkflowSignalOutcome, WorkflowStoreError>>

Idempotently publishes an external signal, buffering it when necessary.

fn publish_control_signal( &self, tenant_id: WorkflowTenantId, signal: WorkflowSignal, ) -> WorkflowStoreFuture<'_, Result<WorkflowSignalOutcome, WorkflowStoreError>>

Publishes durable coordination metadata excluded from signal retention.

fn cancel( &self, tenant_id: WorkflowTenantId, checkpoint_id: CheckpointId, ) -> WorkflowStoreFuture<'_, Result<WorkflowCancelOutcome, WorkflowStoreError>>

Idempotently cancels queued, waiting, or currently leased work.

fn inspect_signal( &self, tenant_id: WorkflowTenantId, signal_id: WorkflowSignalId, ) -> WorkflowStoreFuture<'_, Result<WorkflowSignalSnapshot, WorkflowStoreError>>

Loads safe signal lifecycle metadata without exposing its payload.

fn load_signal_payload( &self, tenant_id: WorkflowTenantId, signal_id: WorkflowSignalId, ) -> WorkflowStoreFuture<'_, Result<Value, WorkflowStoreError>>

Loads one accepted signal payload under tenant authorization.

fn compact_signals( &self, tenant_id: WorkflowTenantId, retention: WorkflowSignalRetention, ) -> WorkflowStoreFuture<'_, Result<u64, WorkflowStoreError>>

Deletes only consumed or dead-letter signals older than retention.

fn inspect( &self, tenant_id: WorkflowTenantId, checkpoint_id: CheckpointId, ) -> WorkflowStoreFuture<'_, Result<WorkflowTaskSnapshot, WorkflowStoreError>>

Loads safe control-plane state for inspection.

fn load_task_input( &self, tenant_id: WorkflowTenantId, checkpoint_id: CheckpointId, ) -> WorkflowStoreFuture<'_, Result<Value, WorkflowStoreError>>

Loads the immutable original task input under tenant authorization.

fn list_checkpoint_history( &self, tenant_id: WorkflowTenantId, checkpoint_id: CheckpointId, after_revision: Option<u64>, limit: WorkflowCheckpointHistoryLimit, ) -> WorkflowStoreFuture<'_, Result<Vec<WorkflowCheckpointRevision>, WorkflowStoreError>>

Lists immutable checkpoint revisions after an optional revision cursor.

fn load_checkpoint_revision( &self, tenant_id: WorkflowTenantId, checkpoint_id: CheckpointId, revision: u64, ) -> WorkflowStoreFuture<'_, Result<WorkflowCheckpointRevision, WorkflowStoreError>>

Loads one exact immutable checkpoint revision for state inspection.

fn fork_workflow( &self, tenant_id: WorkflowTenantId, command: WorkflowForkCommand, ) -> WorkflowStoreFuture<'_, Result<WorkflowForkOutcome, WorkflowStoreError>>

Idempotently creates a new execution branch from immutable history.

fn load_checkpoint( &self, lease: WorkflowLease, ) -> WorkflowStoreFuture<'_, Result<Checkpoint, CheckpointError>>

Loads a checkpoint under a current worker lease.

fn compare_and_swap_checkpoint( &self, lease: WorkflowLease, checkpoint: Checkpoint, expected_revision: Option<u64>, ) -> WorkflowStoreFuture<'_, Result<(), CheckpointError>>

Creates or compare-and-swaps a checkpoint under a current worker lease.

fn decide_interrupt( &self, tenant_id: WorkflowTenantId, command: WorkflowInterruptCommand, ) -> Pin<Box<dyn Future<Output = Result<WorkflowInterruptDecisionOutcome, WorkflowStoreError>> + Send + '_>>

Idempotently applies a typed human decision to a durable interrupt.

impl WorkflowTaskRetentionStore for PostgresWorkflowStore

fn list_task_cleanup_tenants( &self, after: Option<WorkflowTenantId>, limit: WorkflowTenantListLimit, ) -> WorkflowStoreFuture<'_, Result<Vec<WorkflowTenantId>, WorkflowStoreError>>

Discovers tenants that currently own terminal Tasks.

fn claim_task_cleanup( &self, tenant_id: WorkflowTenantId, owner: WorkerId, lease: LeaseDuration, ) -> WorkflowStoreFuture<'_, Result<Option<WorkflowTaskCleanupLease>, WorkflowStoreError>>

Claims one tenant’s cleanup partition if it is idle or expired.

fn compact_terminal_tasks( &self, lease: WorkflowTaskCleanupLease, retention: WorkflowTaskRetention, limit: WorkflowTaskCleanupLimit, ) -> WorkflowStoreFuture<'_, Result<Vec<WorkflowTaskTombstone>, WorkflowStoreError>>

Atomically tombstones and removes one bounded terminal Task batch.

fn heartbeat_task_cleanup( &self, lease: WorkflowTaskCleanupLease, extension: LeaseDuration, ) -> WorkflowStoreFuture<'_, Result<WorkflowTaskCleanupLease, WorkflowStoreError>>

Extends an exact current cleanup lease using store-authoritative time.

fn list_task_tombstones( &self, tenant_id: WorkflowTenantId, after: Option<WorkflowTaskTombstoneCursor>, limit: WorkflowTaskTombstoneLimit, ) -> WorkflowStoreFuture<'_, Result<Vec<WorkflowTaskTombstone>, WorkflowStoreError>>

Lists immutable tombstones strictly after an optional cursor.

fn release_task_cleanup( &self, lease: WorkflowTaskCleanupLease, ) -> WorkflowStoreFuture<'_, Result<(), WorkflowStoreError>>

Releases a current unexpired cleanup lease.

impl WorkflowTaskTombstoneGovernanceStore for PostgresWorkflowStore

fn place_task_tombstone_hold( &self, tenant_id: WorkflowTenantId, checkpoint_id: CheckpointId, actor: WorkerId, reason: WorkflowTaskLegalHoldReason, ) -> WorkflowStoreFuture<'_, Result<WorkflowTaskLegalHold, WorkflowStoreError>>

Places or idempotently reads a legal hold on an existing tombstone.

fn release_task_tombstone_hold( &self, tenant_id: WorkflowTenantId, checkpoint_id: CheckpointId, actor: WorkerId, ) -> WorkflowStoreFuture<'_, Result<WorkflowTaskLegalHold, WorkflowStoreError>>

Releases an exact active legal hold while retaining its audit row.

fn confirm_task_tombstone_export( &self, tenant_id: WorkflowTenantId, through: WorkflowTaskTombstoneCursor, receipt: WorkflowTaskTombstoneExportReceipt, actor: WorkerId, ) -> WorkflowStoreFuture<'_, Result<WorkflowTaskTombstoneExport, WorkflowStoreError>>

Monotonically confirms an externally archived tenant cursor prefix.

fn prepare_task_tombstone_purge( &self, lease: WorkflowTaskCleanupLease, retention: WorkflowTaskTombstoneRetention, limit: WorkflowTaskTombstonePurgeLimit, approval_window: WorkflowTaskTombstoneApprovalWindow, ) -> WorkflowStoreFuture<'_, Result<WorkflowTaskTombstonePurgeIntent, WorkflowStoreError>>

Freezes one bounded, exported, unheld, old-enough purge candidate set.

fn approve_task_tombstone_purge( &self, tenant_id: WorkflowTenantId, purge_id: WorkflowTaskTombstonePurgeId, approver: WorkerId, ) -> WorkflowStoreFuture<'_, Result<WorkflowTaskTombstonePurgeIntent, WorkflowStoreError>>

Approves a pending intent using a different principal from its preparer.

fn list_task_tombstone_purge_approvals( &self, tenant_id: WorkflowTenantId, limit: WorkflowTaskTombstoneApprovalInboxLimit, ) -> WorkflowStoreFuture<'_, Result<Vec<WorkflowTaskTombstoneApprovalInboxItem>, WorkflowStoreError>>

Lists a bounded tenant approval inbox with expired claims normalized.

fn claim_task_tombstone_purge_approval( &self, tenant_id: WorkflowTenantId, reviewer: WorkerId, lease: LeaseDuration, ) -> WorkflowStoreFuture<'_, Result<Option<WorkflowTaskTombstoneApprovalLease>, WorkflowStoreError>>

Atomically claims the oldest eligible request for an independent reviewer.

fn approve_claimed_task_tombstone_purge( &self, lease: WorkflowTaskTombstoneApprovalLease, ) -> WorkflowStoreFuture<'_, Result<WorkflowTaskTombstonePurgeIntent, WorkflowStoreError>>

Approves under an exact, unexpired, fenced reviewer lease.

fn reject_claimed_task_tombstone_purge( &self, lease: WorkflowTaskTombstoneApprovalLease, reason: WorkflowTaskTombstoneRejectionReason, ) -> WorkflowStoreFuture<'_, Result<WorkflowTaskTombstoneApprovalInboxItem, WorkflowStoreError>>

Rejects under an exact reviewer lease and preserves the reason.

fn execute_task_tombstone_purge( &self, lease: WorkflowTaskCleanupLease, purge_id: WorkflowTaskTombstonePurgeId, ) -> WorkflowStoreFuture<'_, Result<WorkflowTaskTombstonePurgeEvidence, WorkflowStoreError>>

Executes an approved intent under a current fenced cleanup lease.

fn get_task_tombstone_purge_evidence( &self, tenant_id: WorkflowTenantId, purge_id: WorkflowTaskTombstonePurgeId, ) -> WorkflowStoreFuture<'_, Result<Option<WorkflowTaskTombstonePurgeEvidence>, WorkflowStoreError>>

Reads immutable evidence for one executed purge.