Struct FlowFabricWorker 

pub struct FlowFabricWorker { /* private fields */ }

FlowFabric worker — connects to Valkey, claims executions, and provides the worker-facing API.

Admission control

claim_next() lives behind the direct-valkey-claim feature flag and bypasses the scheduler’s admission controls: it reads the eligible ZSET directly and mints its own claim grant without consulting budget ({b:M}) or quota ({q:K}) policies. Default-off. Intended for benchmarks, tests, and single-tenant development where the scheduler hop is measurement noise, not for production.

For production deployments, consume scheduler-issued grants via FlowFabricWorker::claim_from_grant — the scheduler enforces budget breach, quota sliding-window, concurrency cap, and capability-match checks before issuing grants.

Usage

use ff_core::backend::BackendConfig;
use ff_core::types::{LaneId, Namespace, WorkerId, WorkerInstanceId};
use ff_sdk::{FlowFabricWorker, WorkerConfig};

let config = WorkerConfig {
    backend: Some(BackendConfig::valkey("localhost", 6379)),
    worker_id: WorkerId::new("w1"),
    worker_instance_id: WorkerInstanceId::new("w1-i1"),
    namespace: Namespace::new("default"),
    lanes: vec![LaneId::new("main")],
    capabilities: Vec::new(),
    lease_ttl_ms: 30_000,
    claim_poll_interval_ms: 1_000,
    max_concurrent_tasks: 1,
    partition_config: None,
};
let worker = FlowFabricWorker::connect(config).await?;

loop {
    if let Some(task) = worker.claim_next().await? {
        // Process task...
        task.complete(Some(b"result".to_vec())).await?;
    } else {
        tokio::time::sleep(Duration::from_secs(1)).await;
    }
}

Implementations

impl FlowFabricWorker

pub async fn describe_execution( &self, id: &ExecutionId, ) -> Result<Option<ExecutionSnapshot>, SdkError>

Read a typed snapshot of one execution. See ExecutionSnapshot for field semantics.

Post-T3 this is a thin forwarder onto the bundled EngineBackend::describe_execution impl; the HGETALL pipeline body and strict-parse decoder both live in ff-backend-valkey / ff-core so alternate backends can serve the same call shape. Parse failures surface as SdkError::Engine(EngineError::Validation { kind: Corruption, .. }).

pub async fn describe_flow( &self, id: &FlowId, ) -> Result<Option<FlowSnapshot>, SdkError>

Read a typed snapshot of one flow. See FlowSnapshot for field semantics.

Post-T3 this is a thin forwarder onto the bundled EngineBackend::describe_flow impl.

pub async fn set_edge_group_policy( &self, flow_id: &FlowId, downstream_execution_id: &ExecutionId, policy: EdgeDependencyPolicy, ) -> Result<SetEdgeGroupPolicyResult, SdkError>

RFC-016 Stage A: declare the inbound edge-group policy for a downstream execution. Must be called BEFORE the first add_dependency(... -> downstream_execution_id) on this flow.

Stage A accepts only EdgeDependencyPolicy::AllOf; AnyOf / Quorum return a typed EngineError::Validation error until Stage B’s resolver lands.

pub async fn list_flows( &self, partition: PartitionKey, cursor: Option<FlowId>, limit: usize, ) -> Result<ListFlowsPage, SdkError>

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

partition is an opaque PartitionKey — typically obtained from a crate::ClaimGrant or from ff_core::partition::flow_partition when the caller already knows a FlowId on the partition it wants to enumerate. cursor is None on the first call and the previous page’s next_cursor on subsequent calls; iteration terminates when the returned next_cursor is None.

Thin forwarder onto EngineBackend::list_flows.

pub async fn describe_edge( &self, flow_id: &FlowId, edge_id: &EdgeId, ) -> Result<Option<EdgeSnapshot>, SdkError>

Read a typed snapshot of one dependency edge.

Takes both flow_id and edge_id: the edge hash is stored under the flow’s partition (ff:flow:{fp:N}:<flow_id>:edge:<edge_id>) and FF does not maintain a global edge_id -> flow_id index. The caller already knows the flow from the staging call result or the consumer’s own metadata.

Returns Ok(None) when the edge hash is absent (never staged, or staged under a different flow).

Post-#160 this is a thin forwarder onto the bundled EngineBackend::describe_edge impl.

pub async fn list_outgoing_edges( &self, upstream_eid: &ExecutionId, ) -> Result<Vec<EdgeSnapshot>, SdkError>

List all outgoing dependency edges originating from an execution.

Returns an empty Vec when the execution has no outgoing edges — including standalone executions not attached to any flow.

