a3s_effect/
actor.rs

1//! An actor is a fold over an immutable log.
2//!
3//! Each component is a Moore machine: `step` consumes one fact, `output`
4//! returns the view and the transitions that state enables. The runtime is
5//! the edge that runs those transitions. A transition that has already left
6//! a `cause` in the log is not selected again, so resuming a thread continues
7//! unfinished work and does not repeat finished work.
8
9use std::sync::Arc;
10
11use crate::effect::{Effect, TraceEvent};
12use crate::error::ActorError;
13use crate::exit::Exit;
14use crate::fact::{cut_log, Fact, LogCut, NewFact};
15use crate::log::LogStore;
16
17pub struct Transition<S> {
18    pub key: String,
19    pub run: Effect<Vec<NewFact>, ActorError, S>,
20}
21
22pub struct ErasedComponent<S, V> {
23    pub(crate) project:
24        Arc<dyn Fn(&[Fact]) -> Result<(V, Vec<Transition<S>>), ActorError> + Send + Sync>,
25}
26
27pub fn component<S, V, St, I, Step, Out>(
28    initial: I,
29    step: Step,
30    output: Out,
31) -> ErasedComponent<S, V>
32where
33    S: Send + Sync + 'static,
34    V: 'static,
35    St: Send + 'static,
36    I: Fn() -> St + Send + Sync + 'static,
37    Step: Fn(St, &Fact) -> St + Send + Sync + 'static,
38    Out: Fn(&St) -> (V, Vec<Transition<S>>) + Send + Sync + 'static,
39{
40    ErasedComponent {
41        project: Arc::new(move |facts| {
42            let mut state = initial();
43            for fact in facts {
44                state = step(state, fact);
45            }
46            Ok(output(&state))
47        }),
48    }
49}
50
51pub struct Actor<S, V> {
52    pub name: &'static str,
53    components: Vec<ErasedComponent<S, V>>,
54    merge: Arc<dyn Fn(Vec<V>) -> V + Send + Sync>,
55}
56
57impl<S, V> Actor<S, V>
58where
59    S: Send + Sync + 'static,
60    V: 'static,
61{
62    pub fn new(
63        name: &'static str,
64        components: Vec<ErasedComponent<S, V>>,
65        merge: impl Fn(Vec<V>) -> V + Send + Sync + 'static,
66    ) -> Self {
67        Self {
68            name,
69            components,
70            merge: Arc::new(merge),
71        }
72    }
73
74    fn project(&self, facts: &[Fact]) -> Result<(V, Vec<Transition<S>>), ActorError> {
75        let mut views = Vec::with_capacity(self.components.len());
76        let mut transitions = Vec::new();
77        for component in &self.components {
78            let (view, enabled) = (component.project)(facts)?;
79            views.push(view);
80            transitions.extend(enabled);
81        }
82        let mut seen = std::collections::BTreeSet::new();
83        for transition in &transitions {
84            if !seen.insert(transition.key.clone()) {
85                return Err(ActorError::DuplicateTransition {
86                    key: transition.key.clone(),
87                });
88            }
89        }
90        let causes: std::collections::BTreeSet<&str> = facts
91            .iter()
92            .filter_map(|fact| fact.cause.as_deref())
93            .collect();
94        transitions.retain(|transition| !causes.contains(transition.key.as_str()));
95        transitions.sort_by(|left, right| left.key.cmp(&right.key));
96        Ok(((self.merge)(views), transitions))
97    }
98}
99
100pub struct Settlement<V> {
101    pub view: V,
102    pub log: Vec<Fact>,
103    pub steps: u32,
104    pub cut: LogCut,
105    pub trace: Vec<TraceEvent>,
106}
107
108pub async fn resume<S, V>(
109    actor: &Actor<S, V>,
110    log: &dyn LogStore,
111    services: Arc<S>,
112    thread_id: &str,
113    limit: u32,
114) -> Result<Settlement<V>, Exit<ActorError>>
115where
116    S: Send + Sync + 'static,
117    V: 'static,
118{
119    let mut steps = 0;
120    let mut trace_events = Vec::new();
121    loop {
122        let facts = match log.read(thread_id) {
123            Ok(facts) => facts,
124            Err(error) => return Err(Exit::Fail(error)),
125        };
126        let (view, transitions) = match actor.project(&facts) {
127            Ok(projected) => projected,
128            Err(error) => return Err(Exit::Fail(error)),
129        };
130        let Some(next) = transitions.into_iter().next() else {
131            return Ok(Settlement {
132                cut: cut_log(&facts),
133                view,
134                log: facts,
135                steps,
136                trace: trace_events,
137            });
138        };
139        if steps >= limit {
140            return Err(Exit::Fail(ActorError::StepLimit { limit }));
141        }
142        let (result, span) = next
143            .run
144            .with_span(format!("transition:{}", next.key))
145            .run(Arc::clone(&services))
146            .await;
147        trace_events.extend(span);
148        let produced = match result {
149            Ok(produced) => produced,
150            Err(error) => return Err(error),
151        };
152        if let Err(error) = log.append(thread_id, &produced, Some(next.key.as_str())) {
153            return Err(Exit::Fail(error));
154        }
155        steps += 1;
156    }
157}
158
159pub async fn ingest<S, V>(
160    actor: &Actor<S, V>,
161    log: &dyn LogStore,
162    services: Arc<S>,
163    thread_id: &str,
164    fact: NewFact,
165    limit: u32,
166) -> Result<Settlement<V>, Exit<ActorError>>
167where
168    S: Send + Sync + 'static,
169    V: 'static,
170{
171    log.append(thread_id, &[fact], None).map_err(Exit::Fail)?;
172    resume(actor, log, services, thread_id, limit).await
173}
a3s_effect/actor.rs

a3s_effect/
actor.rs
