Struct PostgresBackend 

pub struct PostgresBackend { /* private fields */ }

Implementations

impl PostgresBackend

pub async fn connect( config: BackendConfig, ) -> Result<Arc<dyn EngineBackend>, EngineError>

Dial Postgres with BackendConfig and return the backend as Arc<dyn EngineBackend>. Modeled on [ff_backend_valkey::ValkeyBackend::connect] so ff-server / SDK call sites can swap backends without changing the constructor shape.

Wave 0: builds the pool and constructs the backend. Does NOT run migrations (Q12 — operator out-of-band). Does NOT run the schema-version check (Wave 3 adds the version const and wires check_schema_version in). Does NOT start the LISTEN task (Wave 4).

Returns EngineError::Unavailable when the config’s connection arm is not Postgres.

pub fn from_pool(pool: PgPool, partition_config: PartitionConfig) -> Arc<Self>

Test / advanced constructor: build a PostgresBackend from an already-constructed PgPool + explicit partition config. No network I/O. Useful for integration tests against a shared pool and for a future migration CLI that wants to reuse a pool across migrate-run + smoke-check.

pub async fn connect_with_metrics( config: BackendConfig, partition_config: PartitionConfig, metrics: Arc<Metrics>, ) -> Result<Arc<Self>, EngineError>

RFC-017 Wave 8 Stage E1: dial Postgres with an explicit PartitionConfig + shared ff_observability::Metrics. Mirrors [ff_backend_valkey::ValkeyBackend::connect_with_metrics] so ff-server::Server::start_with_metrics can wire the Postgres branch without reaching into the pool builder directly.

Returns a concrete Arc<Self> rather than Arc<dyn EngineBackend> so the caller can cast to the trait object after any additional field installs (parallel to the Valkey path which calls with_scheduler / with_stream_semaphore_permits before the cast). Stage E1 does NOT run apply_migrations — schema provisioning is an operator concern (matches the Wave 0 contract on Self::connect).

pub fn with_scanners(self: &mut Arc<Self>, cfg: PostgresScannerConfig) -> bool

RFC-017 Wave 8 Stage E3: spawn the six Postgres reconcilers as background tick loops. Returns true if the scanner handle was installed; false if the Arc<Self> has outstanding clones (mirrors the Valkey with_* pattern). Callers must invoke this before publishing the Arc<dyn EngineBackend> so the underlying Arc::get_mut succeeds.

pub fn pool(&self) -> &PgPool

Accessor for the underlying PgPool. Stage E1 uses this so ff-server::Server::start_with_metrics can run apply_migrations on the same pool before handing the backend out as Arc<dyn EngineBackend>.

pub async fn create_execution( &self, args: CreateExecutionArgs, ) -> Result<ExecutionId, EngineError>

Create one execution row (+ seed the lane registry if new).

RFC-017 Stage A: this inherent method is retained as a thin wrapper around the module-level impl so existing in-tree callers (ff-server request handlers, integration tests) keep compiling. The trait-lifted entry point is EngineBackend::create_execution below, which calls the same impl. Return shape differs — inherent returns ExecutionId, trait returns CreateExecutionResult per RFC-017 §5 — so we cannot simply replace the inherent method. A follow-up PR may deprecate this inherent alongside the broader ingress shape alignment.

pub async fn create_flow( &self, args: &CreateFlowArgs, ) -> Result<CreateFlowResult, EngineError>

pub async fn add_execution_to_flow( &self, args: &AddExecutionToFlowArgs, ) -> Result<AddExecutionToFlowResult, EngineError>

pub async fn stage_dependency_edge( &self, args: &StageDependencyEdgeArgs, ) -> Result<StageDependencyEdgeResult, EngineError>

pub async fn apply_dependency_to_child( &self, args: &ApplyDependencyToChildArgs, ) -> Result<ApplyDependencyToChildResult, EngineError>

Trait Implementations

impl CompletionBackend for PostgresBackend

fn subscribe_completions<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<CompletionStream, EngineError>> + Send + 'async_trait>>

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

Subscribe to the completion event stream.

fn subscribe_completions_filtered<'life0, 'life1, 'async_trait>( &'life0 self, filter: &'life1 ScannerFilter, ) -> Pin<Box<dyn Future<Output = Result<CompletionStream, EngineError>> + Send + 'async_trait>>

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

