vertep_runtime/
actor.rs

1//! The actor contract and its adapters for charts, async functions and streams.
2//!
3//! An actor receives SCXML events one at a time and produces zero or more outgoing events. A
4//! running chart is one implementation ([`ChartActor`]); an async function ([`FnActor`]) and a
5//! stream source ([`StreamActor`]) are others, so all of them are interchangeable as targets of
6//! `<send>` and `<invoke>`. A turn is either finished at once ([`Turn::Ready`], a chart's
7//! macrostep) or a `Send` future the runtime runs in the background ([`Turn::Pending`]); in both
8//! cases the actor takes no other input until the turn is over.
9
10use std::sync::Arc;
11
12use futures::StreamExt;
13use futures::lock::Mutex as AsyncMutex;
14use futures::stream::BoxStream;
15use vertep_core::datamodel::{
16    Datamodel, DatamodelRegistry, Environment, Event, EventType, SessionInfo, Value,
17};
18use vertep_core::session::{self, Effect, Input, StartInput};
19
20use crate::address::Address;
21use crate::chart::ChartInfo;
22use crate::shell::BoxFuture;
23
24/// The event name of the input that starts an actor.
25pub const START_EVENT: &str = "vt.start";
26/// The event a stream actor sends itself to pull its next item.
27pub const STREAM_NEXT_EVENT: &str = "vt.stream.next";
28
29/// One input to an actor's turn.
30#[derive(Clone, Debug, PartialEq)]
31pub struct ActorInput {
32    /// The event.
33    pub event: Event,
34    /// The message id it arrived with.
35    pub message_id: String,
36    /// The time stamped when it was dequeued, in milliseconds since the epoch.
37    pub received_at_ms: i64,
38}
39
40/// Where an outgoing event goes.
41#[derive(Clone, Debug, PartialEq, Eq)]
42pub enum OutTarget {
43    /// The invoking parent (`#_parent`).
44    Parent,
45    /// The origin of the input being handled (a reply).
46    Origin,
47    /// The actor itself.
48    Itself,
49    /// An address.
50    Address(Address),
51    /// A target text that is not an address; delivery fails with `error.communication`.
52    Invalid(String),
53}
54
55/// One outgoing event.
56#[derive(Clone, Debug, PartialEq)]
57pub struct Outgoing {
58    /// Where it goes.
59    pub target: OutTarget,
60    /// The event.
61    pub event: Event,
62    /// Delay in milliseconds.
63    pub delay_ms: u64,
64}
65
66impl Outgoing {
67    /// `name` with `data` to `target`, now.
68    pub fn new(target: OutTarget, name: impl Into<String>, data: Value) -> Self {
69        Outgoing {
70            target,
71            event: Event {
72                name: name.into(),
73                data,
74                ..Event::default()
75            },
76            delay_ms: 0,
77        }
78    }
79}
80
81/// What a turn produced.
82#[derive(Clone, Debug, Default, PartialEq)]
83pub struct TurnOutput {
84    /// Outgoing events.
85    pub sends: Vec<Outgoing>,
86    /// Send ids of delayed sends to cancel.
87    pub cancels: Vec<String>,
88    /// Set when the actor completed, with its done data.
89    pub done: Option<Value>,
90}
91
92/// A failed turn: an error returned by the actor or a caught panic.
93#[derive(Clone, Debug, PartialEq, Eq)]
94pub struct ActorFailure(pub String);
95
96impl core::fmt::Display for ActorFailure {
97    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
98        write!(f, "actor failed: {}", self.0)
99    }
100}
101
102impl std::error::Error for ActorFailure {}
103
104/// The result of starting a turn.
105pub enum Turn {
106    /// The turn is complete.
107    Ready(Result<TurnOutput, ActorFailure>),
108    /// The turn completes when this future does.
109    Pending(BoxFuture<Result<TurnOutput, ActorFailure>>),
110}
111
112impl std::fmt::Debug for Turn {
113    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
114        match self {
115            Turn::Ready(r) => f.debug_tuple("Ready").field(r).finish(),
116            Turn::Pending(_) => f.write_str("Pending"),
117        }
118    }
119}
120
121/// The actor contract: one event in, zero or more events out, one turn at a time.
122// @tg: implements:: [[openspec:actor-model#Actor Trait]]
123pub trait Actor {
124    /// The actor's address.
125    fn address(&self) -> &Address;
126    /// Handles one input.
127    fn turn(&mut self, input: ActorInput) -> Turn;
128}
129
130/// How an actor kind is started by `<invoke>`.
131#[derive(Clone, Debug, PartialEq)]
132pub struct ActorStart {
133    /// The child's address.
134    pub address: Address,
135    /// The parent's address.
136    pub parent: Address,
137    /// The invoke id.
138    pub invokeid: String,
139    /// The evaluated `src`.
140    pub src: Option<String>,
141    /// The evaluated inline content.
142    pub content: Option<Value>,
143    /// The evaluated `namelist` and `<param>` values.
144    pub params: Value,
145}
146
147/// Creates an actor of one kind for an `<invoke>`.
148pub type ActorFactory =
149    Arc<dyn Fn(&ActorStart) -> Result<Box<dyn Actor>, ActorFailure> + Send + Sync>;
150
151/// The function behind a [`FnActor`]: one input in, a turn's output out.
152pub type ActorFn =
153    Arc<dyn Fn(ActorInput) -> BoxFuture<Result<TurnOutput, ActorFailure>> + Send + Sync>;
154
155/// An actor that runs an async function per input.
156pub struct FnActor {
157    address: Address,
158    f: ActorFn,
159}
160
161impl std::fmt::Debug for FnActor {
162    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
163        f.debug_struct("FnActor")
164            .field("address", &self.address)
165            .finish_non_exhaustive()
166    }
167}
168
169impl FnActor {
170    /// An actor at `address` running `f` for every input.
171    pub fn new(address: Address, f: ActorFn) -> Self {
172        FnActor { address, f }
173    }
174
175    /// The factory of an invokable async function: on start it runs `f` on the invoke's
176    /// parameters, and its result is the child's done data (`done.invoke.<id>`); an error is a
177    /// failure, subject to `vt:restart`.
178    // @tg: implements:: [[openspec:event-journal#External Results As Inputs]]
179    pub fn invokable<F, Fut>(f: F) -> ActorFactory
180    where
181        F: Fn(Value) -> Fut + Send + Sync + 'static,
182        Fut: std::future::Future<Output = Result<Value, String>> + Send + 'static,
183    {
184        let f = Arc::new(f);
185        Arc::new(move |start: &ActorStart| {
186            let f = f.clone();
187            let params = start.params.clone();
188            let run: ActorFn = Arc::new(move |input: ActorInput| {
189                let f = f.clone();
190                let params = params.clone();
191                Box::pin(async move {
192                    if input.event.name != START_EVENT {
193                        return Ok(TurnOutput::default());
194                    }
195                    match f(params).await {
196                        Ok(v) => Ok(TurnOutput {
197                            done: Some(v),
198                            ..TurnOutput::default()
199                        }),
200                        Err(e) => Err(ActorFailure(e)),
201                    }
202                })
203            });
204            Ok(Box::new(FnActor::new(start.address.clone(), run)) as Box<dyn Actor>)
205        })
206    }
207
208    /// A request handler for a stateless worker: `f` gets each event and its result, when
209    /// there is one, is sent back to the event's origin as `vt:reply`.
210    pub fn handler<F, Fut>(f: F) -> ActorFn
211    where
212        F: Fn(Event) -> Fut + Send + Sync + 'static,
213        Fut: std::future::Future<Output = Result<Option<Value>, String>> + Send + 'static,
214    {
215        let f = Arc::new(f);
216        Arc::new(move |input: ActorInput| {
217            let f = f.clone();
218            Box::pin(async move {
219                match f(input.event).await {
220                    Ok(Some(v)) => Ok(TurnOutput {
221                        sends: vec![Outgoing::new(
222                            OutTarget::Origin,
223                            crate::vocab::REPLY_EVENT,
224                            v,
225                        )],
226                        ..TurnOutput::default()
227                    }),
228                    Ok(None) => Ok(TurnOutput::default()),
229                    Err(e) => Err(ActorFailure(e)),
230                }
231            })
232        })
233    }
234}
235
236impl Actor for FnActor {
237    fn address(&self) -> &Address {
238        &self.address
239    }
240
241    fn turn(&mut self, input: ActorInput) -> Turn {
242        Turn::Pending((self.f)(input))
243    }
244}
245
246/// The stream behind a [`StreamActor`].
247pub type ActorStream = BoxStream<'static, Result<Value, String>>;
248
249/// An actor that forwards the items of a stream to its parent as events named `item_event`,
250/// one per turn, and completes when the stream ends.
251pub struct StreamActor {
252    address: Address,
253    item_event: String,
254    stream: Arc<AsyncMutex<ActorStream>>,
255}
256
257impl std::fmt::Debug for StreamActor {
258    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
259        f.debug_struct("StreamActor")
260            .field("address", &self.address)
261            .field("item_event", &self.item_event)
262            .finish_non_exhaustive()
263    }
264}
265
266impl StreamActor {
267    /// An actor at `address` forwarding `stream`.
268    pub fn new(address: Address, item_event: impl Into<String>, stream: ActorStream) -> Self {
269        StreamActor {
270            address,
271            item_event: item_event.into(),
272            stream: Arc::new(AsyncMutex::new(stream)),
273        }
274    }
275
276    /// The factory of an invokable stream: `f` opens the stream from the invoke's parameters.
277    pub fn invokable<F>(item_event: &str, f: F) -> ActorFactory
278    where
279        F: Fn(Value) -> ActorStream + Send + Sync + 'static,
280    {
281        let item_event = item_event.to_owned();
282        Arc::new(move |start: &ActorStart| {
283            Ok(Box::new(StreamActor::new(
284                start.address.clone(),
285                item_event.clone(),
286                f(start.params.clone()),
287            )) as Box<dyn Actor>)
288        })
289    }
290}
291
292impl Actor for StreamActor {
293    fn address(&self) -> &Address {
294        &self.address
295    }
296
297    fn turn(&mut self, input: ActorInput) -> Turn {
298        if input.event.name != START_EVENT && input.event.name != STREAM_NEXT_EVENT {
299            return Turn::Ready(Ok(TurnOutput::default()));
300        }
301        let stream = self.stream.clone();
302        let item_event = self.item_event.clone();
303        Turn::Pending(Box::pin(async move {
304            let next = stream.lock().await.next().await;
305            match next {
306                Some(Ok(v)) => Ok(TurnOutput {
307                    sends: vec![
308                        Outgoing::new(OutTarget::Parent, item_event, v),
309                        Outgoing::new(OutTarget::Itself, STREAM_NEXT_EVENT, Value::Undefined),
310                    ],
311                    ..TurnOutput::default()
312                }),
313                Some(Err(e)) => Err(ActorFailure(e)),
314                None => Ok(TurnOutput {
315                    done: Some(Value::Undefined),
316                    ..TurnOutput::default()
317                }),
318            }
319        }))
320    }
321}
322
323/// A chart run as an actor without a journal: an invoked chart child, whose inputs are
324/// journaled by its parent.
325pub struct ChartActor {
326    address: Address,
327    info: Arc<ChartInfo>,
328    registry: Arc<dyn DatamodelRegistry + Send + Sync>,
329    session: Option<session::Session>,
330    dm: Option<Box<dyn Datamodel>>,
331    start_data: Vec<(String, Value)>,
332    ioprocessors: Value,
333}
334
335impl std::fmt::Debug for ChartActor {
336    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
337        f.debug_struct("ChartActor")
338            .field("address", &self.address)
339            .field("version", &self.info.version.as_str())
340            .finish_non_exhaustive()
341    }
342}
343
344impl ChartActor {
345    /// A chart actor at `address` running `info`; it starts on its first input
346    /// ([`START_EVENT`]) with `start_data` for its top-level data items.
347    pub fn new(
348        address: Address,
349        info: Arc<ChartInfo>,
350        registry: Arc<dyn DatamodelRegistry + Send + Sync>,
351        start_data: Vec<(String, Value)>,
352        ioprocessors: Value,
353    ) -> Self {
354        ChartActor {
355            address,
356            info,
357            registry,
358            session: None,
359            dm: None,
360            start_data,
361            ioprocessors,
362        }
363    }
364
365    /// The active state ids, once started.
366    pub fn active(&self) -> Vec<String> {
367        self.session
368            .as_ref()
369            .map(|s| self.info.active(s))
370            .unwrap_or_default()
371    }
372
373    fn output(&self, effects: Vec<Effect>) -> TurnOutput {
374        let mut out = TurnOutput::default();
375        for e in effects {
376            match e {
377                Effect::Send(r) => {
378                    let target = match r.target.as_deref() {
379                        None | Some("#_internal") => OutTarget::Itself,
380                        Some("#_parent") => OutTarget::Parent,
381                        Some(t) => match t.strip_prefix("#_scxml_") {
382                            Some(sid) => Address::parse(sid)
383                                .map(OutTarget::Address)
384                                .unwrap_or_else(|_| OutTarget::Invalid(t.to_owned())),
385                            None => Address::parse(t)
386                                .map(OutTarget::Address)
387                                .unwrap_or_else(|_| OutTarget::Invalid(t.to_owned())),
388                        },
389                    };
390                    let mut event = r.event;
391                    if event.sendid.is_none() {
392                        event.sendid = Some(r.sendid);
393                    }
394                    out.sends.push(Outgoing {
395                        target,
396                        event,
397                        delay_ms: r.delay_ms,
398                    });
399                }
400                Effect::Cancel { sendid } => out.cancels.push(sendid),
401                Effect::SessionDone { data, cancelled } if !cancelled => out.done = Some(data),
402                _ => {}
403            }
404        }
405        out
406    }
407}
408
409impl Actor for ChartActor {
410    fn address(&self) -> &Address {
411        &self.address
412    }
413
414    fn turn(&mut self, input: ActorInput) -> Turn {
415        let env = Environment {
416            now_ms: input.received_at_ms,
417            seed: crate::address::seed_of(self.address.node()),
418        };
419        let result = if let Some(s) = self.session.take() {
420            let Some(dm) = self.dm.as_mut() else {
421                return Turn::Ready(Err(ActorFailure("chart actor has no datamodel".into())));
422            };
423            if input.event.name == START_EVENT {
424                self.session = Some(s);
425                return Turn::Ready(Ok(TurnOutput::default()));
426            }
427            let event = vertep_core::datamodel::admit(dm.as_ref(), input.event);
428            session::step(
429                &self.info.compiled,
430                s,
431                Input::event(event, env),
432                dm.as_mut(),
433            )
434            .map_err(|e| ActorFailure(format!("{e:?}")))
435        } else {
436            let mut dm = match vertep_core::null::create_datamodel_for(
437                self.registry.as_ref(),
438                self.info.compiled.chart(),
439            ) {
440                Ok(d) => d,
441                Err(e) => return Turn::Ready(Err(ActorFailure(format!("{e:?}")))),
442            };
443            let mut options = session::Options::default();
444            options
445                .send_types
446                .extend(self.info.vt_send_types.iter().cloned());
447            let start = StartInput {
448                info: SessionInfo {
449                    sessionid: self.address.as_str().to_owned(),
450                    name: self.info.ir.name.clone(),
451                    ioprocessors: self.ioprocessors.clone(),
452                },
453                env,
454                options,
455                data: self.start_data.clone(),
456                sessions: None,
457            };
458            let r = session::start(&self.info.compiled, start, dm.as_mut())
459                .map_err(|e| ActorFailure(format!("{e:?}")));
460            self.dm = Some(dm);
461            r
462        };
463        Turn::Ready(result.map(|o| {
464            self.session = Some(o.session);
465            self.output(o.effects)
466        }))
467    }
468}
469
470/// A platform event named `name` with `data`.
471pub fn platform_event(name: &str, data: Value) -> Event {
472    Event {
473        name: name.to_owned(),
474        type_: EventType::Platform,
475        data,
476        ..Event::default()
477    }
478}
vertep_runtime/actor.rs

vertep_runtime/
actor.rs
