Struct Engine

pub struct Engine { /* private fields */ }

Expand description

An embedded Fusee engine. Cheap to clone.

Implementations§

§

impl Engine

pub async fn replay_check(&self) -> Result<Vec<ReplayReport>>

Replays every waiting or finished run’s journal against the registered code now, and returns the runs whose code differs from their journal, each naming the step. Finished runs need their history (Config::keep_history, on under the dev profile). Under the dev profile this runs by itself as runs wait or end.

§

impl Engine

pub async fn open(config: Config) -> Result<Self>

Opens (or creates) an engine. Recovery runs on the shard threads before this returns; runs resume as their workflow types register.

pub fn http_url(&self) -> Option<String>

The address of the HTTP API and UI this engine serves itself (Config::http, or the dev profile’s), as http://host:port.

pub fn is_dev(&self) -> bool

Whether this engine runs under the dev profile.

pub fn set_callback_url(&self, url: &str)

The URL this engine’s HTTP API is reached at from outside (behind a proxy, or served with crate::http::serve): completions’ url is built from it.

pub fn shards(&self) -> u32

pub fn config(&self) -> &Config

pub fn backup(&self) -> Option<Backup>

The continuous backup, when the engine was opened with Config::backup.

pub async fn stream_read( &self, id: &str, run: Option<u64>, name: &str, from: u64, wait: Duration, ) -> Result<StreamBatch>

Reads stream name of workflow id from offset from: the durable chunks there are at once, else the first ones appended within wait (none if none were), and the end once the read has every chunk. run (from an earlier answer’s crate::StreamBatch::run) keeps reading that run when the ID has moved on to a newer one.

pub fn object_type(&self, name: &str) -> ObjectType

Defines an object type (see crate::object): add its handlers, then register it.

pub fn object(&self, wtype: &str, key: &str) -> ObjectRef

Key key of object type wtype: engine.object("cart", "user-7") .call("add", &item).await?.

pub async fn objects(&self, wtype: &str, limit: usize) -> Result<Vec<String>>

Keys of object type wtype, up to limit per shard.

pub fn types_json(&self) -> Value

The registered workflow types and their versions, as the types call answers.

pub fn info_json(&self) -> Value

What this engine is, as the info call answers.

pub fn run<O: DeserializeOwned>(&self, id: &str) -> RunHandle<O>

A handle to an existing run by workflow ID.

pub async fn complete<T: Serialize>(&self, token: &str, value: &T) -> Result<()>

Finishes an async completion (ctx.await_completion) with a result. Resolves once the result is recorded durably; a stale token (an old attempt, a finished step) is rejected.

pub async fn fail(&self, token: &str, error: &str) -> Result<()>

Fails an async completion.

pub async fn reset( &self, id: &str, before_step: &str, reapply_signals: bool, ) -> Result<()>

Starts a new run of id from the current run’s journal cut just before step before_step. See [ShardRuntime::reset].

pub async fn list_where( &self, filter: &Value, limit: usize, ) -> Result<Vec<RunSummary>>

Runs matching a filter (see [crate::visibility]): the newest limit of each shard, each shard’s newest first.

pub async fn list_where_in( &self, shards: &[u32], filter: &Value, limit: usize, ) -> Result<Vec<RunSummary>>

list_where over shards only, in that order: in server mode, the shards this worker holds.

pub fn limit(&self, name: &str, spec: LimitSpec) -> Result<()>

Defines (or resizes) a concurrency and rate limit for steps.

pub fn limit_in_use(&self, name: &str) -> u32

Permits of a limit held right now.

pub fn limit_waiting(&self, name: &str) -> usize

This process’s steps waiting for a permit of a limit right now.

pub fn limits(&self) -> Vec<LimitInfo>

Every limit defined here with its permits now, by name (fusee limits, the UI’s Limits page).

pub async fn schedule( &self, name: &str, spec: ScheduleSpec, ) -> Result<ScheduleHandle>

Creates a schedule (or updates its spec) that starts spec.workflow on its cron expression or interval.

pub fn schedule_handle(&self, name: &str) -> ScheduleHandle

A handle to an existing schedule.

pub fn limit_runs(&self, wtype: &str, limit: u32)

At most limit runs of wtype run at once (per key, with StartOptions::concurrency_key); the others wait their turn, in the order they were started, as fusee.queued/<id> runs. The turns are durable: a run keeps its turn while parked and across crashes. See crate::concurrency. Set it before starting runs of the type, on every engine that starts them.

