Struct Ctx 

pub struct Ctx { /* private fields */ }

The context handed to workflow code.

Implementations

impl Ctx

pub fn workflow_id(&self) -> &str

pub fn workflow_type(&self) -> &str

pub fn version(&self) -> u32

pub fn run_id(&self) -> RunId

pub fn started_ms(&self) -> u64

When the run started (Unix ms), as recorded.

pub fn traceparent(&self) -> Option<&str>

The W3C traceparent the run was started under (and its children join), as recorded: the parent for the spans of the calls its steps make. None when the start was not traced.

pub fn key(&self) -> String

A key stable across retries and recovery for the step whose function is running. Pass it to services that accept idempotency keys to make the step’s effect exactly once.

Call it inside a step’s function. Outside one there is no step to key: the run stops (status stuck) with a message saying so.

pub fn compensate<F, Fut, E>(&self, name: &str, f: F)

where F: Fn() -> Fut + 'static, Fut: Future<Output = Result<(), E>> + 'static, E: IntoFailure + 'static,

Registers a step that undoes what the workflow did so far: if the workflow fails or is cancelled, every registered compensation runs as a step named name, newest first, then the failure is returned. Register it after the step it undoes succeeded:

let hold = ctx.step("reserve", || inventory.reserve(sku)).await?;
ctx.compensate("release", move || inventory.release(hold));
ctx.step("charge", || payments.charge(cents)).await?; // fails: "release" runs

pub fn log(&self) -> WorkflowLog<'_>

The workflow’s log: records tagged with the workflow ID, run, and the step whose function is running, routed through crate::log (stderr by default). Silent while the run replays its journal, so a line is written once; not journaled.

ctx.log().info(format!("charging {cents}"));

pub fn step<'a, T, E, F, Fut>( &'a self, name: &'a str, f: F, ) -> Op<StepOptions, impl Make<StepOptions, Output = Result<T>, Fut = impl Future<Output = Result<T>> + 'a> + 'a>

where T: Serialize + DeserializeOwned + 'a, E: IntoFailure + 'a, F: FnMut() -> Fut + 'a, Fut: Future<Output = Result<T, E>> + 'a,

Runs f as a durable step; options by builder, then .await:

let charge: Charge = ctx
    .step("charge", || payments::charge(&quote, ctx.key()))
    .retry(RetryPolicy::attempts(5))
    .timeout(Duration::from_secs(10))
    .await?;

f runs only once every earlier record of this run is durable, so a crash can never take back an input it saw (unless the step is read_only, which runs at once). Its result is recorded and returned without waiting for the disk. On replay the recorded result is returned and f does not run. Its error is anything IntoFailure: a Failure it chose, Fusee’s Error, or any error type.

pub fn await_completion<'a, T, E, F, Fut>( &'a self, name: &'a str, send: F, ) -> Op<StepOptions, impl Make<StepOptions, Output = Result<T>, Fut = impl Future<Output = Result<T>> + 'a> + 'a>

where T: DeserializeOwned + 'a, E: IntoFailure + 'a, F: FnMut(Completion) -> Fut + 'a, Fut: Future<Output = Result<(), E>> + 'a,

A step completed from outside the workflow. send receives a completion token, starts work elsewhere (an approval email, a job on another system) and returns; the workflow then waits until a process holding the token calls engine.complete(token, result) or engine.fail(token, error). send is retried under the retry policy; each attempt gets a new token and older ones are refused.

pub fn heartbeat<T: Serialize>(&self, progress: &T) -> Result<()>

Reports progress from inside a running step. Coalesced (only the latest is kept) and never waits for the disk. A step retried after a crash receives its last durable progress through progress.

pub fn stream(&self, name: &str) -> Result<StreamWriter>

Opens the run’s stream name for appending (see crate::stream). In workflow code it opens a stream of the run: its appends are recorded once (a replay appends nothing) and close() or the run’s end completes it; steps may append to it too. Inside a step’s function it opens a stream of the step, which the step’s final result completes; an attempt that runs again finds what earlier attempts recorded in StreamWriter::recorded and continues after it. Readers follow a stream with crate::RunHandle::stream.

pub fn progress<T: DeserializeOwned>(&self) -> Result<Option<T>>

The latest progress the running step reported, in this attempt or an earlier one (including one before a crash).

pub fn on_query<A, R, F>(&self, query: impl AsQuery<A, R>, f: F)

where A: DeserializeOwned, R: Serialize, F: Fn(A) -> R + 'static,

Registers a read-only query handler: run.query(name, arg) calls it against the run’s state. It must not change state or call ctx.

pub fn on_update<A, R, V, H, Fut>( &self, update: impl AsUpdate<A, R>, validate: V, handle: H, )