Reads

1. EngineBackend::resolve_execution_flow_id (single HGET on exec_core.flow_id for the Valkey backend).
2. EngineBackend::list_edges on the resolved flow (SMEMBERS
   • pipelined HGETALL on the flow’s partition).

Ordering is unspecified — the adjacency set is an unordered SET. Callers that need deterministic order should sort by EdgeSnapshot::edge_id or created_at.

pub async fn list_incoming_edges( &self, downstream_eid: &ExecutionId, ) -> Result<Vec<EdgeSnapshot>, SdkError>

List all incoming dependency edges landing on an execution. See [list_outgoing_edges] for the read shape.

pub async fn list_lanes( &self, cursor: Option<LaneId>, limit: usize, ) -> Result<ListLanesPage, SdkError>

Enumerate registered lanes with cursor-based pagination.

Thin forwarder onto EngineBackend::list_lanes. Lanes are global (not partition-scoped); the Valkey backend serves this from the ff:idx:lanes SET, sorts by lane name, and returns a limit-sized page starting after cursor (exclusive). Loop until ListLanesPage::next_cursor is None to read the full registry.

pub async fn list_suspended( &self, partition: PartitionKey, cursor: Option<ExecutionId>, limit: usize, ) -> Result<ListSuspendedPage, SdkError>

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

Thin forwarder onto the bundled EngineBackend::list_suspended impl. cursor = None starts a fresh scan; feed the returned ListSuspendedPage::next_cursor back in to page forward until it returns None. See ff_core::contracts::SuspendedExecutionEntry for the per-row fields (including the free-form reason code).

pub async fn list_executions( &self, partition: PartitionKey, cursor: Option<ExecutionId>, limit: usize, ) -> Result<ListExecutionsPage, SdkError>

Forward-only paginated listing of executions in a partition.

Thin forwarder onto EngineBackend::list_executions. Pagination is cursor-based: pass cursor = None for the first page and feed the returned ListExecutionsPage::next_cursor back as cursor until it returns None. See the trait rustdoc for the full contract.

pub async fn rotate_waitpoint_hmac_secret_all( &self, args: RotateWaitpointHmacSecretAllArgs, ) -> Result<RotateWaitpointHmacSecretAllResult, SdkError>

Cluster-wide rotation of the waitpoint HMAC signing kid (v0.7 Q4).

Thin forwarder onto EngineBackend::rotate_waitpoint_hmac_secret_all. On the Valkey backend this fans out one FCALL per execution partition; on the Postgres backend (Wave 4) this resolves to a single global INSERT. Consumers call this for the clean “rotate everywhere” semantic — the older per-partition crate::admin::rotate_waitpoint_hmac_secret_all_partitions free function stays available for direct-Valkey callers on legacy SDKs.

impl FlowFabricWorker

pub async fn connect(config: WorkerConfig) -> Result<Self, SdkError>

Connect to Valkey and prepare the worker.

Establishes the ferriskey connection. Does NOT load the FlowFabric library — that is the server’s responsibility (ff-server calls ff_script::loader::ensure_library() on startup). The SDK assumes the library is already loaded.

Smoke / dev scripts: rotate WorkerInstanceId

The SDK writes a SET-NX liveness sentinel keyed on the worker’s WorkerInstanceId. When a smoke / dev script reuses the same WorkerInstanceId across restarts, subsequent runs trap behind the prior run’s SET-NX until the liveness key’s TTL (≈ 2× the configured lease TTL) expires — the worker appears stuck and claims nothing. Iterative scripts should synthesise a fresh WorkerInstanceId per process (e.g. WorkerInstanceId::new() or embed a UUID/timestamp) rather than hard-coding a stable value. Production workers that cleanly shut down release the key; only crashed / kill -9’d processes hit this trap.

RFC-023 Phase 1a (§4.4 item 10d, v0.12.0): this Valkey-bundled entry point is #[cfg(feature = "valkey-default")]-gated. Consumers on the sqlite-only feature set must use FlowFabricWorker::connect_with and drive the backend directly through the trait surface.

pub async fn connect_with( config: WorkerConfig, backend: Arc<dyn EngineBackend>, completion: Option<Arc<dyn CompletionBackend>>, ) -> Result<Self, SdkError>

Store pre-built EngineBackend and (optional) CompletionBackend handles on the worker. Builds the worker via the legacy FlowFabricWorker::connect path first (so the embedded ferriskey::Client that the Stage 1b non-migrated hot paths still use is dialed), then replaces the default ValkeyBackend wrapper with the caller-supplied trait objects.