pub async fn status(&self, id: &str) -> Result<Option<RunSummary>>

pub async fn describe(&self, id: &str) -> Result<Option<RunDetail>>

A run’s summary and journal, if it exists.

pub async fn list(&self, limit: usize) -> Result<Vec<RunSummary>>

The newest limit runs of each shard, each shard’s newest first.

pub async fn shard_metrics(&self) -> Result<Vec<Metrics>>

Per-shard metrics in shard-number order.

pub async fn metrics(&self) -> Result<Metrics>

Counters summed over shards.

pub async fn close(&self) -> Result<()>

Flushes and stops every shard (and the HTTP API it serves). Pending work resumes on next open. The same as shutdown; close is the name in every SDK.

pub async fn shutdown(&self) -> Result<()>

Flushes and stops every shard. Pending work resumes on next open.

§

impl Engine

pub fn register<I, O, F, Fut>( &self, workflow: impl AsWorkflow<I, O>, f: F, ) -> WorkflowRef<I, O>
where I: Serialize + DeserializeOwned + 'static, O: Serialize + DeserializeOwned + 'static, F: Fn(Ctx, I) -> Fut + Send + Sync + 'static, Fut: Future<Output = Result<O>> + 'static,

Registers the code of a workflow type (at the reference’s version, else version 1) and returns its typed reference:

pub const CHECKOUT: WorkflowRef<Order, String> = WorkflowRef::new("checkout");
engine.register(CHECKOUT, checkout);
let run = engine.start(CHECKOUT, "order-123", &order).await?;

Runs finish on the version they started with; new starts use the newest.

§Panics

If this version of the type is already registered on this engine: two registrations would leave which code runs to chance. The message names both places.

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

Starts a run; resolves once the start is durable, with its handle. Options by builder: .id_policy(p) (or .use_existing()), .timeout(d), .version(v). Starting an ID that is running fails with Error::AlreadyExists, naming the run.

pub fn signal_with_start<'a, I, O, T>( &'a self, workflow: impl AsWorkflow<I, O> + 'a, id: &'a str, input: &'a I, signal: impl AsSignal<T> + 'a, value: &'a T, ) -> Op<StartOptions, impl Make<StartOptions, Output = Result<RunHandle<O>>, Fut = impl Future<Output = Result<RunHandle<O>>> + 'a> + 'a> ⓘ
where I: Serialize + 'a, O: DeserializeOwned + 'a, T: Serialize + 'a,

Signals the run id, starting it first if it is not running (with input): the signal is recorded before the new run’s code first runs. Returns the run.

engine.signal_with_start(CART, "cart-7", &Cart::default(), ADD, &item).await?;

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

Starts a run and waits for its result, in one call to the shard instead of start then RunHandle::result.

pub fn execute_then<I, O>( &self, workflow: impl AsWorkflow<I, O>, id: &str, input: &I, done: impl FnOnce(Result<O>) + Send + 'static, ) -> Result<()>
where I: Serialize, O: DeserializeOwned,

execute, with the result handed to done on the run’s shard thread instead of a future: for callers that record results and need no task or wake per run. done must be short and must not block. An error returned here means done is not called. See run_raw_then.

pub fn debounce<'a, I, O>( &'a self, workflow: impl AsWorkflow<I, O> + 'a, key: &'a str, input: &'a I, rule: &'a Debounce, ) -> Op<EventOptions, impl Make<EventOptions, Output = Result<RunHandle<O>>, Fut = impl Future<Output = Result<RunHandle<O>>> + 'a> + 'a> ⓘ
where I: Serialize + 'a, O: DeserializeOwned + 'a,

Debounces a start under key: starts inside the window collapse into one run, which begins with the latest input when the window closes (see crate::queue). Resolves once this start is durable, with the handle of that run; the run exists once the window closes (RunHandle::started). .event_id(id) makes a retried call join once.

pub fn batch<'a, E, O>( &'a self, workflow: impl AsWorkflow<Vec<E>, O> + 'a, key: &'a str, event: &'a E, rule: &'a Batch, ) -> Op<EventOptions, impl Make<EventOptions, Output = Result<RunHandle<O>>, Fut = impl Future<Output = Result<RunHandle<O>>> + 'a> + 'a> ⓘ
where E: Serialize + 'a, O: DeserializeOwned + 'a,

Adds an event to the batch under key: once the batch holds max_events events or max_wait has passed since its first, one run starts with all of them as its input (see crate::queue). Resolves once the event is durable, with the handle of that run.