Struct Engine 

pub struct Engine { /* private fields */ }

Live embedded workflow engine assembled by crate::EngineBuilder.

Implementations

impl Engine

pub fn store(&self) -> Arc<dyn EventStore>

Event store used by lifecycle and delegated AD/AT operations.

pub fn visibility_store(&self) -> Arc<dyn VisibilityStore>

Visibility store used for workflow summary projections.

pub fn runtime(&self) -> &RuntimeHandle

Runtime boundary assembled for this engine.

pub fn workflow_catalog(&self) -> &Arc<WorkflowCatalog>

Shared workflow package catalog: loaded versions and routing.

pub fn registry(&self) -> &Registry

Active execution registry.

pub fn supervision(&self) -> &SupervisionTree

Supervision tree snapshot/model.

pub const fn delegated(&self) -> &DelegatedSeams

Delegated signal/query/subscribe seams installed for AT/AD integration.

pub fn signal_handoff(&self) -> Arc<SignalResumeHandoff>

Shared in-memory handoff for already-recorded non-resident signals.

pub async fn start_workflow( &self, workflow_type: &str, input: Payload, search_attributes: HashMap<String, SearchAttributeValue>, namespace: String, ) -> Result<WorkflowHandle, EngineError>

Start a loaded workflow type as a new BEAM process.

search_attributes are validated against the engine’s configured SearchAttributeSchema and recorded atomically with the WorkflowStarted event, so visibility metadata can never be lost to a crash between start and a later attribute update.

Errors

Returns EngineError::ShuttingDown after shutdown begins, and EngineError::Durability when a search attribute is unregistered or mistyped (nothing is appended and no process is spawned). Otherwise delegates to the start lifecycle transition and returns its typed errors.

pub async fn start_workflow_with_id( &self, workflow_type: &str, input: Payload, search_attributes: HashMap<String, SearchAttributeValue>, namespace: String, workflow_id: Option<WorkflowId>, routing_key: Option<String>, ) -> Result<WorkflowHandle, EngineError>

Start a loaded workflow type, optionally with a caller-chosen workflow_id and/or R-4 steered-start routing_key.

The request-routing edge supplies workflow_id to place a new start on a shard this node owns: the R-1 unsteered-start remint (any locally-owned shard) or, for a steered start, an id the edge derived on the routing_key’s shard before deciding to run locally. So a start whose id would otherwise hash to a non-owned shard never fences. When workflow_id is None this is identical to Self::start_workflow: the lifecycle mints a fresh WorkflowId, so the default single-node path is unchanged.

routing_key is the caller-chosen steered-start key recorded on the start options. Shard derivation for the cluster path is performed at the edge (which holds the concrete cluster store); here it is threaded through for API completeness and direct callers.

Errors

Identical to Self::start_workflow. A supplied workflow_id is treated as a fresh execution; the caller is responsible for choosing an unused id.

pub async fn adopt_shards(&self, shards: &[usize]) -> Result<(), EngineError>

Absorb a dead peer’s distribution shards into this LIVE engine and resume their orphaned workflows — the SS-5 failover entry point.

This is the production failover step a cluster supervisor invokes when it observes a peer gone (membership loss). It is the post-boot counterpart to the boot path’s EngineBuilder::owned_shards election + recovery, run against an already-running engine:

1. Elect + union-merge. acquire_owned_shards wins the per-shard election for each shards entry (fencing the dead owner) and become_live union-merges that shard’s committed history locally, so every event the dead node had quorum-committed is now present on this node. The election is blocking and runs off the tokio runtime inside the store seam, honouring haematite’s no-blocking-election-in-async constraint, so this async method may call it directly.
2. Widen the scope. extend_owned_shards unions shards into this node’s owned-enumeration set so the adopted workflows, timers, and outbox rows become visible to enumeration WITHOUT dropping this node’s own shards.
3. Publish ownership. publish_shard_owner records this node as each adopted shard’s current owner in the cluster’s quorum-replicated shard-owner directory (SS-3), so a request reaching a DIFFERENT survivor routes to this adopter rather than mis-resolving to the dead declared owner. The publish is fenced by the election just won, so only the true adopter writes it; a non-distributed store no-ops it.
4. Re-resident. Re-run the idempotent active-workflow recovery and timer recovery, which re-spawn every adopted workflow from the union-merged history through the same production recovery seam the boot path uses, skipping the workflows this node already owns.

Detection of the peer’s death is the CALLER’s responsibility (a cluster supervisor / membership-loss trigger); this method performs the re-acquisition and resume once that decision is made. It is idempotent: adopting a shard this node already serves re-acquires (a no-op on the fence it already holds) and recovers nothing new.

