Struct Engine 

pub struct Engine { /* private fields */ }

Live embedded workflow engine assembled by crate::EngineBuilder.

Implementations

impl Engine

pub fn worker_contracts_for_admission( &self, task_queue: &str, ) -> Result<QueueAdmission, EngineError>

Splits task_queue’s retained contracts into the set a registering worker must satisfy and the set nothing can reach.

See the module documentation for the rule and its limits.

Errors

Returns EngineError::CatalogPoisoned when the catalog snapshot or start-pin lock is poisoned, and EngineError::RegistryPoisoned when the active-execution registry lock is poisoned. Admission never guesses on a poisoned lock: an unreadable liveness answer refuses the worker rather than admitting one whose obligations are unknown.

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 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 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 async fn record_activity_lease( &self, id: &WorkflowId, run: &RunId, activity_id: ActivityId, attempt: u32, worker: WorkerAttribution, ) -> Result<(), EngineError>

Record that worker accepted attempt of activity_id on run.

The lease is informational: nothing is delivered to the workflow process and nothing wakes, because a lease changes no wait the run is parked on — it names who is doing the work. It is appended through the run’s single Recorder, serialising with every timer, signal and completion arrival for that run, so concurrent arrivals never race the sequence head.

Errors

Returns EngineError::ActivityLeaseAfterTerminal when the run has already reached a terminal event (nothing recorded), EngineError::WorkflowNotFound when the (workflow, run) pair is unknown or its handle never appears within the registration birth window, and the store or durability error when the append fails.

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 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 aion_core::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 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 [aion_store::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 rename_workflow( &self, id: &WorkflowId, run: &RunId, display_name: &str, ) -> Result<String, EngineError>

Record a new operator-facing display name for a run (#211).

The name is a LABEL over the UUID identity, never an address: nothing resolves a workflow by name. Every rename is a RECORDED SearchAttributesUpdated event (the aion.display_name attribute), so history keeps every name that has been worn and the current name is the last-write-wins fold.

Both halves of that are true at once and are easy to confuse. You ADDRESS a run — this operation takes a workflow id AND a run id, because which run is live decides whether the append is safe — but what you RECORD is a workflow-level attribute with no run id in it. Readers fold it over the whole history, so the name reads back for every run of the workflow, a continue-as-new successor inherits it, and (below) a superseded predecessor cannot be renamed at all.

There is no status precondition on the label itself — a completed run’s name is as legitimate as a running one’s. Status decides only HOW the event is appended: a REGISTERED run appends through its own single-writer recorder under that recorder’s lock, while a run with no registered handle appends through a one-shot recorder at the durable head — and only when its durable status proves no live recorder can exist (terminal, or Paused and so excluded from respawn by design). A non-resident run in any other status is REFUSED rather than raced. A run superseded by a later run of the same workflow is refused too: SearchAttributesUpdated carries no run id, so a recorded name would land on the later run. So is a run that has recorded its own WorkflowContinuedAsNew and is therefore ABOUT to be superseded — its successor’s recorder has already read the head this rename would append to, and appending there would strand the chain with no successor run.

Returns the name exactly as recorded (trimmed).

Errors

Returns EngineError::ShuttingDown after shutdown begins; EngineError::InvalidState when the trimmed name is empty, when the run is non-terminal, not Paused, and not resident on this node (RETRY once it is resident), or when the run has been superseded or has continued as new; EngineError::WorkflowNotFound when no history exists for the pair; and EngineError::Durability when the schema refuses the attribute or the store rejects the append. Every rejection appends nothing.

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, request: &WorkflowListRequest, ) -> Result<WorkflowListPage, EngineError>

Answers one page of the workflow list contract from the visibility projection.

The projection is the single source for listing: every durable append maintains its row through the Recorder, and boot, adoption, and the periodic repair loop reconcile it with history. Nothing here reads history or the live registry.

Errors

Returns EngineError::Store wrapping aion_store::StoreError::InvalidQuery for a zero limit or a cursor minted under a different query, and typed store errors when the projection cannot be read.

impl Engine

pub fn workloop_service(&self) -> Option<Arc<WorkloopService>>

The cadence service, when configured — the operational surface for sweeps and registration state.

pub fn declared_workloop_spec( &self, workflow_type: &str, ) -> Result<Option<WorkloopSpec>, EngineError>

The workloop spec the currently ROUTED deployment of workflow_type declares, or None when that deployment is an ordinary workflow.

🔴 THE START PATH ASKS THIS, AND IT IS WHY A .awl WORKLOOP RUNS

Every caller-facing start surface — HTTP, gRPC, the CLI — funnels into one start verb, and that verb cannot tell a workloop from a workflow by looking at the request: nothing in a start request says “this is a loop”. The DEPLOYED PACKAGE says so, in the contract the compiler bound into its identity, and this is where that is read. Without it a workloop deploys, starts, runs its first iteration and is REFUSED at close_iteration for not being a registered workloop — which is exactly the shape this whole integration exists to close.

Answering from the ROUTED version (not an exact pin) is deliberate: a start goes to whatever version routing would run, so the declaration a start is registered under must come from the same place.

Errors

Propagates catalog failures, and refuses a declaration the engine cannot act on — an unarmed loop, an invariant with no tolerance or no confirming route, a retention window of zero. Those are refused HERE, before anything is registered or started, so the loop that cannot be armed never exists rather than existing and never firing.

pub async fn register_workloop( &self, loop_id: &WorkflowId, namespace: String, spec: WorkloopSpec, ) -> Result<(), EngineError>

Registers a STARTED workflow as a workloop: stamps the aion.kind listing attribute durably in its history and arms the declared cadence and tolerance deadlines on the sweep set. Every declared value — arming, tolerance, retention — arrives validated inside spec; there are no defaults to assume.

Errors

Refuses an unknown or non-Running workflow, a duplicate registration, and propagates store/append failures.

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

Starts a workflow AS A WORKLOOP: seeds generation 1’s carry, registers the loop, and only then lets the body run.

🔴 THIS EXISTS BECAUSE START-THEN-REGISTER IS A RACE

register_workloop requires an already-RUNNING workflow, so the only way to stand a loop up was to start it and then register it — and the started body begins executing immediately. Generation 1 could reach close_iteration BEFORE its registration landed, and the close would then refuse with “not a registered workloop” and FAIL the run. A loop’s very first iteration could lose a race with its own registration.

The registration row is therefore written FIRST, before the workflow exists, so the close always finds it. A failed start removes the row again rather than leaving a registration for a loop that never ran.

🔴 AND BECAUSE GENERATION 1 HAS NO PREVIOUS ITERATION TO CARRY FROM

Every later generation gets its carry from the previous iteration’s route start payload. Generation 1 has none, and the compiled input codec requires the carry fields regardless — so a start payload that passes schema admission would then be UNDECODABLE by the workflow it was admitted for. The declared carry defaults are merged into the generation-1 input here, filling only ABSENT fields so a caller-supplied value is never clobbered by a default.

🔴 THE KIND STAMP IS PART OF THE START, NOT A THIRD STEP

The aion.kind attribute travels in the START’s own attribute map, so WorkflowStarted and SearchAttributesUpdated land in ONE append (Recorder::record_workflow_started_with_attributes). It used to be a separate durable step after the start, and the gap was not cosmetic: boot recovery reads the kind FROM HISTORY to decide whether a Running workflow with no resident process is a crashed workflow to resurrect or a parked LOOP to leave alone. A crash between the start and the stamp left a registered loop that boot recovery did not recognise as one — it resurrected the generation resident while the sweep set still carried the row and would wake it, which is two paths driving one loop, the exact thing the skip exists to prevent.

🔴 WHAT IS STILL NOT ATOMIC, SAID PLAINLY

Two durable systems are involved — the workloop registration (a KV row) and the workflow’s event history — and nothing spans them. The registration is written FIRST because the alternative loses the race above, so a crash between it and the start leaves a sweep-set row for a workflow with no history. That row is not left to alarm forever: the sweep refuses to fire at a workflow with no recorded start and reports the fault, and engine boot reconciliation (WorkloopService::withdraw_unstarted_registrations) withdraws every such row. Boot is the point at which the answer is unambiguous — no start can be in flight across a process boundary — which is why the reconciliation lives there rather than in the sweep, where it would race the birth window it is supposed to tolerate.

Errors

Refuses an unconfigured workloop service, a start payload that is not a JSON object when carry is declared, a duplicate registration, a caller attempting to set aion.kind itself, and propagates start failures.

pub async fn close_workloop_iteration( &self, loop_id: &WorkflowId, close: WorkloopIterationClose, ) -> Result<RunId, EngineError>

Closes the current iteration at the continue-as-new boundary (R3.1): derives one health sample per invariant from the taken routes (R3.3), records IterationClosed + WorkflowContinuedAsNew + the successor generation’s WorkflowStarted in ONE atomic batch through the loop’s Recorder — spawning NO successor process (R13.3) — then installs the produced invariant current-state records with retention pruning (R7/ R8) and feeds the samples into tolerance accounting. Returns the successor generation’s run id.

Errors

Refuses an unregistered loop, a terminal run, pending work, and undeclared invariants; propagates store/append failures.

pub async fn retire_workloop( &self, loop_id: &WorkflowId, reason: String, result: Payload, ) -> Result<(), EngineError>

Retires a workloop (R2.5): records LoopRetired { reason } and its WorkflowCompleted terminal in ONE atomic batch — the declared, recorded way to stop that is not failure — and removes the loop from the sweep set. Invariant current-state records survive indefinitely (R8.1).

Errors

Refuses an unconfigured service and a terminal run; propagates store/append failures.

pub async fn retire_workloop_without_body( &self, loop_id: &WorkflowId, reason: String, result: Payload, ) -> Result<(), EngineError>

Engine::retire_workloop for a loop that declares NO retire body.

Kept as a separate verb rather than a flag on the main one because the difference is a DECLARATION, not a caller preference: a loop whose document declares retire must run it, and a caller must never be able to skip a declared cleanup by passing an argument. The AWL-driven path selects between them from the compiled contract; this exists so an operator retiring a bodyless loop is not forced through an entry probe that would refuse a module which correctly exports nothing.

Errors

As Engine::retire_workloop, minus the retire-body refusals.

pub async fn retire_declared_workloop( &self, loop_id: &WorkflowId, reason: String, result: Payload, ) -> Result<(), EngineError>

Retires a workloop, taking the retire-body decision FROM ITS DEPLOYED DECLARATION rather than from the caller.

🔴 THE OPERATOR SURFACE, AND WHY IT HAS NO “SKIP CLEANUP” FLAG

Engine::retire_workloop and Engine::retire_workloop_without_body are two verbs precisely so a caller cannot choose: a document that declares retire must run it, and letting an argument skip a declared cleanup is how a lease is released twice or a queue is stranded. This is the verb every operator-facing surface calls, and it reads the answer out of the package the loop’s CURRENT generation is pinned to — the same package whose module the body would be spawned from.

A loop whose deployment declares no workloop surface at all (registered through the Rust API rather than deployed from a .awl document) is retired without a body: the engine has no declaration saying there is one, and inventing a retire/1 probe for it would refuse every such loop for lacking an entry it was never meant to export.

Errors

As Engine::retire_workloop, plus catalog and history failures while resolving the loop’s deployed declaration.

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

Starts a DETACHED top-level workflow under the mandatory dedupe identity (R13.1): WorkflowId = hatch_workflow_id(namespace, type, key). Not a child — no lifecycle tie, no supervision edge. A duplicate hatch is a recorded no-op returning the existing workflow id; two racing first-hatches are settled by the store’s optimistic append, the loser resolving to the winner’s workflow.

Errors

Refuses empty/NUL identity parts and propagates start failures other than the dedupe race.

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.

The signal is admitted against the signal type declared by the EXACT package identity the target run is pinned to, BEFORE anything is recorded and before the arrival can be consumed. A refused signal therefore leaves the run’s history byte-identical and the run parked on exactly the wait it was parked on — the refusal reaches the caller, not the run.

Errors

Returns EngineError::SignalRefused when the name is not declared by the target’s package or the payload does not satisfy the declared type (nothing recorded, nothing consumed), 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>, arguments: Payload, ) -> Result<Payload, EngineError>