The completion argument is explicit: 0.3.3 previously accepted only backend and completion_backend() silently returned None on this path because Arc<dyn EngineBackend> cannot be upcast to Arc<dyn CompletionBackend> without loss of trait-object identity. 0.3.4 lets the caller decide.

• Some(arc) — caller supplies a completion backend. Self::completion_backend returns Some(clone).
• None — this backend does not support push-based completion (future Postgres backend without LISTEN/NOTIFY, test mocks). Self::completion_backend returns None.

When the underlying backend implements both traits (as ValkeyBackend does), pass the same Arc twice — the two trait-object views share one allocation:

use std::sync::Arc;
use ff_backend_valkey::ValkeyBackend;
use ff_sdk::{FlowFabricWorker, WorkerConfig};

// Valkey (completion supported):
let valkey = Arc::new(ValkeyBackend::connect(backend_config).await?);
let worker = FlowFabricWorker::connect_with(
    worker_config,
    valkey.clone(),
    Some(valkey),
).await?;

Backend without completion support:

let worker = FlowFabricWorker::connect_with(
    worker_config,
    backend,
    None,
).await?;

Stage 1b + Round-7 scope — what the injected backend covers today. The injected backend currently covers these per-task ClaimedTask ops: update_progress / resume_signals / delay_execution / move_to_waiting_children / complete / cancel / fail / create_pending_waitpoint / append_frame / report_usage. A mock backend therefore sees that portion of the worker’s per-task write surface. Lease renewal also routes through backend.renew(&handle). Round-7 (#135/#145) closed the four trait-shape gaps tracked by #117, but suspend still reaches the embedded ferriskey::Client directly via ff_suspend_execution — this is the deferred suspend per RFC-012 §R7.6.1, pending Stage 1d input-shape work. claim_next / claim_from_grant / claim_from_reclaim_grant / deliver_signal / admin queries are Stage 1c hot-path work. Stage 1d removes the embedded client entirely.

Today’s constructor is therefore NOT yet a drop-in way to swap in a non-Valkey backend — it requires a reachable Valkey node for suspend plus the remaining hot-path ops. Tests that exercise only the migrated per-task ops can run fully against a mock backend.

pub fn backend(&self) -> Option<&Arc<dyn EngineBackend>>

Borrow the EngineBackend this worker forwards Stage-1b trait ops through.

RFC-012 Stage 1b. Always returns Some(&self.backend) — the Option wrapper is retained for API stability with the Stage-1a shape. Stage 1c narrows the return type to &Arc<dyn EngineBackend>.

pub fn completion_backend(&self) -> Option<Arc<dyn CompletionBackend>>

Handle to the completion-event subscription backend, for consumers that need to observe execution completions (DAG reconcilers, tenant-isolated subscribers).

Returns Some when the worker was built through Self::connect on the default valkey-default feature (the bundled ValkeyBackend implements CompletionBackend), or via Self::connect_with with a Some(..) completion handle. Returns None when the caller passed None to Self::connect_with — i.e. the backend does not support push-based completion streams (future Postgres without LISTEN/NOTIFY, test mocks).

The returned handle shares the same underlying allocation as Self::backend; calls through it (e.g. subscribe_completions_filtered) hit the same connection the worker itself uses.

pub fn config(&self) -> &WorkerConfig

Get the worker config.

pub fn partition_config(&self) -> &PartitionConfig

Get the server-published partition config this worker bound to at connect(). Exposed so consumers that mint custom ExecutionIds (e.g. for describe_execution lookups on ids produced outside this worker) stay aligned with the server’s num_flow_partitions — using PartitionConfig::default() assumes 256 partitions and silently misses data on deployments with any other value.

pub async fn claim_next_via_backend( &self, ) -> Result<Option<ClaimedTask>, SdkError>

Attempt to claim the next eligible execution through the [EngineBackend] trait — works on any backend with scan_eligible_executions + issue_claim_grant + claim_execution trait bodies (Valkey today; Postgres/SQLite when their grant-consumer RFC lands).

Simplified claim flow:

1. Pick a lane (round-robin across configured lanes)
2. Scan a [PARTITION_SCAN_CHUNK] window of partitions for eligible executions via backend.scan_eligible_executions
3. Issue a claim grant via backend.issue_claim_grant
4. Claim the execution via backend.claim_execution
5. Return a ClaimedTask with auto lease renewal

Scheduler bypass — read before production use

This path bypasses the scheduler’s budget / quota admission checks. Enable only when the scheduler hop is measurement noise (benches, single-tenant dev, examples demonstrating the claim primitive) or when the test harness wants a deterministic worker-local path. For production, consume scheduler-issued grants via Self::claim_from_grant — the scheduler enforces budget breach, quota sliding-window, concurrency cap, and capability-match checks before issuing grants.

None semantics

Ok(None) means no work was found in the partition window this poll covered, not “the cluster is idle”. Each call scans a chunk of [PARTITION_SCAN_CHUNK] partitions starting at the rolling scan_cursor; the cursor advances by that chunk size on every invocation, so a worker covers every partition exactly once every ceil(num_flow_partitions / PARTITION_SCAN_CHUNK) polls.

Callers should treat None as “poll again soon” (typically after config.claim_poll_interval_ms) rather than “sleep for a long time”. Backing off too aggressively on None can starve workers when work lives on partitions outside the current window.

Returns Err on backend errors or script failures.

pub async fn claim_from_grant( &self, lane: LaneId, grant: ClaimGrant, ) -> Result<ClaimedTask, SdkError>

Consume a ClaimGrant and claim the granted execution on this worker. The intended production entry point: pair with ff_scheduler::Scheduler::claim_for_worker to flow scheduler-issued grants into the SDK without enabling the direct-valkey-claim feature (which bypasses budget/quota admission control).

The worker’s concurrency semaphore is checked BEFORE the FCALL so a saturated worker does not consume the grant: the grant stays valid for its remaining TTL and the caller can either release it back to the scheduler or retry after some other in-flight task completes.

On success the returned ClaimedTask holds a concurrency permit that releases automatically on complete/fail/cancel/drop — same contract as claim_next.

Arguments

• lane — the lane the grant was issued for. Must match what was passed to Scheduler::claim_for_worker; the Lua FCALL uses it to look up lane_eligible, lane_active, and the worker_leases index slot.
• grant — the ClaimGrant returned by the scheduler.

Errors

• SdkError::WorkerAtCapacity — max_concurrent_tasks permits all held. Retryable; the grant is untouched.
• ScriptError::InvalidClaimGrant — grant missing, consumed, or worker_id mismatch (wrapped in SdkError::Engine).
• ScriptError::ClaimGrantExpired — grant TTL elapsed (wrapped in SdkError::Engine).
• ScriptError::CapabilityMismatch — execution’s required capabilities not a subset of this worker’s caps (wrapped in SdkError::Engine). Surfaced post-grant if a race between grant issuance and caps change allows it.
• ScriptError::Parse — ff_claim_execution returned an unexpected shape (wrapped in SdkError::Engine).
• SdkError::Backend / SdkError::BackendContext — transport error during the FCALL or the read_execution_context follow-up.

Backend coverage (v0.12 PR-5.5)

Method ungated across backends. The Valkey backend handles the grant fully. Postgres + SQLite backends return EngineError::Unavailable from EngineBackend::claim_execution today — grants on those backends flow through the scheduler-routed claim_via_server path (the PG/SQLite scheduler lives outside the EngineBackend trait in this release). See project_claim_from_grant_pg_sqlite_gap.md for motivation and planned follow-up.

pub async fn claim_via_server( &self, admin: &FlowFabricAdminClient, lane: &LaneId, grant_ttl_ms: u64, ) -> Result<Option<ClaimedTask>, SdkError>

Scheduler-routed claim: POST the server’s /v1/workers/{id}/claim, then chain to Self::claim_from_grant.

Batch C item 2 PR-B. This is the production entry point — budget + quota + capability admission run server-side inside ff_scheduler::Scheduler::claim_for_worker. Callers don’t enable the direct-valkey-claim feature.

Returns Ok(None) when the server says no eligible execution (HTTP 204). Callers typically back off by config.claim_poll_interval_ms and try again, same cadence as the direct-claim path’s Ok(None).

The admin client is the established HTTP surface (FlowFabricAdminClient) reused here so workers don’t keep a second reqwest client around. Build once at worker boot and hand in by reference on every claim.

pub async fn claim_from_resume_grant( &self, grant: ResumeGrant, ) -> Result<ClaimedTask, SdkError>

Consume a ResumeGrant and transition the granted attempt_interrupted execution into a started state on this worker. Symmetric partner to claim_from_grant for the resume path.

Renamed from claim_from_reclaim_grant (RFC-024 PR-B+C). The new claim_from_reclaim_grant method lands with PR-G and dispatches to [EngineBackend::reclaim_execution] for the lease-reclaim path (distinct semantic, distinct grant type).

The grant must have been issued to THIS worker (matching worker_id at grant time). A mismatch returns Err(Script(InvalidClaimGrant)). The grant is consumed atomically by ff_claim_resumed_execution; a second call with the same grant also returns InvalidClaimGrant.

Concurrency

The worker’s concurrency semaphore is checked BEFORE the FCALL (same contract as claim_from_grant). Reclaim does NOT assume pre-existing capacity on this worker — a reclaim can land on a fresh worker instance that just came up after a crash/restart and is picking up a previously-interrupted execution. If the worker is saturated, the grant stays valid for its remaining TTL and the caller can release it or retry.

On success the returned ClaimedTask holds a concurrency permit that releases automatically on complete/fail/cancel/drop.

Errors

• SdkError::WorkerAtCapacity — max_concurrent_tasks permits all held. Retryable; the grant is untouched (no FCALL was issued, so ff_claim_resumed_execution did not atomically consume the grant key).
• ScriptError::InvalidClaimGrant — grant missing, consumed, or worker_id mismatch.
• ScriptError::ClaimGrantExpired — grant TTL elapsed.
• ScriptError::NotAResumedExecution — attempt_state is not attempt_interrupted.
• ScriptError::ExecutionNotLeaseable — lifecycle_phase is not runnable.
• ScriptError::ExecutionNotFound — core key missing.
• SdkError::Backend / SdkError::BackendContext — transport.

pub async fn claim_from_reclaim_grant( &self, grant: ReclaimGrant, args: ReclaimExecutionArgs, ) -> Result<ReclaimExecutionOutcome, SdkError>

Consume a ReclaimGrant to mint a fresh attempt for a lease-expired / lease-revoked execution (RFC-024 §3.4).

Backend-agnostic. Routes through EngineBackend::reclaim_execution on whatever backend the worker was connected with (Valkey via connect / connect_with, Postgres / SQLite via connect_with). Distinct from claim_from_resume_grant: reclaim creates a NEW attempt row and bumps lease_reclaim_count (HandleKind::Reclaimed), while resume re-uses the existing attempt under ff_claim_resumed_execution (HandleKind::Resumed).

Return shape

Returns the raw ReclaimExecutionOutcome so consumers match on the four outcomes (Claimed(Handle), NotReclaimable, ReclaimCapExceeded, GrantNotFound) and decide their dispatch. The Claimed variant carries a Handle whose kind is HandleKind::Reclaimed; downstream ops (complete, fail, renew, append_frame, …) take the handle directly via the EngineBackend trait.

This contrasts with claim_from_resume_grant, which wraps the handle in a ClaimedTask with a concurrency-permit and auto-lease-renewal loop. Those affordances are valkey-default-gated today (they depend on the bundled ferriskey::Client + lease_ttl_ms renewal timer). The reclaim surface is intentionally narrower so it compiles under --no-default-features, features = ["sqlite"] and consumers can drive the reclaim flow on any backend.

Feature compatibility

No cfg-gate. Compiles + runs under every feature set ff-sdk supports (including sqlite-only). Verified by the compile-time type assertion worker_claim_from_reclaim_grant_is_backend_agnostic_at_type_level in crates/ff-sdk/tests/rfc024_sdk.rs, which pins the method’s full signature under the default feature set and is paralleled by sqlite_only_compile_surface_tests in this file for the --no-default-features, features = ["sqlite"] compile anchor on the rest of the backend-agnostic surface.

worker_capabilities

ReclaimExecutionArgs::worker_capabilities is NOT part of the Lua FCALL (the reclaim Lua validates grant consumption via grant.worker_id == args.worker_id only — see crates/ff-script/src/flowfabric.lua:3088 and RFC-024 §4.4). Capability matching happens at grant-issuance time (see FlowFabricAdminClient::issue_reclaim_grant — in the ff-sdk::admin module, valkey-default-gated).

Errors

• SdkError::Engine — the backend’s reclaim_execution returned an EngineError (transport fault, validation failure, Unavailable from a backend that does not implement RFC-024 — currently only pre-RFC out-of-tree backends).

pub async fn deliver_signal( &self, execution_id: &ExecutionId, waitpoint_id: &WaitpointId, signal: Signal, ) -> Result<SignalOutcome, SdkError>

Deliver a signal to a suspended execution’s waitpoint.

The engine atomically records the signal, evaluates the resume condition, and optionally transitions the execution from suspended to runnable.

Forwards through EngineBackend::deliver_signal — the trait-level trigger surface landed in issue #150.

Backend-agnostic as of v0.12 PR-3. Compiles + runs under every feature set ff-sdk supports (including --no-default-features --features sqlite); pinned by sqlite_only_compile_surface_tests::deliver_signal_addressable_under_sqlite_only.