Errors

Returns EngineError::ShuttingDown after shutdown begins, store errors from the election / union-merge (EngineError::Durability), and any typed recovery error from re-residenting an adopted workflow.

pub fn resume_workflow( &self, id: &WorkflowId, run: &RunId, ) -> Result<WorkflowHandle, EngineError>

Resume a suspended workflow run and flush deferred signals through its mailbox.

Errors

Returns EngineError::WorkflowNotFound when the (workflow, run) pair is absent, or registry errors from the residency transition. Deferred delivery failures are logged and dropped because signals are already durable.

pub async fn cancel( &self, id: &WorkflowId, run: &RunId, reason: impl Into<String>, ) -> Result<(), EngineError>

Cancel a live workflow run by killing its runtime process.

Errors

Returns EngineError::ShuttingDown after shutdown begins, and EngineError::WorkflowNotFound when the (workflow, run) pair is not live. Other typed errors come from the cancel transition.

pub async fn continue_as_new( &self, id: &WorkflowId, run: &RunId, input: Payload, workflow_type: Option<String>, ) -> Result<WorkflowHandle, EngineError>

Continue a live workflow run as a new run under the same workflow id.

Errors

Returns EngineError::ShuttingDown after shutdown begins, and EngineError::WorkflowNotFound when the (workflow, run) pair is not live. Other typed errors come from the continue-as-new transition.

pub async fn reopen_workflow( &self, id: &WorkflowId, run: &RunId, ) -> Result<WorkflowHandle, EngineError>

Reopen a terminal-Failed or terminal-Cancelled run and re-drive it.

Appends a single WorkflowReopened that supersedes the run’s terminal event (returning it to Running), then respawns and re-drives the SAME run through the existing recovery path so replay returns every recorded result and only the reopened / in-flight step re-dispatches live, in the workflow’s own namespace. Takes only a workflow id and run; the reopened steps and the namespace are derived from history.

Errors

Returns EngineError::ShuttingDown after shutdown begins, EngineError::WorkflowNotFound when no history exists for the pair, and EngineError::InvalidState when the run is not a reopenable terminal (not terminal, terminal for Completed/TimedOut, or already Running).

pub fn paused_runs(&self) -> PausedRuns

