Skip to main content

ActionBackend

app_kit::workflow

Trait ActionBackend 

Source 
pub trait ActionBackend:
    Send
    + Sync
    + 'static {
    type Stream: Stream<Item = Result<ActionEvent, AppError>> + 'static;

    // Required methods
    fn declaration(&self) -> &ActionDeclaration;
    fn start(&self, request: ActionRequest) -> Self::Stream;
}
Expand description

A caller-owned action whose lazy source produces results and explicit writes.

The declaration remains stable for its logical key and revision. App retains the exact implementation without requiring Clone or normalizing it through an erased handle. The returned source begins external work only when polled.

An explicit source error is an action failure, including AppError::Integration. A captured construction or polling panic is an execution failure; an App activation deadline is a timeout failure. EOF without a completed result succeeds with no output. Before storage commit admission, these outcomes, owning drop and abort discard staged workflow state and preserve the prior action position. None automatically retries.

The caller owns the source’s internal I/O, cancellation, independently spawned tasks and external reconciliation. Workflow-state rollback does not undo those effects; a later explicit execution can repeat them. App does not infer an Unknown effect merely from polling the source. Separately recorded App-owned lower-operation or admitted-storage uncertainty retains its own reconciliation authority after source observation ends.

§Example

use app_kit::{AppError, workflow::{ActionBackend, ActionDeclaration, ActionEvent,
    ActionMode, ActionRequest, StateChange}};
use serde_json::{Map, Value};
struct Increment { declaration: ActionDeclaration }
impl ActionBackend for Increment {
    type Stream = futures_util::stream::Iter<std::vec::IntoIter<Result<ActionEvent, AppError>>>;
    fn declaration(&self) -> &ActionDeclaration { &self.declaration }
    fn start(&self, request: ActionRequest) -> Self::Stream {
        let count = request.state.values()["count"].as_i64().unwrap() + 1;
        futures_util::stream::iter(vec![Ok(ActionEvent::Completed {
            result: Map::from_iter([("new_count".into(), Value::from(count))]),
            writes: vec![StateChange::Increment { values: vec![("count".into(), Value::from(1))] }],
        })])
    }
}
let action = Increment { declaration: ActionDeclaration {
    key: "increment".into(), revision: "r1".into(), reads: vec!["count".into()],
    writes: vec!["count".into()], required_inputs: Vec::new(), optional_inputs: Vec::new(),
    bound_inputs: Map::new(), default_inputs: Map::new(), tags: vec!["counter".into()],
    source: Some("increment count by one".into()), mode: ActionMode::Both,
} };
use app_kit::workflow::{ActionContext, WorkflowState};
let request = ActionRequest {
    state: WorkflowState::new(Map::from_iter([("count".into(), Value::from(2))]))?,
    inputs: Map::new(), context: ActionContext {
    workspace: "appws_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?,
    workflow: "counter".parse()?,
    run: "run_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?,
    occurrence: "00000000000040008000000000000001".into(),
    checkpoint: 0, action_ordinal: 0, node: "increment".into(), attempt: 0,
    mode: ActionMode::Asynchronous, parent: None, parent_occurrence: None,
},
};
use futures_util::StreamExt;
let mut source = action.start(request);
let ActionEvent::Completed { result, writes } = source.next().await.unwrap()? else {
    panic!("increment emits a completed result");
};
assert_eq!(result["new_count"], 3);
assert_eq!(writes.len(), 1);

Required Associated Types§

Source

type Stream: Stream<Item = Result<ActionEvent, AppError>> + 'static

The exact caller source, which may be local to one task and need not be Send.

App stops polling at the first Completed event or error. A source that ends without either leaves the attempted action uncommitted.

Required Methods§

Source

fn declaration(&self) -> &ActionDeclaration

Returns the stable declaration validated when App binds the inventory.

§Example
use app_kit::{AppError, workflow::{ActionBackend, ActionDeclaration, ActionEvent,
    ActionMode, ActionRequest, StateChange}};