Subscribe to the completion event stream with a per-event ScannerFilter applied at the backend boundary (issue #122).

impl EngineBackend for PostgresBackend

fn create_execution<'life0, 'async_trait>( &'life0 self, args: CreateExecutionArgs, ) -> Pin<Box<dyn Future<Output = Result<CreateExecutionResult, EngineError>> + Send + 'async_trait>>

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

RFC-017 Wave 8 Stage E1: lift the inherent PostgresBackend::create_execution onto the trait so ff-server’s migrated HTTP handler can dispatch to Postgres. Post-insert the row is idempotent; the Postgres impl does not distinguish Created from Duplicate at the helper level (both paths commit and return the execution id), so we always surface Created { public_state: Waiting } here. A follow-up may lift the distinction if a consumer relies on it.

fn capabilities(&self) -> Capabilities

RFC-018 Stage A: populate the Capabilities snapshot from the static [postgres_supports_base] shape. The Postgres backend landed through RFC-017 Stage E4 at v0.8.0; fields still false correspond to Wave-9 follow-up work (cancel_flow_header, ack_cancel_member, read-model, operator control, budget / quota, list_pending_waitpoints). See docs/POSTGRES_PARITY_MATRIX.md for the per-row breakdown.

fn prepare<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<PrepareOutcome, EngineError>> + Send + 'async_trait>>

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

Issue #281: no-op. Schema migrations are applied out-of-band per rfcs/drafts/v0.7-migration-master.md §Q12 (operator runs sqlx migrate run or the future ff-migrate CLI). Boot runs a schema-version check at connect time (crate::version::check_schema_version) and refuses to start on mismatch, so by the time prepare() is callable there is nothing further to do.

fn claim_for_worker<'life0, 'async_trait>( &'life0 self, args: ClaimForWorkerArgs, ) -> Pin<Box<dyn Future<Output = Result<ClaimForWorkerOutcome, EngineError>> + Send + 'async_trait>>

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

RFC-017 Wave 8 Stage E3 (§4 row 9, §7): forward the claim to the Postgres-native admission pipeline. Returns NoWork when no eligible execution is admissible this scan cycle. Budget breaches surface as NoWork (leaving the row eligible for a retry by another worker); validation-class rejections (malformed partition, unknown kid) surface as typed EngineError variants mapped to the Server’s 400/503 arms.

fn shutdown_prepare<'life0, 'async_trait>( &'life0 self, grace: Duration, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

RFC-017 Wave 8 Stage E3: drain the scanner supervisor’s reconciler tasks up to grace, then close the sqlx pool. Matches the Valkey backend’s shutdown_prepare contract — bounded best-effort drain, never returns an error.

fn claim<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, lane: &'life1 LaneId, capabilities: &'life2 CapabilitySet, policy: ClaimPolicy, ) -> Pin<Box<dyn Future<Output = Result<Option<Handle>, EngineError>> + Send + 'async_trait>>

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

Fresh-work claim. Returns Ok(None) when no work is currently available; Err only on transport or input-validation faults.

fn renew<'life0, 'life1, 'async_trait>( &'life0 self, handle: &'life1 Handle, ) -> Pin<Box<dyn Future<Output = Result<LeaseRenewal, EngineError>> + Send + 'async_trait>>

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

Renew a held lease. Returns the updated expiry + epoch on success; typed State::StaleLease / State::LeaseExpired when the lease has been stolen or timed out.

fn progress<'life0, 'life1, 'async_trait>( &'life0 self, handle: &'life1 Handle, percent: Option<u8>, message: Option<String>, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Numeric-progress heartbeat.

fn append_frame<'life0, 'life1, 'async_trait>( &'life0 self, handle: &'life1 Handle, frame: Frame, ) -> Pin<Box<dyn Future<Output = Result<AppendFrameOutcome, EngineError>> + Send + 'async_trait>>

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

Append one stream frame. Distinct from progress per RFC-012 §3.1.1 K#6. Returns the backend-assigned stream entry id and post-append frame count (RFC-012 §R7.2.1).

fn complete<'life0, 'life1, 'async_trait>( &'life0 self, handle: &'life1 Handle, payload: Option<Vec<u8>>, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Terminal success. Borrows handle (round-4 M-D2) so callers can retry under EngineError::Transport without losing the cookie. Payload is Option<Vec<u8>> per the note above.

fn fail<'life0, 'life1, 'async_trait>( &'life0 self, handle: &'life1 Handle, reason: FailureReason, classification: FailureClass, ) -> Pin<Box<dyn Future<Output = Result<FailOutcome, EngineError>> + Send + 'async_trait>>

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

Terminal failure with classification. Returns FailOutcome so the caller learns whether a retry was scheduled.

fn cancel<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, handle: &'life1 Handle, reason: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Cooperative cancel by the worker holding the lease.

fn suspend<'life0, 'life1, 'async_trait>( &'life0 self, handle: &'life1 Handle, args: SuspendArgs, ) -> Pin<Box<dyn Future<Output = Result<SuspendOutcome, EngineError>> + Send + 'async_trait>>

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

Suspend the execution awaiting a typed resume condition (RFC-013 Stage 1d).

fn suspend_by_triple<'life0, 'async_trait>( &'life0 self, exec_id: ExecutionId, triple: LeaseFence, args: SuspendArgs, ) -> Pin<Box<dyn Future<Output = Result<SuspendOutcome, EngineError>> + Send + 'async_trait>>

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

Suspend by execution id + lease fence triple, for service-layer callers that hold a run record / lease-claim descriptor but no worker Handle (cairn issue #322).

fn create_waitpoint<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, handle: &'life1 Handle, waitpoint_key: &'life2 str, expires_in: Duration, ) -> Pin<Box<dyn Future<Output = Result<PendingWaitpoint, EngineError>> + Send + 'async_trait>>

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

Issue a pending waitpoint for future signal delivery.

fn read_waitpoint_token<'life0, 'life1, 'async_trait>( &'life0 self, partition: PartitionKey, waitpoint_id: &'life1 WaitpointId, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, EngineError>> + Send + 'async_trait>>

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

Read the HMAC token stored on a waitpoint record, keyed by (partition, waitpoint_id).

fn observe_signals<'life0, 'life1, 'async_trait>( &'life0 self, handle: &'life1 Handle, ) -> Pin<Box<dyn Future<Output = Result<Vec<ResumeSignal>, EngineError>> + Send + 'async_trait>>

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

Non-mutating observation of signals that satisfied the handle’s resume condition.

fn claim_from_resume_grant<'life0, 'async_trait>( &'life0 self, token: ResumeToken, ) -> Pin<Box<dyn Future<Output = Result<Option<Handle>, EngineError>> + Send + 'async_trait>>

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

Consume a resume grant (via ResumeToken) to mint a resumed-kind handle. Routes to ff_claim_resumed_execution on Valkey / the epoch-bump reconciler on PG/SQLite. Returns Ok(None) when the grant’s target execution is no longer resumable (already reclaimed, terminal, etc.).

fn issue_reclaim_grant<'life0, 'async_trait>( &'life0 self, args: IssueReclaimGrantArgs, ) -> Pin<Box<dyn Future<Output = Result<IssueReclaimGrantOutcome, EngineError>> + Send + 'async_trait>>

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

Issue a lease-reclaim grant (RFC-024 §3.2). Admits executions in lease_expired_reclaimable or lease_revoked state to the reclaim path; the returned IssueReclaimGrantOutcome::Granted carries a crate::contracts::ReclaimGrant which is then fed to Self::reclaim_execution to mint a fresh attempt.

fn reclaim_execution<'life0, 'async_trait>( &'life0 self, args: ReclaimExecutionArgs, ) -> Pin<Box<dyn Future<Output = Result<ReclaimExecutionOutcome, EngineError>> + Send + 'async_trait>>

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

Consume a crate::contracts::ReclaimGrant to mint a fresh attempt for a previously lease-expired / lease-revoked execution (RFC-024 §3.2). Creates a new attempt row, bumps the execution’s lease_reclaim_count, and mints a crate::backend::HandleKind::Reclaimed handle.

fn delay<'life0, 'life1, 'async_trait>( &'life0 self, handle: &'life1 Handle, delay_until: TimestampMs, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Park the execution until delay_until, releasing the lease.

fn wait_children<'life0, 'life1, 'async_trait>( &'life0 self, handle: &'life1 Handle, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Mark the execution as waiting for its child flow to complete, releasing the lease.

fn describe_execution<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 ExecutionId, ) -> Pin<Box<dyn Future<Output = Result<Option<ExecutionSnapshot>, EngineError>> + Send + 'async_trait>>

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

Snapshot an execution by id. Ok(None) ⇒ no such execution.

fn read_execution_context<'life0, 'life1, 'async_trait>( &'life0 self, execution_id: &'life1 ExecutionId, ) -> Pin<Box<dyn Future<Output = Result<ExecutionContext, EngineError>> + Send + 'async_trait>>

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

Point-read of the execution-scoped (input_payload, execution_kind, tags) bundle used by the SDK worker when assembling a ClaimedTask (see ff_sdk::ClaimedTask) after a successful claim.

fn read_current_attempt_index<'life0, 'life1, 'async_trait>( &'life0 self, execution_id: &'life1 ExecutionId, ) -> Pin<Box<dyn Future<Output = Result<AttemptIndex, EngineError>> + Send + 'async_trait>>

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

Point-read of the execution’s current attempt-index pointer — the index of the currently-leased attempt row.

fn read_total_attempt_count<'life0, 'life1, 'async_trait>( &'life0 self, execution_id: &'life1 ExecutionId, ) -> Pin<Box<dyn Future<Output = Result<AttemptIndex, EngineError>> + Send + 'async_trait>>

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

Point-read of the execution’s total attempt counter — the monotonic count of claims that have ever fired against this execution (including the in-flight one once claimed).

fn describe_flow<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 FlowId, ) -> Pin<Box<dyn Future<Output = Result<Option<FlowSnapshot>, EngineError>> + Send + 'async_trait>>

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

Snapshot a flow by id. Ok(None) ⇒ no such flow.

fn set_execution_tag<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, execution_id: &'life1 ExecutionId, key: &'life2 str, value: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Set a single namespaced tag on an execution. Tag key MUST match the reserved caller-namespace pattern ^[a-z][a-z0-9_]*\.[a-z0-9_][a-z0-9_.]*$ — i.e. <caller>.<field> — or the call returns EngineError::Validation { kind: ValidationKind::InvalidInput, .. } with the offending key in detail. value is arbitrary UTF-8.

fn set_flow_tag<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, flow_id: &'life1 FlowId, key: &'life2 str, value: &'life3 str, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Set a single namespaced tag on a flow. Same namespace rule as Self::set_execution_tag: key MUST match ^[a-z][a-z0-9_]*\.[a-z0-9_][a-z0-9_.]*$.

fn get_execution_tag<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, execution_id: &'life1 ExecutionId, key: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, EngineError>> + Send + 'async_trait>>

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

Read a single namespaced execution tag. Returns Ok(None) when the tag is absent or the execution row does not exist — the two cases are not distinguished on the read path. Callers that need to distinguish should call Self::describe_execution first (an Ok(None) from that method proves the execution is absent). This matches Valkey’s native HGET semantics and keeps the read path at a single round-trip on every backend.

fn get_flow_tag<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, flow_id: &'life1 FlowId, key: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, EngineError>> + Send + 'async_trait>>

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

Read a single namespaced flow tag. Returns Ok(None) when the tag is absent or the flow row does not exist (same collapse semantics as Self::get_execution_tag). Symmetry partner — consumers like cairn read cairn.session_id off flows for archival.

fn get_execution_namespace<'life0, 'life1, 'async_trait>( &'life0 self, execution_id: &'life1 ExecutionId, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, EngineError>> + Send + 'async_trait>>

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

Read an execution’s namespace scalar. Returns Ok(None) when the row is absent or the field is unset. Dedicated point-read used by the scanner per-candidate filter (should_skip_candidate) to preserve the 1-HGET cost contract documented in ff_engine::scanner::should_skip_candidate — describe_execution is heavier (HGETALL / full snapshot) and unnecessary when only the namespace scalar is needed.

fn list_edges<'life0, 'life1, 'async_trait>( &'life0 self, flow_id: &'life1 FlowId, direction: EdgeDirection, ) -> Pin<Box<dyn Future<Output = Result<Vec<EdgeSnapshot>, EngineError>> + Send + 'async_trait>>

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

List dependency edges adjacent to an execution. Read-only; the backend resolves the subject execution’s flow, reads the direction-specific adjacency SET, and decodes each member’s flow-scoped edge:<edge_id> hash.

fn describe_edge<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, flow_id: &'life1 FlowId, edge_id: &'life2 EdgeId, ) -> Pin<Box<dyn Future<Output = Result<Option<EdgeSnapshot>, EngineError>> + Send + 'async_trait>>

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

Snapshot a single dependency edge by its owning flow + edge id.

fn resolve_execution_flow_id<'life0, 'life1, 'async_trait>( &'life0 self, eid: &'life1 ExecutionId, ) -> Pin<Box<dyn Future<Output = Result<Option<FlowId>, EngineError>> + Send + 'async_trait>>

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

Resolve an execution’s owning flow id, if any.

fn list_flows<'life0, 'async_trait>( &'life0 self, partition: PartitionKey, cursor: Option<FlowId>, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<ListFlowsPage, EngineError>> + Send + 'async_trait>>

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

List flows on a partition with cursor-based pagination (issue #185).

fn list_lanes<'life0, 'async_trait>( &'life0 self, cursor: Option<LaneId>, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<ListLanesPage, EngineError>> + Send + 'async_trait>>

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

Enumerate registered lanes with cursor-based pagination.

fn list_suspended<'life0, 'async_trait>( &'life0 self, partition: PartitionKey, cursor: Option<ExecutionId>, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<ListSuspendedPage, EngineError>> + Send + 'async_trait>>

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

List suspended executions in one partition, cursor-paginated, with each entry’s suspension reason_code populated (issue #183).

fn list_executions<'life0, 'async_trait>( &'life0 self, partition: PartitionKey, cursor: Option<ExecutionId>, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<ListExecutionsPage, EngineError>> + Send + 'async_trait>>

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

Forward-only paginated listing of the executions indexed under one partition.

fn deliver_signal<'life0, 'async_trait>( &'life0 self, args: DeliverSignalArgs, ) -> Pin<Box<dyn Future<Output = Result<DeliverSignalResult, EngineError>> + Send + 'async_trait>>

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

Deliver an external signal to a suspended execution’s waitpoint.

fn deliver_approval_signal<'life0, 'async_trait>( &'life0 self, args: DeliverApprovalSignalArgs, ) -> Pin<Box<dyn Future<Output = Result<DeliverSignalResult, EngineError>> + Send + 'async_trait>>

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

Operator-driven approval-signal delivery.

fn claim_resumed_execution<'life0, 'async_trait>( &'life0 self, args: ClaimResumedExecutionArgs, ) -> Pin<Box<dyn Future<Output = Result<ClaimResumedExecutionResult, EngineError>> + Send + 'async_trait>>

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

Claim a resumed execution — a previously-suspended attempt that has cleared its resume condition (e.g. via Self::deliver_signal) and now needs a worker to pick up the same attempt index.

fn read_execution_state<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 ExecutionId, ) -> Pin<Box<dyn Future<Output = Result<Option<PublicState>, EngineError>> + Send + 'async_trait>>

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

RFC-017 Stage E2: narrow public_state read used by the GET /v1/executions/{id}/state HTTP route. Returns Ok(None) when the execution is missing. Default Unavailable.

fn read_execution_info<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 ExecutionId, ) -> Pin<Box<dyn Future<Output = Result<Option<ExecutionInfo>, EngineError>> + Send + 'async_trait>>

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

RFC-017 Stage E2: full-shape execution read used by the GET /v1/executions/{id} HTTP route. Returns the legacy ExecutionInfo wire shape (not the decoupled ExecutionSnapshot) so the existing HTTP response bytes stay identical across the migration.

fn get_execution_result<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 ExecutionId, ) -> Pin<Box<dyn Future<Output = Result<Option<Vec<u8>>, EngineError>> + Send + 'async_trait>>

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

Fetch the stored result payload for a completed execution (row 4). Returns Ok(None) when the execution is missing, not yet complete, or its payload was trimmed by retention policy.

fn list_pending_waitpoints<'life0, 'async_trait>( &'life0 self, args: ListPendingWaitpointsArgs, ) -> Pin<Box<dyn Future<Output = Result<ListPendingWaitpointsResult, EngineError>> + Send + 'async_trait>>

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

List the pending-or-active waitpoints for an execution, cursor paginated (row 5 / §8). Stage A preserves the existing PendingWaitpointInfo shape; Stage D ships the §8 HMAC sanitisation + (token_kid, token_fingerprint) schema.

fn cancel_execution<'life0, 'async_trait>( &'life0 self, args: CancelExecutionArgs, ) -> Pin<Box<dyn Future<Output = Result<CancelExecutionResult, EngineError>> + Send + 'async_trait>>

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

Operator-initiated execution cancel (row 2).

fn revoke_lease<'life0, 'async_trait>( &'life0 self, args: RevokeLeaseArgs, ) -> Pin<Box<dyn Future<Output = Result<RevokeLeaseResult, EngineError>> + Send + 'async_trait>>

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

Operator-initiated lease revoke (row 19).

fn change_priority<'life0, 'async_trait>( &'life0 self, args: ChangePriorityArgs, ) -> Pin<Box<dyn Future<Output = Result<ChangePriorityResult, EngineError>> + Send + 'async_trait>>

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

Re-score an execution’s eligibility priority (row 17).

fn replay_execution<'life0, 'async_trait>( &'life0 self, args: ReplayExecutionArgs, ) -> Pin<Box<dyn Future<Output = Result<ReplayExecutionResult, EngineError>> + Send + 'async_trait>>

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

Replay a terminal execution (row 22). Variadic KEYS handling (inbound-edge pre-read) is hidden inside the Valkey impl per RFC-017 §4 row 3.

fn cancel_flow_header<'life0, 'async_trait>( &'life0 self, args: CancelFlowArgs, ) -> Pin<Box<dyn Future<Output = Result<CancelFlowHeader, EngineError>> + Send + 'async_trait>>

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

RFC-017 Stage E2: the “header” portion of cancel_flow — run the atomic flow-state flip (Valkey: ff_cancel_flow FCALL; Postgres: cancel_flow_once tx), decode policy + membership, and surface the flow_already_terminal idempotency branch as a first-class [CancelFlowHeader::AlreadyTerminal] so the Server can build the wire CancelFlowResult without reaching for a raw Client. Separate from the existing EngineBackend::cancel_flow entry point (which takes the enum-typed (policy, wait) split and returns the wait-collapsed CancelFlowResult) because the Server owns its own wait-dispatch + member-cancel machinery via EngineBackend::cancel_execution + backlog ack.

fn ack_cancel_member<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, flow_id: &'life1 FlowId, execution_id: &'life2 ExecutionId, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

RFC-017 Stage E2: best-effort acknowledgement that one member of a cancel_all flow has completed its per-member cancel. Drains the member from the flow’s pending_cancels set and, if empty, removes the flow from the partition-level cancel_backlog (Valkey: ff_ack_cancel_member FCALL; Postgres: table write — default Unavailable until Wave 9).

fn create_flow<'life0, 'async_trait>( &'life0 self, args: CreateFlowArgs, ) -> Pin<Box<dyn Future<Output = Result<CreateFlowResult, EngineError>> + Send + 'async_trait>>

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

Create a flow header. Ingress row 5.

fn add_execution_to_flow<'life0, 'async_trait>( &'life0 self, args: AddExecutionToFlowArgs, ) -> Pin<Box<dyn Future<Output = Result<AddExecutionToFlowResult, EngineError>> + Send + 'async_trait>>

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

Atomically add an execution to a flow (single-FCALL co-located commit on Valkey; single-transaction UPSERT on Postgres).

fn stage_dependency_edge<'life0, 'async_trait>( &'life0 self, args: StageDependencyEdgeArgs, ) -> Pin<Box<dyn Future<Output = Result<StageDependencyEdgeResult, EngineError>> + Send + 'async_trait>>

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

Stage a dependency edge between flow members. CAS-guarded on graph_revision — stale rev returns Contention(StaleGraphRevision).

fn apply_dependency_to_child<'life0, 'async_trait>( &'life0 self, args: ApplyDependencyToChildArgs, ) -> Pin<Box<dyn Future<Output = Result<ApplyDependencyToChildResult, EngineError>> + Send + 'async_trait>>

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

Apply a staged dependency edge to its downstream child.

fn cascade_completion<'life0, 'life1, 'async_trait>( &'life0 self, payload: &'life1 CompletionPayload, ) -> Pin<Box<dyn Future<Output = Result<CascadeOutcome, EngineError>> + Send + 'async_trait>>

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

Cascade a terminal-execution completion into its downstream edges. Consumed by ff-engine::completion_listener::spawn_dispatch_loop (PR-7b Cluster 4) to trait-route the post-completion DAG-promotion path through Arc<dyn EngineBackend>.

fn backend_label(&self) -> &'static str

Static observability label identifying the backend family in logs + metrics (RFC-017 §5.4 + §9 Stage B). Default impl returns "unknown" so legacy impl EngineBackend blocks that have not upgraded keep compiling; every in-tree backend overrides — ValkeyBackend → "valkey", PostgresBackend → "postgres".

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

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

Backend-level reachability probe (row 1). Valkey: PING; Postgres: SELECT 1.

fn cancel_flow<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 FlowId, policy: CancelFlowPolicy, wait: CancelFlowWait, ) -> Pin<Box<dyn Future<Output = Result<CancelFlowResult, EngineError>> + Send + 'async_trait>>

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

Operator-initiated cancellation of a flow and (optionally) its member executions. See RFC-012 §3.1.1 for the policy /wait matrix.

fn set_edge_group_policy<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, flow_id: &'life1 FlowId, downstream_execution_id: &'life2 ExecutionId, policy: EdgeDependencyPolicy, ) -> Pin<Box<dyn Future<Output = Result<SetEdgeGroupPolicyResult, EngineError>> + Send + 'async_trait>>

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

RFC-016 Stage A: set the inbound-edge-group policy for a downstream execution. Must be called before the first add_dependency(... -> downstream_execution_id) — the backend rejects with EngineError::Conflict if edges have already been staged for this group.

fn report_usage<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, _handle: &'life1 Handle, budget: &'life2 BudgetId, dimensions: UsageDimensions, ) -> Pin<Box<dyn Future<Output = Result<ReportUsageResult, EngineError>> + Send + 'async_trait>>

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

Report usage against a budget and check limits. Returns the typed ReportUsageResult variant; backends enforce idempotency via the caller-supplied [UsageDimensions::dedup_key] (RFC-012 §R7.2.3 — replaces the pre-Round-7 AdmissionDecision return).

fn create_budget<'life0, 'async_trait>( &'life0 self, args: CreateBudgetArgs, ) -> Pin<Box<dyn Future<Output = Result<CreateBudgetResult, EngineError>> + Send + 'async_trait>>

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

Create a budget definition (row 6).

fn reset_budget<'life0, 'async_trait>( &'life0 self, args: ResetBudgetArgs, ) -> Pin<Box<dyn Future<Output = Result<ResetBudgetResult, EngineError>> + Send + 'async_trait>>

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

Reset a budget’s usage counters (row 10).

fn create_quota_policy<'life0, 'async_trait>( &'life0 self, args: CreateQuotaPolicyArgs, ) -> Pin<Box<dyn Future<Output = Result<CreateQuotaPolicyResult, EngineError>> + Send + 'async_trait>>

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

Create a quota policy (row 7).

fn get_budget_status<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 BudgetId, ) -> Pin<Box<dyn Future<Output = Result<BudgetStatus, EngineError>> + Send + 'async_trait>>

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

Read-only budget status for operator visibility (row 8).

fn report_usage_admin<'life0, 'life1, 'async_trait>( &'life0 self, budget_id: &'life1 BudgetId, args: ReportUsageAdminArgs, ) -> Pin<Box<dyn Future<Output = Result<ReportUsageResult, EngineError>> + Send + 'async_trait>>

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

Admin-path report_usage (row 9 + RFC-017 §5 round-1 F4). Distinct from the existing Self::report_usage which takes a worker handle — the admin path has no lease context.

fn record_spend<'life0, 'async_trait>( &'life0 self, args: RecordSpendArgs, ) -> Pin<Box<dyn Future<Output = Result<ReportUsageResult, EngineError>> + Send + 'async_trait>>

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

Per-execution budget spend with tenant-open dimensions.

fn release_budget<'life0, 'async_trait>( &'life0 self, args: ReleaseBudgetArgs, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Per-execution budget attribution release.

fn release_admission<'life0, 'async_trait>( &'life0 self, args: ReleaseAdmissionArgs, ) -> Pin<Box<dyn Future<Output = Result<ReleaseAdmissionResult, EngineError>> + Send + 'async_trait>>

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

Release a quota-admission slot that was recorded via Self::check_admission / ff_check_admission_and_record but for which issue_claim_grant subsequently failed. Idempotent: releasing an already-released slot is a no-op.

fn read_quota_policy_limits<'life0, 'life1, 'async_trait>( &'life0 self, quota_policy_id: &'life1 QuotaPolicyId, ) -> Pin<Box<dyn Future<Output = Result<Option<QuotaPolicyLimits>, EngineError>> + Send + 'async_trait>>

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

Read the admission-relevant fields of a quota policy (rate limit, window, concurrency cap, jitter). Replaces the Valkey-shaped 4-HGET pattern on ff:quota:{K}:def that ff_scheduler used pre-FF #511. Returns None when the policy row is absent; absence is a well-defined “no admission configured” signal, not an error.

fn rotate_waitpoint_hmac_secret_all<'life0, 'async_trait>( &'life0 self, args: RotateWaitpointHmacSecretAllArgs, ) -> Pin<Box<dyn Future<Output = Result<RotateWaitpointHmacSecretAllResult, EngineError>> + Send + 'async_trait>>

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

Rotate the waitpoint HMAC signing kid cluster-wide.

fn seed_waitpoint_hmac_secret<'life0, 'async_trait>( &'life0 self, args: SeedWaitpointHmacSecretArgs, ) -> Pin<Box<dyn Future<Output = Result<SeedOutcome, EngineError>> + Send + 'async_trait>>

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

Seed the initial waitpoint HMAC secret for a fresh deployment (issue #280).

fn read_stream<'life0, 'life1, 'async_trait>( &'life0 self, execution_id: &'life1 ExecutionId, attempt_index: AttemptIndex, from: StreamCursor, to: StreamCursor, count_limit: u64, ) -> Pin<Box<dyn Future<Output = Result<StreamFrames, EngineError>> + Send + 'async_trait>>

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

Read frames from a completed or in-flight attempt’s stream.

fn tail_stream<'life0, 'life1, 'async_trait>( &'life0 self, execution_id: &'life1 ExecutionId, attempt_index: AttemptIndex, after: StreamCursor, block_ms: u64, count_limit: u64, visibility: TailVisibility, ) -> Pin<Box<dyn Future<Output = Result<StreamFrames, EngineError>> + Send + 'async_trait>>

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

Tail a live attempt’s stream.

fn read_summary<'life0, 'life1, 'async_trait>( &'life0 self, execution_id: &'life1 ExecutionId, attempt_index: AttemptIndex, ) -> Pin<Box<dyn Future<Output = Result<Option<SummaryDocument>, EngineError>> + Send + 'async_trait>>

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

Read the rolling summary document for an attempt (RFC-015 §6.3).

fn subscribe_completion<'life0, 'life1, 'async_trait>( &'life0 self, _cursor: StreamCursor, filter: &'life1 ScannerFilter, ) -> Pin<Box<dyn Future<Output = Result<CompletionSubscription, EngineError>> + Send + 'async_trait>>

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

Subscribe to completion events (terminal state transitions).

fn subscribe_lease_history<'life0, 'life1, 'async_trait>( &'life0 self, cursor: StreamCursor, filter: &'life1 ScannerFilter, ) -> Pin<Box<dyn Future<Output = Result<LeaseHistorySubscription, EngineError>> + Send + 'async_trait>>

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

Subscribe to lease lifecycle events (acquired / renewed / expired / reclaimed / revoked) for the partition this backend is configured with.

fn subscribe_signal_delivery<'life0, 'life1, 'async_trait>( &'life0 self, cursor: StreamCursor, filter: &'life1 ScannerFilter, ) -> Pin<Box<dyn Future<Output = Result<SignalDeliverySubscription, EngineError>> + Send + 'async_trait>>

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

Subscribe to signal-delivery events (satisfied / buffered / deduped).

fn mark_lease_expired_if_due<'life0, 'life1, 'async_trait>( &'life0 self, partition: Partition, execution_id: &'life1 ExecutionId, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Lease-expiry scanner hook — mark an expired lease as reclaimable so another worker can redeem the execution. Atomic per call.

fn promote_delayed<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, partition: Partition, _lane: &'life1 LaneId, execution_id: &'life2 ExecutionId, now_ms: TimestampMs, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Delayed-promoter scanner hook — promote a delayed execution to eligible_now once its delay_until has passed.

fn close_waitpoint<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, partition: Partition, _execution_id: &'life1 ExecutionId, waitpoint_id: &'life2 str, now_ms: TimestampMs, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Pending-waitpoint-expiry scanner hook — close a pending waitpoint whose deadline has passed (wake the suspended execution with a timeout signal).

fn expire_execution<'life0, 'life1, 'async_trait>( &'life0 self, partition: Partition, execution_id: &'life1 ExecutionId, phase: ExpirePhase, now_ms: TimestampMs, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Shared hook for the attempt-timeout and execution-deadline scanners — terminate or retry an execution whose wall-clock budget has elapsed. phase discriminates which of the two scanner paths is calling so the backend can preserve diagnostic breadcrumbs without forking the surface.

fn expire_suspension<'life0, 'life1, 'async_trait>( &'life0 self, partition: Partition, execution_id: &'life1 ExecutionId, _now_ms: TimestampMs, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Suspension-timeout scanner hook — expire a suspended execution whose suspension deadline has passed (wake with timeout).

fn project_flow_summary<'life0, 'life1, 'async_trait>( &'life0 self, partition: Partition, flow_id: &'life1 FlowId, now_ms: TimestampMs, ) -> Pin<Box<dyn Future<Output = Result<bool, EngineError>> + Send + 'async_trait>>

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

Flow-projector scanner hook — sample flow members, derive an aggregate public_flow_state + per-state counts, and write them to the flow summary projection. Returns Ok(true) when the summary was updated, Ok(false) when the flow had no members or the index entry was defensively pruned (core missing).

fn trim_retention<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, partition: Partition, lane_id: &'life1 LaneId, retention_ms: u64, now_ms: TimestampMs, batch_size: u32, filter: &'life2 ScannerFilter, ) -> Pin<Box<dyn Future<Output = Result<u32, EngineError>> + Send + 'async_trait>>

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

Retention-trimmer scanner hook — delete terminal executions and all their subordinate keys/rows once they are older than the configured retention window. Returns the number of executions actually purged in this call (so the scanner can loop when it hits batch_size).

fn renew_lease<'life0, 'async_trait>( &'life0 self, args: RenewLeaseArgs, ) -> Pin<Box<dyn Future<Output = Result<RenewLeaseResult, EngineError>> + Send + 'async_trait>>

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

Service-layer renew_lease — peer of Self::renew that takes a fence triple instead of a worker Handle.

fn complete_execution<'life0, 'async_trait>( &'life0 self, args: CompleteExecutionArgs, ) -> Pin<Box<dyn Future<Output = Result<CompleteExecutionResult, EngineError>> + Send + 'async_trait>>

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

Service-layer complete_execution — peer of Self::complete that takes a fence triple instead of a worker Handle. See the group preamble above for cairn-migration context.

fn fail_execution<'life0, 'async_trait>( &'life0 self, args: FailExecutionArgs, ) -> Pin<Box<dyn Future<Output = Result<FailExecutionResult, EngineError>> + Send + 'async_trait>>

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

Service-layer fail_execution — peer of Self::fail that takes a fence triple instead of a worker Handle.

fn resume_execution<'life0, 'async_trait>( &'life0 self, args: ResumeExecutionArgs, ) -> Pin<Box<dyn Future<Output = Result<ResumeExecutionResult, EngineError>> + Send + 'async_trait>>

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

Service-layer resume_execution — transitions a suspended execution back to runnable. Distinct from Self::claim_from_resume_grant (which mints a worker handle against an already-eligible resumed execution): this method is the lifecycle transition primitive the control plane calls when an operator / auto-resume policy moves a suspended execution forward.

fn check_admission<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, quota_policy_id: &'life1 QuotaPolicyId, _dimension: &'life2 str, args: CheckAdmissionArgs, ) -> Pin<Box<dyn Future<Output = Result<CheckAdmissionResult, EngineError>> + Send + 'async_trait>>

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

Service-layer check_admission_and_record — atomic admission check against a quota policy. Callers supply the policy id + dimension (quota keys live on their own {q:<policy>} partition that cannot be derived from execution_id, so these travel outside CheckAdmissionArgs). dimension defaults to "default" inside the Valkey body when the caller passes an empty string — matches cairn’s pre-migration default.

fn evaluate_flow_eligibility<'life0, 'async_trait>( &'life0 self, args: EvaluateFlowEligibilityArgs, ) -> Pin<Box<dyn Future<Output = Result<EvaluateFlowEligibilityResult, EngineError>> + Send + 'async_trait>>

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

Service-layer evaluate_flow_eligibility — read-only check that returns the execution’s current eligibility state (eligible, blocked_by_dependencies, or a backend-specific status string). Called by cairn’s dependency-resolution path to decide whether a downstream execution can proceed.

fn claim_execution<'life0, 'async_trait>( &'life0 self, args: ClaimExecutionArgs, ) -> Pin<Box<dyn Future<Output = Result<ClaimExecutionResult, EngineError>> + Send + 'async_trait>>

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

Consume a scheduler-issued claim grant to mint a fresh attempt.

fn issue_grant_and_claim<'life0, 'async_trait>( &'life0 self, args: IssueGrantAndClaimArgs, ) -> Pin<Box<dyn Future<Output = Result<ClaimGrantOutcome, EngineError>> + Send + 'async_trait>>

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

Backend-atomic issue_claim_grant + claim_execution.

fn read_exec_core_fields<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, partition: Partition, execution_id: &'life1 ExecutionId, fields: &'life2 [&'life3 str], ) -> Pin<Box<dyn Future<Output = Result<HashMap<String, Option<String>>, EngineError>> + Send + 'async_trait>>

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

Point-read N fields from the exec_core hash for a given execution. Returns a map of field-name → Option (None for fields absent or stored as NULL). Scanner call sites formerly issuing raw HGET/HMGET on ExecKeyContext::core() route through this trait method (cairn #436 / PR-7b Wave 0a).

fn server_time_ms<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<u64, EngineError>> + Send + 'async_trait>>

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

Returns the backend’s current wall-clock epoch milliseconds.

fn register_worker<'life0, 'async_trait>( &'life0 self, args: RegisterWorkerArgs, ) -> Pin<Box<dyn Future<Output = Result<RegisterWorkerOutcome, EngineError>> + Send + 'async_trait>>

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

Register (or idempotently refresh) a worker instance. See RFC-025 §4.

fn heartbeat_worker<'life0, 'async_trait>( &'life0 self, args: HeartbeatWorkerArgs, ) -> Pin<Box<dyn Future<Output = Result<HeartbeatWorkerOutcome, EngineError>> + Send + 'async_trait>>

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

Refresh the worker-instance liveness TTL.

fn mark_worker_dead<'life0, 'async_trait>( &'life0 self, args: MarkWorkerDeadArgs, ) -> Pin<Box<dyn Future<Output = Result<MarkWorkerDeadOutcome, EngineError>> + Send + 'async_trait>>

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

Operator-driven worker death (distinct from passive TTL expiry).

fn list_expired_leases<'life0, 'async_trait>( &'life0 self, args: ListExpiredLeasesArgs, ) -> Pin<Box<dyn Future<Output = Result<ListExpiredLeasesResult, EngineError>> + Send + 'async_trait>>

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

Enumerate expired leases for reclaim-decision tooling.

fn list_workers<'life0, 'async_trait>( &'life0 self, args: ListWorkersArgs, ) -> Pin<Box<dyn Future<Output = Result<ListWorkersResult, EngineError>> + Send + 'async_trait>>

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

Enumerate live workers (RFC-025 Phase 6, §9.4).

fn scan_eligible_executions<'life0, 'async_trait>( &'life0 self, _args: ScanEligibleArgs, ) -> Pin<Box<dyn Future<Output = Result<Vec<ExecutionId>, EngineError>> + Send + 'async_trait>>

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

Scan a lane’s eligible ZSET on one partition for highest-priority executions awaiting a worker (v0.12 PR-5).

fn issue_claim_grant<'life0, 'async_trait>( &'life0 self, _args: IssueClaimGrantArgs, ) -> Pin<Box<dyn Future<Output = Result<IssueClaimGrantOutcome, EngineError>> + Send + 'async_trait>>

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

Issue a claim grant — the scheduler’s admission write — for a single execution on a single lane (v0.12 PR-5).

fn block_route<'life0, 'async_trait>( &'life0 self, _args: BlockRouteArgs, ) -> Pin<Box<dyn Future<Output = Result<BlockRouteOutcome, EngineError>> + Send + 'async_trait>>

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

Move an execution from a lane’s eligible ZSET into its blocked_route ZSET (v0.12 PR-5).

fn resolve_dependency<'life0, 'async_trait>( &'life0 self, _args: ResolveDependencyArgs, ) -> Pin<Box<dyn Future<Output = Result<ResolveDependencyOutcome, EngineError>> + Send + 'async_trait>>

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

Resolve one dependency edge after its upstream reached a terminal outcome — satisfy on “success”, mark impossible otherwise. Idempotent (AlreadyResolved on replay).

fn block_execution_for_admission<'life0, 'async_trait>( &'life0 self, _args: BlockExecutionForAdmissionArgs, ) -> Pin<Box<dyn Future<Output = Result<BlockExecutionForAdmissionOutcome, EngineError>> + Send + 'async_trait>>

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

Generalised admission block — covers budget / quota / capability denial paths. BlockingReason selects both the eligibility state written to exec_core and the target blocked_<reason> lane index. Valkey wraps the existing ff_block_execution_for_admission FCALL (KEYS=3); PG/SQLite write the equivalent row transition.

fn read_budget_usage_and_limits<'life0, 'life1, 'async_trait>( &'life0 self, _budget_id: &'life1 BudgetId, ) -> Pin<Box<dyn Future<Output = Result<BudgetUsageAndLimits, EngineError>> + Send + 'async_trait>>

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

FF #511 Phase 3 — typed snapshot of a budget’s usage + limits hashes. Replaces the scheduler’s Valkey-shaped HGETALL/HGET pattern on ff:budget:{K}:{id}:limits + ff:budget:{K}:{id}:usage. Returns BudgetUsageAndLimits::empty when the limits hash is absent (“no limits configured” — not an error).

fn as_any(&self) -> &(dyn Any + 'static)

Backend downcast escape hatch (v0.12 PR-7a transitional).

fn subscribe_instance_tags<'life0, 'async_trait>( &'life0 self, _cursor: StreamCursor, ) -> Pin<Box<dyn Future<Output = Result<Pin<Box<dyn Stream<Item = Result<InstanceTagEvent, EngineError>> + Send>>, EngineError>> + Send + 'async_trait>>

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

Subscribe to instance-tag events (tag attached / cleared).

fn unblock_execution<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _partition: Partition, _lane_id: &'life1 LaneId, _execution_id: &'life2 ExecutionId, _expected_blocking_reason: &'life3 str, _now_ms: TimestampMs, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

Unblock-scanner hook — move an execution from a blocked ZSET back to eligible_now once its blocking condition has cleared (budget under limit, quota window drained, or a capable worker has come online). expected_blocking_reason discriminates which of the blocked:{budget,quota,route} sets the execution is leaving and also fences against a stale unblock (Lua rejects if the core’s blocking_reason no longer matches).

fn drain_sibling_cancel_group<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, _flow_partition: Partition, _flow_id: &'life1 FlowId, _downstream_eid: &'life2 ExecutionId, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

RFC-016 Stage C — sibling-cancel group drain.

fn reconcile_sibling_cancel_group<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, _flow_partition: Partition, _flow_id: &'life1 FlowId, _downstream_eid: &'life2 ExecutionId, ) -> Pin<Box<dyn Future<Output = Result<SiblingCancelReconcileAction, EngineError>> + Send + 'async_trait>>

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

RFC-016 Stage D — sibling-cancel group reconcile (Invariant Q6 safety net).

fn reconcile_execution_index<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, _partition: Partition, _lanes: &'life1 [LaneId], _filter: &'life2 ScannerFilter, ) -> Pin<Box<dyn Future<Output = Result<ReconcileCounts, EngineError>> + Send + 'async_trait>>

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

Index-reconciler pass — walks ff:idx:{p:N}:all_executions and verifies each execution appears in the correct scheduling sorted set (eligible / delayed / blocked:* / active / suspended / terminal) for its current (lifecycle_phase, eligibility_state, ownership_state) triple. Phase 1 is log-only; auto-fix is deferred to a later phase (RFC-010 §6.14).

fn reconcile_budget_counters<'life0, 'async_trait>( &'life0 self, _partition: Partition, _now_ms: TimestampMs, ) -> Pin<Box<dyn Future<Output = Result<ReconcileCounts, EngineError>> + Send + 'async_trait>>

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

Budget-reconciler pass — walks ff:budget:{b:M}:policies_idx, reads each budget’s definition / usage / limits, and reconciles the breached_at marker against hard limits. Resetting budgets (non-zero reset_interval_ms) are skipped — they are handled by the budget_reset scanner (cluster 2). Drops index entries for budgets whose definition hash has been deleted (retention purge / manual). RFC-008 §Budget Reconciliation, RFC-010 §6.5.

fn reconcile_quota_counters<'life0, 'async_trait>( &'life0 self, _partition: Partition, _now_ms: TimestampMs, ) -> Pin<Box<dyn Future<Output = Result<ReconcileCounts, EngineError>> + Send + 'async_trait>>

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

Quota-reconciler pass — walks ff:quota:{q:M}:policies_idx, trims expired entries from rate-limit sliding windows, and recomputes each policy’s concurrency counter by walking its admitted_set and pruning entries whose admission guard key has TTLed out. RFC-008 §Quota Reconciliation, RFC-010 §6.6.