The shared dispatch-hold set for durable pause (#204).

Handed to the outbox dispatcher at wiring time so a held (paused) run’s rows are never claimed, and rebuilt from [EventStore::list_paused] at startup/adoption.

pub async fn rebuild_paused_runs(&self) -> Result<(), EngineError>

Rebuild the dispatch-hold set from durable state (startup / shard adoption). A run projecting Paused is excluded from list_active respawn for free; this repopulates the hold so its pre-pause outbox rows stay unclaimed after a restart.

Errors

Returns store errors from the list_paused scan.

pub async fn pause_workflow( &self, id: &WorkflowId, run: &RunId, reason: Option<String>, operator: Option<String>, ) -> Result<WorkflowHandle, EngineError>

Pause a live Running run, durably holding NEW activity dispatch (#204).

Appends WorkflowPaused through the resident handle’s own recorder and inserts the run into the dispatch-hold set; the resident process stays alive and keeps recording (timer fires, signals, drained completions).

Errors

Returns EngineError::ShuttingDown after shutdown begins, EngineError::WorkflowNotFound when the pair has no history / no resident handle, and EngineError::InvalidState — naming the actual status — when the run is not Running.

pub async fn resume_paused_workflow( &self, id: &WorkflowId, run: &RunId, operator: Option<String>, ) -> Result<WorkflowHandle, EngineError>

Resume a Paused run, releasing the dispatch hold (#204).

Named resume_paused_workflow to avoid colliding with the existing residency-flip Engine::resume_workflow. Appends WorkflowResumed, removes the run from the dispatch-hold set, and — when the run crashed while paused and is no longer resident — respawns it via the reopen recovery path, re-arming unfired timers. The ordinary sweep then claims the released rows.

Errors

Returns EngineError::ShuttingDown after shutdown begins, EngineError::WorkflowNotFound when the pair has no history, and EngineError::InvalidState — naming the actual status — when the run is not Paused.

pub async fn result( &self, id: &WorkflowId, run: &RunId, ) -> Result<Result<Payload, WorkflowError>, EngineError>

Await a workflow run’s terminal result.

Already-terminal histories return immediately. Live workflows await their completion notifier. Unknown workflow/run pairs return not found.

Errors

Returns store, registry, or runtime channel errors as typed EngineError variants, or EngineError::WorkflowNotFound when no live handle or terminal history exists for the requested pair.

pub async fn list_workflows( &self, filter: WorkflowFilter, ) -> Result<Vec<WorkflowSummary>, EngineError>

List live and terminal workflow summaries matching filter.

Store projections are authoritative; live registry entries are projected from durable history before being merged and deduplicated.

Errors

Returns typed store or registry errors when visibility data cannot be read.

pub fn shutdown(&self) -> Result<(), EngineError>

Gracefully stop accepting new starts and shut down the embedded runtime.

Errors

Returns registry poison or runtime shutdown failures as typed errors.

impl Engine

pub const fn schedule_coordinator_workflow_id(&self) -> &WorkflowId

Workflow history used for durable schedule events and timer ownership.

pub async fn create_schedule( &self, config: ScheduleConfig, ) -> Result<ScheduleId, EngineError>

Create a durable schedule and arm its first timer.

Errors

Returns shutdown, durability, schedule projection, or timer arming errors.

pub async fn update_schedule( &self, schedule_id: &ScheduleId, config: ScheduleConfig, ) -> Result<(), EngineError>

Update an existing schedule’s configuration and re-arm it.

Errors

Returns EngineError::ShuttingDown after shutdown begins, EngineError::ScheduleNotFound for absent/deleted schedules, or typed durability, projection, and timer errors.

pub async fn pause_schedule( &self, schedule_id: &ScheduleId, ) -> Result<(), EngineError>

Pause an existing schedule.

Errors

Returns EngineError::ShuttingDown after shutdown begins, EngineError::ScheduleNotFound for absent/deleted schedules, or typed durability and projection errors.

pub async fn resume_schedule( &self, schedule_id: &ScheduleId, ) -> Result<(), EngineError>

Resume an existing schedule and re-arm it.

Errors

Returns EngineError::ShuttingDown after shutdown begins, EngineError::ScheduleNotFound for absent/deleted schedules, or typed durability, projection, and timer errors.

pub async fn delete_schedule( &self, schedule_id: &ScheduleId, ) -> Result<(), EngineError>

Delete an existing schedule so it is no longer listed or armed.

Errors

Returns EngineError::ShuttingDown after shutdown begins, EngineError::ScheduleNotFound for absent/deleted schedules, or typed durability and projection errors.

pub async fn list_schedules(&self) -> Result<Vec<ScheduleState>, EngineError>

List all non-deleted schedules from projected state.

Errors

Currently returns only infallible projected state, wrapped for API consistency.

pub async fn describe_schedule( &self, schedule_id: &ScheduleId, ) -> Result<ScheduleState, EngineError>

Describe one non-deleted schedule.

Errors

Returns EngineError::ScheduleNotFound for absent or deleted schedules.

pub async fn handle_schedule_timer_fired( &self, schedule_id: &ScheduleId, fire_at: DateTime<Utc>, ) -> Result<TimerEvaluationOutcome, EngineError>

Handles a fired durable schedule timer through the schedule evaluator.

Errors

Returns schedule evaluator or shutdown errors.

pub async fn recover_schedules_on_startup( &self, now: DateTime<Utc>, ) -> Result<(), EngineError>

Rebuilds schedule state from durable coordinator history and re-arms active schedules.

Errors

Returns schedule projection, catch-up, timer, or workflow-start errors.

impl Engine

pub async fn signal( &self, id: &WorkflowId, run: &RunId, name: impl Into<String>, payload: Payload, ) -> Result<(), EngineError>

Send a signal to a live workflow run through the AT routing seam.

Errors

Returns EngineError::WorkflowNotFound when the (workflow, run) pair is unknown, SignalRouterError::Terminal when it is durably terminal, or other typed errors from the configured signal seam.

pub async fn query( &self, id: &WorkflowId, run: &RunId, name: impl Into<String>, ) -> Result<Payload, EngineError>

Dispatch a read-only query to a live workflow run through the AT seam.

Errors

Returns EngineError::Query with crate::query::QueryError::NotRunning when the (workflow, run) pair is durably terminal, EngineError::WorkflowNotFound when it is unknown, and other typed errors from the configured query seam.

pub fn subscribe( &self, filter: EventFilter, ) -> BoxStream<'static, Result<Event, EventStreamLagged>>

Subscribe to the live event stream through the AD/AT publisher seam.

A subscriber that falls behind receives one Err(EventStreamLagged) item with the skipped count and then continues with subsequent events.

impl Engine

pub async fn load_package( &self, source: impl Into<WorkflowPackageSource>, ) -> Result<LoadOutcome, EngineError>

Loads a validated package into the running engine and atomically routes its workflow type’s new dispatches to it.

Every start that resolved before the route flip completes on the version it resolved (loads never unregister anything); every start after this call returns resolves the new version. Re-loading an already-loaded hash is idempotent (nothing registers, freshly_loaded = false) but still re-points the route at it — re-deploying a previously rolled-back version must take effect (route_changed reports whether it did).

Verified modules and catalog entries are staged first, then the archive and route are persisted, and only then are in-memory routes published. Therefore no start can record a hash whose archive is not durable. startup reloads every persisted package before recovery resolves any run’s recorded pinned version. Idempotent re-loads re-persist — re-deploying is a routing intent and the durable pointer must mirror it.

Errors

Returns EngineError::ShuttingDown once shutdown begins, EngineError::Load for archive, collision, registration, or entry-verification failures, and EngineError::ManifestMismatch when an idempotent re-load presents the resident content hash with a different manifest. On those failures live routing is untouched: routing, loaded versions, and in-flight dispatches are unaffected. Returns EngineError::Store when persistence fails; newly staged modules and entries are rolled back before the error is returned.

pub fn list_workflow_versions( &self, ) -> Result<Vec<WorkflowVersionInfo>, EngineError>

Lists every loaded workflow version with its routing flag, sorted by (workflow_type, loaded_at).

Errors

Returns EngineError::CatalogPoisoned when the catalog lock is poisoned.

pub async fn route_workflow_version( &self, workflow_type: &str, version: &ContentHash, ) -> Result<(), EngineError>

Re-points routing for workflow_type at an already-loaded version (rollback / roll-forward). Atomic and idempotent.

The pointer is persisted so the re-point survives a restart; startup restores persisted pointers after reloading persisted packages.

Errors

Returns EngineError::ShuttingDown once shutdown begins, EngineError::UnknownVersion naming the loaded set when (type, version) is not loaded — routing to a never-loaded hash is impossible — and EngineError::Store when the durable pointer could not be written. The in-memory route is published only after that write.

pub async fn unload_workflow_version( &self, workflow_type: &str, version: &ContentHash, ) -> Result<(), EngineError>

Unloads a workflow version after verifying nothing pins it (D2).

Refusal conditions, each typed and naming what pins the version: route-inactive is required (the route-active version of a type can never be unloaded), no in-flight start may pin it, no live registry handle may run on it, and no recoverable instance in the store — including a recorded-but-never-started child — may be pinned to it.

The engine owns the mechanism; the embedding platform owns when to unload. There is no automatic garbage collection.

Unload deletes the persisted deploy artifact too (a no-op for versions loaded from operator files, which were never persisted), so an unloaded version does not resurrect at the next restart.

Errors

Returns EngineError::ShuttingDown once shutdown begins, EngineError::UnknownVersion when (type, version) is not loaded, EngineError::RouteActive when the version is route-active, EngineError::VersionPinned naming the concrete pin holder (with the catalog restored untouched), EngineError::Store when the persisted artifact could not be deleted (the catalog is restored and the unload did not happen), and EngineError::Runtime when module unregistration fails after the catalog commit.

Trait Implementations

impl EngineHandle for Engine

fn resolve_workflow( &self, workflow_id: &WorkflowId, ) -> Result<WorkflowResidency, EngineSeamError>

Resolves a workflow identifier to its current residency state.

fn deliver_workflow_message( &self, process: WorkflowProcessHandle, message: WorkflowMailboxMessage, ) -> Result<(), EngineSeamError>

Delivers a message to a resident workflow process mailbox.

fn spawn_child_workflow( &self, request: ChildWorkflowSpawnRequest, ) -> Result<ChildWorkflowSpawnResult, EngineSeamError>

Requests AE to spawn a child workflow execution linked to the parent process.

fn terminate_linked_child_workflow( &self, parent_workflow_id: &WorkflowId, child_process: WorkflowProcessHandle, correlation: u64, ) -> Result<(), EngineSeamError>

Terminates a linked child workflow process through AE’s process-link boundary.

fn terminate_linked_activity( &self, parent_workflow_id: &WorkflowId, activity_process: Pid, correlation: u64, ) -> Result<(), EngineSeamError>

Terminates a linked in-VM activity process through AE’s process-link boundary.

fn arm_timer(&self, entry: TimerWheelEntry) -> Result<(), EngineSeamError>

Arms a timer-wheel entry for a resident workflow process.

fn disarm_timer( &self, process: WorkflowProcessHandle, timer_id: &TimerId, ) -> Result<(), EngineSeamError>

Disarms a timer-wheel entry for a resident workflow process.

fn record_workflow_event( &self, workflow_id: &WorkflowId, event: Event, ) -> Result<RecordOutcome, EngineSeamError>

Records an event through the target workflow’s single AD Recorder.