where A: DeserializeOwned + 'static, R: Serialize + 'static, V: Fn(&A) -> Result<(), String> + 'static, H: Fn(A) -> Fut + 'static, Fut: Future<Output = Result<R>> + 'static,

Registers an update. validate runs first, against the current state; a rejection leaves no record. An accepted update is recorded with its ID and handle runs (it may run steps); the caller gets its result once it is durable.

pub fn object(&self, wtype: &str, key: &str) -> ObjectCall<'_>

Key key of object type wtype, to call from this workflow (or object handler): its calls are journal operations, sent through the durable outbox and run exactly once (see crate::object).

pub async fn continue_as_new<I: Serialize, T>(&self, input: &I) -> Result<T>

Ends this run and starts a new run of the same workflow ID with input and an empty journal, in one record. Signals and updates that arrive afterwards go to the new run. Never returns.

pub fn journal_size(&self) -> u32

Journal operations this run has made so far. Deterministic, so a long-lived workflow can use it to decide when to continue as new.

pub fn patched(&self, id: &str) -> Result<bool>

Lets running workflows pick up new code at this point: true for new runs and for running ones the first time they get here after the deploy; false while replaying a journal that passed this point before the patch existed. The decision is recorded.

pub fn deprecate_patch(&self, id: &str) -> Result<()>

Marks a patch whose old branch has been removed: every running workflow has passed it. Safe to delete later.

pub fn set_attributes<T: Serialize>(&self, attributes: &T) -> Result<()>

Records search attributes (an object; null removes a key). They stay readable when payloads are encrypted, so do not put secrets in them. engine.list_where filters on them.

pub fn cancel_workflow(&self, target: &str, reason: &str) -> Result<()>

Asks another workflow to cancel, once this point is durable.

pub async fn sleep(&self, name: &str, duration: Duration) -> Result<()>

Sleeps durably: the wake-up time is recorded, survives restarts, and the run is parked out of memory while it waits.

pub async fn sleep_until(&self, name: &str, at_ms: u64) -> Result<()>

Sleeps until the Unix time at_ms.

pub fn wait_for<'a, T: DeserializeOwned + 'a>( &'a self, signal: impl AsSignal<T> + 'a, ) -> Op<WaitOptions, impl Make<WaitOptions, Output = Result<Option<T>>, Fut = impl Future<Output = Result<Option<T>>> + 'a> + 'a>

Waits for the next signal; None once .timeout(d) passes: ctx.wait_for(REVIEW).timeout(DAY).await?.

pub fn child<'a, I: Serialize + 'a, O: DeserializeOwned + 'a>( &'a self, workflow: impl AsWorkflow<I, O> + 'a, id: &'a str, input: &'a I, ) -> Op<ChildOptions, impl Make<ChildOptions, Output = Result<O>, Fut = impl Future<Output = Result<O>> + 'a> + 'a>

Starts a child workflow and waits for its result. The child’s ID makes the start idempotent. Options by builder: .timeout(d), .parent_close(p).

pub fn start_child<'a, I: Serialize + 'a>( &'a self, workflow: impl NamesWorkflow<I> + 'a, id: &'a str, input: &'a I, ) -> Op<ChildOptions, impl Make<ChildOptions, Output = Result<()>, Fut = impl Future<Output = Result<()>> + 'a> + 'a>

Starts a child workflow without waiting for its result (await the start). It is terminated when this workflow closes; with .parent_close(ParentClose::Abandon) it outlives the parent.

pub fn signal_workflow<T: Serialize>( &self, id: &str, signal: impl AsSignal<T>, value: &T, ) -> Result<()>

Sends a signal to another workflow once this run’s journal is durable through this point.

pub fn now_ms(&self) -> Result<u64>

The current time, recorded: replay returns the same instant.

pub fn random(&self) -> Result<u64>

A random value, recorded.

pub fn uuid(&self) -> Result<String>

A random (version 4) UUID, recorded.

pub async fn all<F: Future>( &self, futures: impl IntoIterator<Item = F>, ) -> Vec<F::Output>

Runs futures concurrently and returns all results in argument order.

A future is polled again only when its waker fires (or the run is woken as a whole), so a fan-out costs time linear in its width; a future that can make progress must wake its waker, as the Future contract requires.

pub async fn race<F: Future>( &self, futures: impl IntoIterator<Item = F>, ) -> (usize, F::Output)

Runs futures concurrently; returns the index and result of the first to finish and drops the rest. Which one wins is recorded by the order of deliveries in the journal, so replay picks the same winner. Futures are polled as in all.

Trait Implementations

impl Clone for Ctx

fn clone(&self) -> Ctx

Returns a duplicate of the value.

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

Performs copy-assignment from source.