Dispatch a read-only query, with its arguments, to a live workflow run through the AT seam.

arguments is the type-erased JSON document the workflow’s registered handler decodes; null is the canonical “no arguments” document. Nothing on this path records an event — arguments are inputs to a read-only handler, never history.

Errors

Returns EngineError::Query with crate::query::QueryError::NotRunning when the (workflow, run) pair is durably terminal — or durably started with no live handle to answer from (a workloop’s deferred successor between iterations, a run whose engine is gone): the run EXISTS and is not running, which is a different answer from unknown (#214). EngineError::WorkflowNotFound when no history names the run, crate::query::QueryError::InvalidArguments when arguments is not a well-formed JSON document, 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 fn worker_contracts_for_queue( &self, task_queue: &str, ) -> Result<Vec<DeployedWorkerContract>, EngineError>

Returns every retained .v4 package contract declaring task_queue.

This is the RAW retained set, which under content-hash namespacing holds every coexisting version — including ones nothing can reach any more. Worker admission must not be decided from it directly; use Engine::worker_contracts_for_admission, which splits it into the reachable versions that bind a connection and the unreachable ones that bind nobody. Demanding the raw set of one connection is what made a queue with a single stale version permanently unservable.

Errors

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

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

Returns one read of the task queues the retained contracts declare.

The server’s queue-service classifier (R1) reads this to tell a structurally undeclared queue apart from a declared but unserved one. The answer reports whether it covered every retained entry: a queue missing from a read that could not decode some entry is unknowable, not undeclared. An empty set likewise means this catalog declares no queues at all and therefore cannot contradict any dispatch — see WorkflowCatalog::declared_task_queues.

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 — running, durably paused, or 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.

impl Engine

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

Run the startup recovery steps a deferred build skipped, in the same order build() runs them: active-workflow recovery replay, timer recovery, schedule-coordinator catch-up, schedule recovery.

Call this exactly once, after every seam the activity dispatcher consults is installed — recovery replay re-dispatches in-flight activities through the dispatcher immediately.

Errors

Returns EngineError::StartupRecoveryNotDeferred when the engine was built without crate::EngineBuilder::defer_startup_recovery (its build already ran recovery), EngineError::StartupRecoveryAlreadyRan on a second call, EngineError::StartupRecoverySlotPoisoned when the slot lock was poisoned, and any recovery-step error the deferred steps themselves surface.

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

Run ONLY the active-workflow recovery leg of a deferred build: every durably-active workflow is replayed to residency and registered, so signals, queries, and cancels answer correctly the moment transports serve. The catch-up legs — owed timer fires, schedule-coordinator catch-up, schedule recovery — remain owed to Engine::run_startup_catchup, which a host may run behind already- open doors: an owed-fire backlog has no upper bound, and every fire it delivers goes through the same idempotent record-once path the live timer wheel uses, so serving during catch-up is the steady-state contract, not a special mode.

Errors

Returns EngineError::StartupRecoveryNotDeferred when the engine was built without crate::EngineBuilder::defer_startup_recovery, EngineError::StartupRecoveryAlreadyRan on a second call, EngineError::StartupRecoverySlotPoisoned when the slot lock was poisoned, and any error the workflow-recovery leg itself surfaces.

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

Run the catch-up legs a Engine::recover_workflows_on_startup call left owed, in the order build() runs them: timer recovery (owed fires and future re-arms), schedule-coordinator catch-up, schedule recovery. Safe to run while the host is serving — every leg is the same idempotent machinery the live paths run against a serving engine.

Errors

Returns EngineError::StartupCatchupBeforeWorkflowRecovery when the workflow-recovery leg has not run, EngineError::StartupRecoveryNotDeferred when the build was not deferred, EngineError::StartupRecoveryAlreadyRan on a second call, EngineError::StartupRecoverySlotPoisoned when the slot lock was poisoned, and any error the catch-up legs themselves surface.

Trait Implementations

impl Drop for Engine

fn drop(&mut self)

Close the engine-task epoch when the engine is released, whether or not Engine::shutdown was ever called or ever succeeded.

Without this, an engine dropped without a successful shutdown left completion retries armed and appending terminal events. They could not be stopped by EngineTaskRuntime::drop either: an attempt in flight upgrades its weak reference and holds the RuntimeHandle strongly for the length of the attempt, so the refcount never reaches zero and that backstop is unreachable for precisely the span of the append it exists to stop. This drop runs before the engine’s own fields are released, so it does not depend on that refcount at all.

Closes the engine-task epoch with EngineTaskRuntime::shutdown, whose runtime drop is isolated on a plain joiner thread. That makes joined cleanup safe even when this Drop runs inside a host async context: the epoch is gated, every task is aborted, and the executor’s I/O driver is released before Drop returns. The gate remains load-bearing for an attempt already past an await boundary: its append boundary reads is_epoch_open and refuses.

The visibility reconciliation task is aborted here for the same reason

It runs on the HOST runtime, not the engine-task executor, so the epoch gate does not reach it — and dropping its JoinHandle detaches rather than cancels. It is an unbounded loop holding the event store and the visibility store, and reconcile_visibility WRITES. Left detached, an engine released without shutdown went on upserting visibility rows for the life of the process, against a store a successor engine may already own. Engine::shutdown aborts it as its first act; this does the same, so the two paths agree.

The live timer wheel is disarmed here for the third time, same reason

🔴 THIS WAS MISSING, AND IT LEFT A DURABLE WRITER ARMED. Live-wheel timer tasks are tokio::spawned on the HOST runtime (runtime/nif_timer_bridge.rs), so — exactly like the reconciliation loop — the engine-task epoch gate does not reach them. Their body is fire_wheel_timer, which records a durable TimerFired. They hold a Weak<EngineNifState>, and this drop deliberately does NOT clear the seams (see below), so that upgrade succeeds and the fire proceeds.

An engine released without shutdown therefore kept a durable-append path armed for the life of the process. Engine::shutdown names the consequence precisely: across a failover, the dead owner’s orphaned wheel task races the survivor’s adoption-armed timer and can record the one durable TimerFired first, leaving the survivor’s resident sleeper parked forever. That is the single-writer invariant, and nothing about it cares whether the engine was shut down or dropped.

Safe in a Drop: shutdown_timer_wheel sets a flag and then performs a DashMap drain plus abort() — non-blocking, structurally identical to the visibility_reconciliation_task.abort() above. It therefore remains safe before the joined engine-task shutdown below.

🔴 AND IT IS A GATE, NOT ONLY A DRAIN — which it had to become for this Drop to be worth anything. A drain closes the set of timers armed at one instant; this Drop deliberately leaves the beamr scheduler and the engine seams alive, so a workflow process still runnable could reach sleep a moment later and arm a fresh durable TimerFired writer through a wheel this drop believed it had emptied. arm_timer now refuses once the flag is set (nif_timer_bridge.rs, shut_down), so the guarantee below is a property of the wheel from here on rather than of one instant.

🔴 WHAT THIS DOES NOT DO, STATED SO NOBODY READS MORE INTO IT

It does not clear the engine NIF seams. Those hold Arcs back to the RuntimeHandle, so until clear_engine_seams runs the handle, its beamr scheduler and every store clone they reach outlive this drop. Engine::shutdown clears them only after the scheduler has stopped and the child-task and timer-wheel epochs have closed; none of that has happened here, and a NIF could still read a slot this drop cleared. Trading a scheduler leak for a use-after-clear is the wrong direction, so that leak stands and is named: an engine released without explicit shutdown still holds its scheduler and installed seams. The dedicated engine-task executor is different: it is joined below so its I/O driver cannot accumulate process descriptors. What this drop guarantees for durability is still narrower — no durable writer this drop can reach keeps writing, and no writer it cannot reach can end a run. The first clause covers FOUR BACKGROUND writers, stopped in two different ways:

1. anything armed on the engine-task epoch — shutdown() below;
2. the visibility reconciliation loop — abort() below;
3. the live timer wheel — shutdown_timer_wheel() below, which gates and drains, and refuses at the point of writing, because abort cannot stop a task already inside a poll. That refusal is in TWO places, not one, and the second is easy to miss: an ordinary timer is refused at the bridge’s append boundary (nif_timer_bridge.rs, record_workflow_event), but a reserved deadline:{run} fire never reaches that boundary — fire_timer_guarded demuxes it to the deadline handler first — so it is refused inside crate::lifecycle::deadline::WorkflowDeadlineHandler instead, off the same latch;
4. the activity completion / retry task ([crate::runtime::nif_activity_retry_dispatch::spawn_completion_task]), which this drop cannot reach at all: its JoinHandle is discarded, so it is detached on the host runtime and nothing here registers or aborts it. It is stopped instead at its append boundary, which reads is_epoch_open() under the recorder lock — so step 1’s the engine-task epoch closure is what silences it, one indirection away.

🔴 AND THERE IS A FIFTH, WHICH IS NOT A BACKGROUND WRITER AT ALL

The four above are things the engine spawned; this drop stops them because it can reach them. The fifth is the workflow process itself, and this drop deliberately does not stop it — it leaves the beamr scheduler running and the NIF seams installed, which is exactly what the section above says it is trading for. A still-runnable workflow process therefore keeps calling NIFs after the Engine is gone, and 13 of the 24 registered engine NIFs perform durable writes — dispatch_activity, dispatch_activity_in_vm, await_activity_result, sleep, start_timer, cancel_timer, with_timeout, continue_as_new, send_signal, spawn_child, collect_all, collect_race, collect_map. The other 11 read or reply and record nothing. The registration table is runtime::engine_nifs::engine_nif_entries, whose own test asserts the total, so both halves of that split are checkable against a closed set rather than taken on trust — which is the point, since the first draft of this paragraph carried a transposed count. None of the 13 consults the engine-task epoch, and nothing in the append path does either: NifContext::block_on_recorder takes the recorder lock and nothing else, and Recorder::append_one goes straight to store.append.

An earlier revision of this doc said “there are FOUR” full stop, and was wrong in the way that matters most: it did not omit an obscure writer, it omitted the one that executes user code.

What has been closed is the part that can END A RUN. WorkflowContinuedAsNew is a TERMINAL, it was the ONE terminal this fifth writer could still record, and it is now refused off the same epoch (runtime::nif_continue_as_new::record_continuation). The reason it had to be, in one line: the successor run that terminal obliges was already refused at completion::start_continuation_replacement, so the two halves of one transition disagreed and the run was left terminal with no continuation. Every other terminal reachable from workflow code was already gated — process exit at the completion append boundary, WorkflowTimedOut off the timer bridge’s stand-down latch.

And the refusal ENDS THE PROCESS, which is the half that makes it a gain rather than a trade. Before the gate, the recorder call either succeeded or aborted the NIF, and the success path always reached cancel_pid — that instruction is where this fifth writer died. A refusal that merely returned early would have removed it, leaving the process runnable and free to make every ungated write listed below. So runtime::nif_continue_as_new terminates on the epoch refusal too — and, of the refusals, on that one ONLY. A pre-terminal store fault is an ordinary error workflow code may handle, and killing a process for it would turn a transient blip into a dead run; an already-terminal run is spared for a different reason — its terminal was recorded by a seam that owns its own teardown, and of those owners some end the pid (a second cancel_pid from here would race them) while some only deregister (a kill from here would usurp them). The predicate’s doc carries that split; the “Five ordinary terminal paths” paragraph in lifecycle/completion.rs carries the one enumeration of the owners. It ALSO terminates whenever the terminal actually landed, including the half-completed case where the terminal is durable but the deadline retirement that follows it failed — because the question that decides this is “did the terminal land”, not “was there an error”. The predicate is outcome_must_end_the_process, pinned by a test with both negative controls.

The cost, stated because it is not zero: the refusal returns before retire_run_deadline, so the predecessor’s deadline row stays armed. A restart gap longer than the run’s remaining budget times the run out instead of continuing it. That is the same exposure every other in-flight run already carries across an outage; the old path escaped it only by recording a terminal for a transition that never completed.

🔴 WHAT IS STILL OPEN, AND WHY IT IS NOT CLOSED HERE

A workflow process refused by the EPOCH gate is now stopped, so the writes below are not reachable from that path. Say “the epoch gate” and not “was refused”: the other refusals deliberately leave the process alive, so a reader who takes this sentence at its widest reading would believe an exposure is closed that is open by design.

They remain fully open on every other path — a process that never calls continue_as_new is untouched by any of this and keeps writing.

The fifth writer’s NON-terminal durable writes are ungated and remain so: TimerStarted plus a durable timer row (sleep, start_timer, with_timeout — TimerService::schedule writes the row and only then arms, so the wheel’s refusal lands after both), activity schedule/start and completion records, spawn_child’s whole child-start chain, and send_signal, which writes into a THIRD workflow’s history.

Two things bound that, and neither is what a reader might assume:

• WriteToken fences NOTHING. It is a zero-sized marker with a public recorder() constructor and no engine, epoch, lease or node identity; two engines over one store both mint valid ones. Its own doc says so — it exists to stop an Arc<dyn EventStore> alone being write authority.
• SequenceConflict catches only the LOSER of a head race, and a released engine is structurally positioned to be the winner: its Recorder is the one already at the current head, because it is the one that has been appending. If it writes first, its write succeeds and the SUCCESSOR takes the conflict.

So the remaining exposure is real and is stated rather than denied. It is not closed here because no flag in this crate distinguishes “released” from “shutting down” — begin_close sets one bit and both Engine::drop and Engine::shutdown set it. A gate on that bit at a workflow-process write path would therefore also fire during an ORDINARY graceful shutdown, for the whole unbounded span between begin_close() and runtime.shutdown() further down this file, and there the failure is an {error, _} returned INSIDE running workflow code — a failed sleep, a failed spawn_child — on runs the shutdown was trying to leave intact. The terminal was worth that trade because its successor was already refused at start_continuation_replacement: recording it could only produce a run that is terminal with no continuation.

⚠️ Refusing it is not free, and an earlier revision of this sentence said it was. It read “refusing cost nothing that was not already lost”, which is the exact claim runtime::nif_continue_as_new’s own documentation exists to retract — and which the “cost, stated because it is not zero” paragraph above already contradicts. The price is stated there and holds here: the refusal returns before retire_run_deadline, so the predecessor’s deadline stays armed and a long enough outage times the run out instead of continuing it. What makes the trade worth taking is not that it is free but that the alternative bought its exemption with a false terminal.

Refusing ordinary progress is a different bargain and needs a latch that means what it says. Do not add one of these gates without adding that latch.

🔴 THAT LIST IS A CLAIM ABOUT DURABLE WRITERS AND IT IS ONLY AS GOOD AS ITS ENUMERATION — four times proven. An earlier revision named two and was wrong: the timer wheel was the third, and it was armed. The revision after that named three and was also wrong: the completion task was the fourth, it had no epoch check of any kind, and it sleeps an SDK-declared backoff with no ceiling between attempts. And the revision after THAT — the one that added the wheel’s append-boundary refusal — wrote entry 3 as though that boundary covered the whole wheel, when the deadline path is demuxed away before it and had no refusal at all: an engine released without shutdown could still record a durable WorkflowTimedOut and tear a run down. The enumeration was right and the mechanism named under it was not, which is the harder failure to see, because the list looked complete.

And the FOURTH time is the section above: every revision so far had enumerated only what this drop reaches, and then written a guarantee over every writer that exists. The workflow process is not on any list of things a Recorder grep or a spawn grep produces, because nobody spawned it here and it holds no handle this file can see — it is reached through an installed NIF seam by code the operator wrote. A search shaped like the mechanism you already know will not find the writer you do not. That is why the method below now starts from the NIF registration table, which is a closed set that something asserts the size of, rather than from a grep whose completeness nothing checks.

The way to check this list is: take runtime::engine_nifs::engine_nif_entries and account for every entry; grep the crate for every construction of a Recorder handle and every detached spawn; and then, for each writer either search yields, follow the ACTUAL route from the wake to the append and confirm the named gate sits on it. Not to re-read this sentence and find it plausible.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.

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.

fn record_redelivered_timer_fire( &self, workflow_id: &WorkflowId, timer_id: &TimerId, ) -> Result<RedeliveredFire, EngineSeamError>

Answers, under the recorder lock, whether a REDELIVERY of an already-recorded fire for timer_id still owes its mailbox wake.