use serde_json::{Map, Value};
struct Increment { declaration: ActionDeclaration }
impl ActionBackend for Increment {
    type Stream = futures_util::stream::Iter<std::vec::IntoIter<Result<ActionEvent, AppError>>>;
    fn declaration(&self) -> &ActionDeclaration { &self.declaration }
    fn start(&self, request: ActionRequest) -> Self::Stream {
        let count = request.state.values()["count"].as_i64().unwrap() + 1;
        futures_util::stream::iter(vec![Ok(ActionEvent::Completed {
            result: Map::from_iter([("new_count".into(), Value::from(count))]),
            writes: vec![StateChange::Increment { values: vec![("count".into(), Value::from(1))] }],
        })])
    }
}
let action = Increment { declaration: ActionDeclaration {
    key: "increment".into(), revision: "r1".into(), reads: vec!["count".into()],
    writes: vec!["count".into()], required_inputs: Vec::new(), optional_inputs: Vec::new(),
    bound_inputs: Map::new(), default_inputs: Map::new(), tags: vec!["counter".into()],
    source: Some("increment count by one".into()), mode: ActionMode::Both,
} };
assert_eq!(action.declaration().reads, ["count"]);
assert_eq!(action.declaration().revision, "r1");
Source

fn start(&self, request: ActionRequest) -> Self::Stream

Constructs the source for one attempt without performing external work.

The source may be local and non-Send. Construct it without starting external I/O: the first poll accepts that work. Emit partial results for observation and one Completed result with explicit state writes; EOF before Completed commits no action. The action owns cancellation and reconciliation of its own I/O and spawned tasks. State rollback cannot undo caller external effects.

§Example
use app_kit::{AppError, workflow::{ActionBackend, ActionDeclaration, ActionEvent,
    ActionMode, ActionRequest, StateChange}};
use serde_json::{Map, Value};
struct Increment { declaration: ActionDeclaration }
impl ActionBackend for Increment {
    type Stream = futures_util::stream::Iter<std::vec::IntoIter<Result<ActionEvent, AppError>>>;
    fn declaration(&self) -> &ActionDeclaration { &self.declaration }
    fn start(&self, request: ActionRequest) -> Self::Stream {
        let count = request.state.values()["count"].as_i64().unwrap() + 1;
        futures_util::stream::iter(vec![Ok(ActionEvent::Completed {
            result: Map::from_iter([("new_count".into(), Value::from(count))]),
            writes: vec![StateChange::Increment { values: vec![("count".into(), Value::from(1))] }],
        })])
    }
}
let action = Increment { declaration: ActionDeclaration {
    key: "increment".into(), revision: "r1".into(), reads: vec!["count".into()],
    writes: vec!["count".into()], required_inputs: Vec::new(), optional_inputs: Vec::new(),
    bound_inputs: Map::new(), default_inputs: Map::new(), tags: vec!["counter".into()],
    source: Some("increment count by one".into()), mode: ActionMode::Both,
} };
use app_kit::workflow::{ActionContext, WorkflowState};
let request = ActionRequest {
    state: WorkflowState::new(Map::from_iter([("count".into(), Value::from(2))]))?,
    inputs: Map::new(), context: ActionContext {
    workspace: "appws_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?,
    workflow: "counter".parse()?,
    run: "run_8a7c251e-b5b8-4d9e-8f0b-123456789abc".parse()?,
    occurrence: "00000000000040008000000000000001".into(),
    checkpoint: 0, action_ordinal: 0, node: "increment".into(), attempt: 0,
    mode: ActionMode::Asynchronous, parent: None, parent_occurrence: None,
},
};
use futures_util::StreamExt;
let mut source = action.start(request);
let ActionEvent::Completed { result, writes } = source.next().await.unwrap()? else {
    panic!("increment emits a completed result");
};
assert_eq!(result["new_count"], 3);
assert_eq!(writes.len(), 1);

Dyn Compatibility§

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors§