Struct SqliteBackend 

pub struct SqliteBackend { /* private fields */ }

RFC-023 SQLite dev-only backend.

Construction demands FF_DEV_MODE=1 (§4.5). Identical paths within a process return the same handle via the §4.2 B6 registry.

Implementations

impl SqliteBackend

pub async fn new(path: &str) -> Result<Arc<Self>, BackendError>

RFC-023 Phase 1a entry point. path accepts a filesystem path, :memory:, or a file:...?mode=memory&cache=shared URI.

Uses the SqliteServerConfig defaults (pool size 4, WAL on for file paths). For operator-tuned pool/WAL settings, call [SqliteBackend::new_with_config].

Errors

• BackendError::RequiresDevMode when FF_DEV_MODE is unset or not "1".
• BackendError::Valkey (historical name — the classifier is backend-agnostic despite the variant name) when the pool cannot be constructed.

pub async fn new_with_tuning( path: &str, pool_size: u32, wal_mode: bool, ) -> Result<Arc<Self>, BackendError>

Operator-tuned entry point. pool_size sets the pool’s max connections; wal_mode enables PRAGMA journal_mode=WAL for filesystem-backed databases (ignored for :memory: variants per RFC-023 §4.6).

pub fn with_scanners(&self, cfg: SqliteScannerConfig) -> bool

RFC-023 Phase 3.5: spawn the N=1 scanner supervisor (currently budget_reset only) as a background tick loop. Idempotent: the first caller wins; subsequent calls on the same registry-shared backend no-op. Drained on EngineBackend::shutdown_prepare.

Returns true if this call installed the supervisor, false if a supervisor was already present.

Trait Implementations

impl Clone for SqliteBackend

fn clone(&self) -> SqliteBackend

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for SqliteBackend

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

Formats the value using the given formatter.

impl EngineBackend for SqliteBackend

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-023 Phase 3.5: drain the scanner supervisor (if installed) up to grace. Matches the PG 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 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 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 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 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 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 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 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 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_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 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,

Create an execution. Ingress row 6 (RFC-017 §4). Wraps ff_create_execution on Valkey; INSERT INTO ff_execution ... on Postgres. The idempotency_key + backend-side default dedup_ttl_ms = 86400000 make duplicate submissions idempotent.

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 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 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 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 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 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 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 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 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 capabilities(&self) -> Capabilities

RFC-018 Stage A: snapshot of this backend’s identity + the flat Supports surface it can actually service. Consumers use this at startup to gate UI features / choose between alternative code paths before dispatching. See rfcs/RFC-018-backend-capability-discovery.md for the full discovery contract and the four owner-adjudicated open questions (granularity: coarse; version: struct; sync; no event stream).

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: run one-time backend-specific boot preparation.

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 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 cascade_completion<'life0, 'life1, 'async_trait>( &'life0 self, _payload: &'life1 CompletionPayload, ) -> Pin<Box<dyn Future<Output = Result<CascadeOutcome, EngineError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, Self: '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 ping<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), EngineError>> + Send + 'async_trait>>

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

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

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

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

Scheduler-routed claim entrypoint (row 18, RFC-017 §7). Valkey forwards to its ff_scheduler::Scheduler cursor; Postgres forwards to PostgresScheduler’s FOR UPDATE SKIP LOCKED path.

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 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 'life0: 'async_trait, 'life1: 'async_trait, Self: '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 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, Self: '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 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, Self: '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 'life0: 'async_trait, 'life1: 'async_trait, Self: '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 'life0: 'async_trait, 'life1: 'async_trait, Self: 'async_trait,

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

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.

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 'life0: 'async_trait, 'life1: 'async_trait, Self: '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 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, Self: '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).