use crate::cel::{self, CelError, Env};
use crate::machine::{HistoryKind, Machine, NodeId, StateDef, StateKind};
use crate::model::Action;
use crate::value::{map1, Value};
use std::collections::{BTreeMap, BTreeSet, HashSet, VecDeque};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Status {
Active,
Faulted,
Terminated,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Mode {
Auto,
Manual,
}
pub type InstanceId = String;
#[derive(Debug, Clone, PartialEq)]
pub struct QueuedEvent {
pub etype: String,
pub payload: Value,
pub origin: Option<String>,
}
#[derive(Debug, Clone, PartialEq)]
enum QItem {
Event(QueuedEvent),
Timer { state: NodeId, after_idx: usize },
}
impl QItem {
fn label(&self) -> String {
match self {
QItem::Event(e) => e.etype.clone(),
QItem::Timer { .. } => "__time__".into(),
}
}
}
#[derive(Debug, Clone)]
pub struct ArmedTimer {
pub state: NodeId,
pub after_idx: usize,
pub due: u64,
}
#[derive(Debug, Clone)]
pub struct DeadLetterRecord {
pub event: QueuedEvent,
pub reason: String,
}
#[derive(Debug, Clone)]
enum HistRecord {
Shallow(Vec<NodeId>),
Deep(Vec<NodeId>),
}
pub struct Instance {
pub id: InstanceId,
pub parent: Option<InstanceId>,
pub def_id: String,
pub def_version: i64,
pub status: Status,
pub mode: Mode,
pub active: BTreeSet<NodeId>,
pub esvs: BTreeMap<String, Value>,
pub external_source: BTreeMap<String, Value>,
queue: VecDeque<QItem>,
deferred: Vec<QItem>,
pub timers: Vec<ArmedTimer>,
pub spawn_counter: u64,
pub dead_letter: Vec<DeadLetterRecord>,
history: BTreeMap<NodeId, HistRecord>,
done_emitted: HashSet<NodeId>,
}
#[derive(Debug, Clone, Default)]
pub struct StepRecord {
pub event: String,
pub transition: Option<String>,
pub entered: Vec<String>,
pub exited: Vec<String>,
pub published: Vec<String>,
pub spawned: Vec<String>,
pub faulted: bool,
}
#[derive(Debug, Clone, Default)]
pub struct RunResult {
pub published: Vec<String>,
pub spawned: Vec<String>,
}
fn ancestors_inclusive(m: &Machine, n: NodeId) -> Vec<NodeId> {
let mut out = Vec::new();
let mut cur = Some(n);
while let Some(c) = cur {
out.push(c);
cur = m.get(c).parent;
}
out
}
fn proper_ancestors(m: &Machine, n: NodeId) -> Vec<NodeId> {
let mut out = Vec::new();
let mut cur = m.get(n).parent;
while let Some(c) = cur {
out.push(c);
cur = m.get(c).parent;
}
out
}
fn is_ancestor_or_equal(m: &Machine, anc: NodeId, n: NodeId) -> bool {
let mut cur = Some(n);
while let Some(c) = cur {
if c == anc {
return true;
}
cur = m.get(c).parent;
}
false
}
fn is_strictly_below(m: &Machine, anc: NodeId, n: NodeId) -> bool {
n != anc && is_ancestor_or_equal(m, anc, n)
}
fn is_leaf(m: &Machine, n: NodeId) -> bool {
matches!(m.get(n).kind, StateKind::Simple | StateKind::Final)
}
fn active_leaves(inst: &Instance, m: &Machine) -> Vec<NodeId> {
inst.active
.iter()
.copied()
.filter(|&n| is_leaf(m, n))
.collect()
}
fn esv_key(state: &StateDef, name: &str) -> String {
format!("{}::{}", state.path, name)
}
fn nearest_declaring(inst: &Instance, m: &Machine, scope: NodeId, name: &str) -> Option<NodeId> {
for s in scope_chain(m, scope) {
if inst.active.contains(&s) && m.get(s).esvs.iter().any(|(n, _)| n == name) {
return Some(s);
}
}
None
}
fn resolve_visible(inst: &Instance, m: &Machine, scope: NodeId) -> BTreeMap<String, Value> {
let mut out = BTreeMap::new();
for s in scope_chain(m, scope) {
if !inst.active.contains(&s) {
continue;
}
let sd = m.get(s);
for (name, _) in &sd.esvs {
if out.contains_key(name) {
continue;
}
if let Some(v) = inst.esvs.get(&esv_key(sd, name)) {
out.insert(name.clone(), v.clone());
}
}
}
out
}
fn scope_chain(m: &Machine, scope: NodeId) -> Vec<NodeId> {
let mut chain = vec![scope];
let mut cur = scope;
loop {
if m.get(cur).is_sm_boundary {
break;
}
match m.get(cur).parent {
Some(p) => {
chain.push(p);
cur = p;
if m.get(cur).is_sm_boundary {
break;
}
}
None => break,
}
}
chain
}
struct SpawnReq {
parent: InstanceId,
scope: NodeId,
def_id: String,
payload: BTreeMap<String, Value>,
result_var: Option<String>,
}
#[derive(Default)]
struct StepBuf {
deliveries: Vec<(InstanceId, QueuedEvent)>,
undirected: Vec<(InstanceId, String, Value)>,
spawns: Vec<SpawnReq>,
published: Vec<String>,
spawned: Vec<String>,
spawned_def: Vec<String>,
stop: bool,
}
pub struct Engine {
pub defs: BTreeMap<(String, i64), Machine>,
pub latest: BTreeMap<String, i64>,
pub instances: BTreeMap<InstanceId, Instance>,
pub clock: u64,
pub mode: Mode,
pub observer: Option<Box<dyn FnMut(&StepRecord) + Send>>,
run_published: Vec<String>,
run_spawned: Vec<String>,
}
impl Default for Engine {
fn default() -> Self {
Self::new()
}
}
#[derive(Debug)]
pub enum EngineError {
NotFound(String),
Validation(String),
Faulted(String),
Other(String),
}
#[derive(Debug, Clone)]
pub struct StateView {
pub instance: InstanceId,
pub def: String,
pub status: Status,
pub config: Vec<String>,
pub esvs: BTreeMap<String, Value>,
}
#[derive(Debug, Clone)]
pub struct ListView {
pub id: InstanceId,
pub def: String,
pub parent: Option<InstanceId>,
pub status: Status,
pub config: Vec<String>,
}
#[derive(Debug, Clone)]
pub struct InspectView {
pub instance: InstanceId,
pub status: Status,
pub config: Vec<String>,
pub esvs: BTreeMap<String, Value>,
pub enabled: Vec<String>,
pub queue: Vec<QueuedEvent>,
pub deferred: Vec<QueuedEvent>,
pub timers: Vec<ArmedTimer>,
pub history: BTreeMap<String, String>,
pub dead_letter: Vec<DeadLetterRecord>,
}
impl Engine {
pub fn new() -> Self {
Engine {
defs: BTreeMap::new(),
latest: BTreeMap::new(),
instances: BTreeMap::new(),
clock: 0,
mode: Mode::Auto,
observer: None,
run_published: Vec::new(),
run_spawned: Vec::new(),
}
}
pub fn set_observer<F: FnMut(&StepRecord) + Send + 'static>(&mut self, f: F) {
self.observer = Some(Box::new(f));
}
pub fn register(&mut self, m: Machine) {
let id = m.id.clone();
let ver = m.version;
self.defs.insert((id.clone(), ver), m);
let e = self.latest.entry(id).or_insert(0);
if ver > *e {
*e = ver;
}
}
fn def_ref(&self, id: &str, version: i64) -> &Machine {
self.defs
.get(&(id.to_string(), version))
.expect("definition registered")
}
pub fn def(&self, id: &str, version: Option<i64>) -> Result<&Machine, EngineError> {
let v = match version {
Some(v) => v,
None => *self
.latest
.get(id)
.ok_or_else(|| EngineError::NotFound(format!("definition '{id}'")))?,
};
self.defs
.get(&(id.to_string(), v))
.ok_or_else(|| EngineError::NotFound(format!("definition '{id}@{v}'")))
}
pub fn create_root(
&mut self,
id: &str,
def_id: &str,
version: Option<i64>,
external: &BTreeMap<String, Value>,
) -> Result<(), EngineError> {
if self.instances.contains_key(id) {
return Err(EngineError::Other(format!("instance '{id}' already exists")));
}
let (did, ver) = {
let m = self.def(def_id, version)?;
(m.id.clone(), m.version)
};
let mut inst = new_instance(id, None, did, ver, self.mode, external);
let m = self.def_ref(&inst.def_id, inst.def_version).clone();
let mut buf = StepBuf::default();
let mut rec = StepRecord::default();
enter(&mut inst, &m, m.top, self.clock, &Value::Null, None, &mut buf, &mut rec);
descend(&mut inst, &m, m.top, self.clock, &Value::Null, None, &mut buf, &mut rec);
self.instances.insert(id.to_string(), inst);
self.commit(buf)?;
self.run_to_quiescence()?;
Ok(())
}
pub fn instance(&self, id: &str) -> Result<&Instance, EngineError> {
self.instances
.get(id)
.ok_or_else(|| EngineError::NotFound(format!("instance '{id}'")))
}
pub fn validate_event(&self, inst_id: &str, etype: &str, payload: &Value) -> Result<(), String> {
let inst = self
.instances
.get(inst_id)
.ok_or_else(|| format!("instance '{inst_id}' not found"))?;
let m = self.def_ref(&inst.def_id, inst.def_version);
validate_event_payload(m, etype, payload)
}
pub fn inject(&mut self, inst_id: &str, etype: &str, payload: Value) -> Result<bool, EngineError> {
let ok = {
let inst = self.instances.get(inst_id).ok_or_else(|| {
EngineError::NotFound(format!("instance '{inst_id}'"))
})?;
let m = self.def_ref(&inst.def_id, inst.def_version);
validate_event_payload(m, etype, &payload).is_ok()
};
if ok {
let inst = self.instances.get_mut(inst_id).unwrap();
inst.queue.push_back(QItem::Event(QueuedEvent {
etype: etype.to_string(),
payload,
origin: None,
}));
}
Ok(ok)
}
pub fn send(&mut self, inst_id: &str, etype: &str, payload: Value) -> Result<RunResult, EngineError> {
self.run_published.clear();
self.run_spawned.clear();
let accepted = self.inject(inst_id, etype, payload)?;
if accepted && self.mode == Mode::Auto {
self.run_to_quiescence()?;
}
Ok(RunResult {
published: std::mem::take(&mut self.run_published),
spawned: std::mem::take(&mut self.run_spawned),
})
}
pub fn advance(&mut self, duration_ms: u64) -> Result<RunResult, EngineError> {
self.run_published.clear();
self.run_spawned.clear();
self.clock = self.clock.saturating_add(duration_ms);
loop {
let pick: Option<(InstanceId, usize, u64)> = {
let mut best: Option<(InstanceId, usize, u64)> = None;
for (iid, inst) in &self.instances {
for (i, t) in inst.timers.iter().enumerate() {
if t.due <= self.clock {
let cand = (iid.clone(), i, t.due);
match &best {
None => best = Some(cand),
Some(b) => {
if (t.due, &iid.clone()) < (b.2, &b.0) {
best = Some(cand);
}
}
}
}
}
}
best
};
match pick {
None => break,
Some((iid, i, _)) => {
let inst = self.instances.get_mut(&iid).unwrap();
let timer = inst.timers.remove(i);
inst.queue.push_back(QItem::Timer {
state: timer.state,
after_idx: timer.after_idx,
});
}
}
}
self.run_to_quiescence()?;
Ok(RunResult {
published: std::mem::take(&mut self.run_published),
spawned: std::mem::take(&mut self.run_spawned),
})
}
pub fn step(&mut self, inst_id: &str, n: usize) -> Result<Vec<StepRecord>, EngineError> {
self.run_published.clear();
self.run_spawned.clear();
let mut records = Vec::new();
for _ in 0..n {
let has = self
.instances
.get(inst_id)
.map(|i| !i.queue.is_empty())
.unwrap_or(false);
if !has {
break;
}
let rec = self.dispatch_one(inst_id)?;
records.push(rec);
self.run_to_quiescence()?;
}
Ok(records)
}
pub fn env_change(
&mut self,
inst_id: &str,
changed: BTreeMap<String, Value>,
) -> Result<RunResult, EngineError> {
let mut payload = BTreeMap::new();
payload.insert("changed".to_string(), Value::Map(changed));
self.send(inst_id, "env", Value::Map(payload))
}
pub fn set_mode(&mut self, mode: Mode) {
self.mode = mode;
for inst in self.instances.values_mut() {
inst.mode = mode;
}
}
pub fn get_mode(&self) -> Mode {
self.mode
}
pub fn migrate_quiescent(&mut self) -> Result<(), EngineError> {
let candidates: Vec<(InstanceId, String, i64)> = self
.instances
.iter()
.filter(|(_, i)| i.status == Status::Active && i.queue.is_empty() && i.deferred.is_empty())
.map(|(id, i)| (id.clone(), i.def_id.clone(), i.def_version))
.collect();
for (iid, def_id, v_old) in candidates {
if let Some(new_ver) = self.find_migration_target(&def_id, v_old) {
self.apply_migration(&iid, &def_id, v_old, new_ver)?;
}
}
Ok(())
}
fn find_migration_target(&self, def_id: &str, from: i64) -> Option<i64> {
let latest = *self.latest.get(def_id)?;
if latest <= from {
return None;
}
for v in (from + 1)..=latest {
if let Some(new_m) = self.defs.get(&(def_id.to_string(), v)) {
if new_m.migrations.iter().any(|mig| mig.from == from && mig.to == v) {
return Some(v);
}
}
}
None
}
fn apply_migration(
&mut self,
iid: &str,
def_id: &str,
from: i64,
to: i64,
) -> Result<(), EngineError> {
let migration = {
let new_m = self.defs.get(&(def_id.to_string(), to)).cloned();
let old_m = self.defs.get(&(def_id.to_string(), from)).cloned();
let (new_m, old_m) = match (new_m, old_m) {
(Some(n), Some(o)) => (n, o),
_ => return Ok(()),
};
new_m
.migrations
.iter()
.find(|m| m.from == from && m.to == to)
.cloned()
.map(|m| (m, new_m, old_m))
};
let (migration, new_m, old_m) = match migration {
Some(x) => x,
None => return Ok(()),
};
let leaf_name = {
let inst = self.instances.get(iid).unwrap();
let leaves = active_leaves(inst, &old_m);
if leaves.len() != 1 {
return Ok(()); }
old_m.get(leaves[0]).id.clone()
};
if let Some(when) = &migration.when {
let _inst = self.instances.get(iid).unwrap();
let mut env = Env::new();
env = env.with("state", Value::Str(leaf_name.clone()));
if !cel::eval_bool(when, &env).unwrap_or(false) {
return Ok(());
}
}
let new_leaf_name = match migration.state_map.get(&leaf_name) {
Some(n) => n.clone(),
None => return Ok(()),
};
let new_leaf = match new_m.by_name.get(&new_leaf_name).copied() {
Some(n) => n,
None => return Ok(()),
};
let inst = self.instances.get_mut(iid).unwrap();
inst.active.clear();
for a in ancestors_inclusive(&new_m, new_leaf) {
inst.active.insert(a);
}
inst.def_version = to;
let mut buf = StepBuf::default();
let mut rec = StepRecord::default();
let event = Value::Null;
let scope = new_m.top;
let mut env = build_env(inst, &new_m, scope, &event);
env = env.with("state", Value::Str(leaf_name.clone()));
for a in &migration.esvs {
if let Action::Assign(pairs) = a {
for (name, expr) in pairs {
if let Ok(val) = cel::eval(expr, &env) {
if let Some(st) = nearest_declaring(inst, &new_m, scope, name) {
let decl = new_m
.get(st)
.esvs
.iter()
.find(|(n, _)| n == name)
.map(|(_, d)| d.clone())
.unwrap();
if let Some(c) = val.coerce_to(&decl.ty) {
inst.esvs.insert(esv_key(new_m.get(st), name), c);
}
}
}
}
} else {
let _ = run_one(inst, &new_m, scope, a, &event, None, &mut buf, &mut rec);
}
}
let _ = (buf, rec);
Ok(())
}
fn run_to_quiescence(&mut self) -> Result<(), EngineError> {
let mut guard = 0u32;
loop {
guard += 1;
if guard > 200_000 {
return Err(EngineError::Other("runaway quiescence loop".into()));
}
let target = self
.instances
.iter()
.filter(|(_, i)| i.status == Status::Active && i.mode == Mode::Auto && !i.queue.is_empty())
.map(|(id, _)| id.clone())
.next();
match target {
Some(id) => {
self.dispatch_one(&id)?;
}
None => break,
}
}
Ok(())
}
fn dispatch_one(&mut self, inst_id: &str) -> Result<StepRecord, EngineError> {
let (did, ver) = {
let inst = self.instances.get(inst_id).ok_or_else(|| {
EngineError::NotFound(format!("instance '{inst_id}'"))
})?;
(inst.def_id.clone(), inst.def_version)
};
let m = self.def_ref(&did, ver).clone();
let item = self.instances.get_mut(inst_id).unwrap().queue.pop_front();
let rec = match item {
None => StepRecord::default(),
Some(qi) => {
let rec = self.execute_rtc(inst_id, &m, qi);
if let Some(o) = &mut self.observer {
o(&rec);
}
rec
}
};
Ok(rec)
}
fn execute_rtc(&mut self, inst_id: &str, m: &Machine, item: QItem) -> StepRecord {
let mut rec = StepRecord {
event: item.label(),
..Default::default()
};
let etype = match &item {
QItem::Event(e) => Some(e.etype.clone()),
QItem::Timer { .. } => None,
};
let event_value = match &item {
QItem::Event(e) => event_binding(e),
QItem::Timer { .. } => Value::Null,
};
let backup = self.instances.get(inst_id).map(|i| Instance::snapshot_state(i));
if etype.as_deref() == Some("env") {
if let Some(changed) = changed_of(&event_value) {
for (k, v) in changed {
self.instances
.get_mut(inst_id)
.unwrap()
.external_source
.insert(k, v);
}
}
}
let handlers = match &item {
QItem::Event(e) => find_handlers(self.instances.get_mut(inst_id).unwrap(), m, &e.etype, &event_value),
QItem::Timer { state, after_idx } => vec![Handler { state: *state, kind: HKind::Timer(*after_idx) }],
};
let mut buf = StepBuf::default();
if handlers.is_empty() {
let defer_set = effective_defer(self.instances.get(inst_id).unwrap(), m);
let do_defer = match &item {
QItem::Event(e) => defer_set.contains(&e.etype),
QItem::Timer { .. } => false,
};
if do_defer {
self.instances.get_mut(inst_id).unwrap().deferred.push(item);
}
return rec;
}
let mut faulted: Option<CelError> = None;
for h in &handlers {
let inst = self.instances.get_mut(inst_id).unwrap();
if let Err(e) = execute_handler(inst, m, h, &event_value, etype.as_deref(), &mut buf, &mut rec) {
faulted = Some(e);
break;
}
}
if let Some(fault) = faulted {
if let Some(b) = backup {
let inst = self.instances.get_mut(inst_id).unwrap();
inst.restore_state(b);
}
rec.faulted = true;
rec.published.clear();
rec.spawned.clear();
rec.entered.clear();
rec.exited.clear();
rec.transition = None;
if let QItem::Event(e) = &item {
self.instances.get_mut(inst_id).unwrap().dead_letter.push(DeadLetterRecord {
event: e.clone(),
reason: fault.to_string(),
});
}
let has_err = scope_has_error_handler(self.instances.get(inst_id).unwrap(), m);
if has_err {
self.instances.get_mut(inst_id).unwrap().queue.push_front(QItem::Event(QueuedEvent {
etype: "error".into(),
payload: Value::Map(error_payload(&fault)),
origin: None,
}));
} else {
self.instances.get_mut(inst_id).unwrap().status = Status::Faulted;
}
return rec;
}
let _ = self.commit(buf);
let m2 = m.clone();
self.post_step(inst_id, &m2);
if self.instances.get(inst_id).map(|i| i.status == Status::Active).unwrap_or(false) {
}
rec
}
fn post_step(&mut self, inst_id: &str, m: &Machine) {
let complete: Vec<NodeId> = self
.instances
.get(inst_id)
.map(|i| complete_states(i, m))
.unwrap_or_default();
for s in complete {
if s == m.top {
continue;
}
let already = self.instances.get(inst_id).unwrap().done_emitted.contains(&s);
if already {
continue;
}
let leaf_name = self
.instances
.get(inst_id)
.map(|i| composite_leaf_name(i, m, s))
.unwrap_or_default();
let inst = self.instances.get_mut(inst_id).unwrap();
inst.done_emitted.insert(s);
inst.queue.push_back(QItem::Event(QueuedEvent {
etype: "done".into(),
payload: map1("state", Value::Str(leaf_name)),
origin: None,
}));
}
self.handle_termination(inst_id, m);
let inst = self.instances.get_mut(inst_id).unwrap();
replay_deferred(inst, m);
}
fn handle_termination(&mut self, inst_id: &str, m: &Machine) {
let (terminate, parent_id) = {
let inst = match self.instances.get(inst_id) {
Some(i) if i.status == Status::Active => i,
_ => return,
};
let top_final = inst.active.iter().any(|&n| {
m.get(n).kind == StateKind::Final && m.get(n).parent == Some(m.top)
});
let stop_pending = inst
.queue
.iter()
.any(|q| matches!(q, QItem::Event(e) if e.etype == "__stop__"));
let do_term = (top_final && inst.parent.is_some()) || stop_pending;
(do_term, inst.parent.clone())
};
if terminate {
let inst = self.instances.get_mut(inst_id).unwrap();
inst.queue.retain(|q| !matches!(q, QItem::Event(e) if e.etype == "__stop__"));
let leaves: Vec<NodeId> = inst.active.iter().copied().filter(|&n| is_leaf(m, n)).collect();
let mut sink = StepBuf::default();
for leaf in leaves {
let mut rec = StepRecord::default();
exit_up_to(inst, m, leaf, Some(m.top), &mut sink, &mut rec);
}
inst.active.clear();
inst.status = Status::Terminated;
inst.done_emitted.clear();
let _ = sink;
if let Some(parent) = parent_id {
if let Some(p) = self.instances.get_mut(&parent) {
if p.status == Status::Active {
p.queue.push_back(QItem::Event(QueuedEvent {
etype: "done".into(),
payload: Value::Null,
origin: Some(inst_id.to_string()),
}));
}
}
}
}
}
fn commit(&mut self, buf: StepBuf) -> Result<(), EngineError> {
self.run_published.extend(buf.published);
self.run_spawned.extend(buf.spawned_def);
for (target, ev) in buf.deliveries {
if let Some(t) = self.instances.get_mut(&target) {
if t.status == Status::Active {
t.queue.push_back(QItem::Event(ev));
}
}
}
for (publisher, etype, payload) in buf.undirected {
let scope = self
.instances
.get(&publisher)
.and_then(|p| {
let m = self.defs.get(&(p.def_id.clone(), p.def_version))?;
m.events.get(&etype).map(|d| d.scope)
})
.unwrap_or(crate::machine::Scope::Internal);
let targets = self.undirected_targets(&publisher, &etype, scope);
for t in targets {
if let Some(ti) = self.instances.get_mut(&t) {
if ti.status == Status::Active {
ti.queue.push_back(QItem::Event(QueuedEvent {
etype: etype.clone(),
payload: payload.clone(),
origin: Some(publisher.clone()),
}));
}
}
}
}
for sp in buf.spawns {
let parent_id = sp.parent.clone();
let n = {
let p = self.instances.get_mut(&parent_id).unwrap();
p.spawn_counter += 1;
p.spawn_counter
};
let child_id = format!("{parent_id}/{n}");
if let Some(res) = &sp.result_var {
let m_clone = self
.defs
.get({
let p = self.instances.get(&parent_id).unwrap();
&(p.def_id.clone(), p.def_version)
})
.cloned();
if let Some(m) = m_clone {
let p = self.instances.get_mut(&parent_id).unwrap();
if let Some(st) = nearest_declaring(p, &m, sp.scope, res) {
p.esvs.insert(esv_key(m.get(st), res), Value::Str(child_id.clone()));
}
}
}
let child_m = match self.def(&sp.def_id, None) {
Ok(cm) => cm.clone(),
Err(_) => continue,
};
let mut child = new_instance(
&child_id,
Some(parent_id.clone()),
child_m.id.clone(),
child_m.version,
self.mode,
&BTreeMap::new(),
);
for (name, _) in &child_m.get(child_m.top).esvs.clone() {
if let Some(v) = sp.payload.get(name) {
child.external_source.insert(name.clone(), v.clone());
}
}
let mut cbuf = StepBuf::default();
let mut crec = StepRecord::default();
enter(&mut child, &child_m, child_m.top, self.clock, &Value::Null, None, &mut cbuf, &mut crec);
descend(&mut child, &child_m, child_m.top, self.clock, &Value::Null, None, &mut cbuf, &mut crec);
self.instances.insert(child_id, child);
self.commit(cbuf)?;
}
Ok(())
}
fn undirected_targets(
&self,
publisher: &str,
etype: &str,
scope: crate::machine::Scope,
) -> Vec<InstanceId> {
use crate::machine::Scope;
let candidates: Vec<InstanceId> = match scope {
Scope::Internal => vec![publisher.to_string()],
Scope::Global => self.instances.keys().cloned().collect(),
Scope::Local => {
let mut set: HashSet<InstanceId> = HashSet::new();
let mut cur = Some(publisher.to_string());
while let Some(id) = cur {
set.insert(id.clone());
cur = self.instances.get(&id).and_then(|i| i.parent.clone());
}
let mut stack = vec![publisher.to_string()];
while let Some(id) = stack.pop() {
set.insert(id.clone());
for (other_id, other) in &self.instances {
if other.parent.as_deref() == Some(id.as_str()) {
stack.push(other_id.clone());
}
}
}
set.into_iter().collect()
}
};
candidates
.into_iter()
.filter(|cid| {
if scope == Scope::Internal && cid == publisher {
return true; }
let inst = match self.instances.get(cid) {
Some(i) => i,
None => return false,
};
let m = self.defs.get(&(inst.def_id.clone(), inst.def_version));
m.map(|mm| mm.subscribe.iter().any(|s| s == etype))
.unwrap_or(false)
})
.collect()
}
pub fn state_view(&self, inst_id: &str) -> Result<StateView, EngineError> {
let inst = self.instance(inst_id)?;
let m = self.def_ref(&inst.def_id, inst.def_version);
Ok(StateView {
instance: inst_id.to_string(),
def: format!("{}@{}", inst.def_id, inst.def_version),
status: inst.status,
config: config_of(inst, m),
esvs: reported_esvs(inst, m),
})
}
pub fn list_view(&self) -> Vec<ListView> {
self.instances
.iter()
.map(|(id, inst)| {
let m = self.def_ref(&inst.def_id, inst.def_version);
ListView {
id: id.clone(),
def: format!("{}@{}", inst.def_id, inst.def_version),
parent: inst.parent.clone(),
status: inst.status,
config: config_of(inst, m),
}
})
.collect()
}
pub fn inspect_view(&self, inst_id: &str) -> Result<InspectView, EngineError> {
let inst = self.instance(inst_id)?;
let m = self.def_ref(&inst.def_id, inst.def_version);
let history = inst
.history
.iter()
.map(|(n, h)| {
let kind = match h {
HistRecord::Shallow(_) => "shallow",
HistRecord::Deep(_) => "deep",
};
(m.get(*n).path.clone(), kind.to_string())
})
.collect();
Ok(InspectView {
instance: inst_id.to_string(),
status: inst.status,
config: config_of(inst, m),
esvs: reported_esvs(inst, m),
enabled: enabled_events(inst, m),
queue: inst.queue.iter().filter_map(|q| match q {
QItem::Event(e) => Some(e.clone()),
_ => None,
}).collect(),
deferred: inst.deferred.iter().filter_map(|q| match q {
QItem::Event(e) => Some(e.clone()),
_ => None,
}).collect(),
timers: inst.timers.clone(),
history,
dead_letter: inst.dead_letter.clone(),
})
}
pub fn enabled_events(&self, inst_id: &str) -> Result<Vec<String>, EngineError> {
let inst = self.instance(inst_id)?;
let m = self.def_ref(&inst.def_id, inst.def_version);
Ok(enabled_events(inst, m))
}
pub fn snapshot(&self, inst_id: &str) -> Result<Snapshot, EngineError> {
let inst = self.instance(inst_id)?;
let m = self.def_ref(&inst.def_id, inst.def_version);
Snapshot::from_instance(inst, m)
}
pub fn restore(&mut self, snap: Snapshot) -> Result<(), EngineError> {
let m = self
.defs
.get(&(snap.def_id.clone(), snap.def_version))
.ok_or_else(|| EngineError::NotFound(format!("definition '{}@{}'", snap.def_id, snap.def_version)))?
.clone();
let inst = snap.into_instance(&m);
self.instances.insert(inst.id.clone(), inst);
Ok(())
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Snapshot {
pub def_id: String,
pub def_version: i64,
pub id: InstanceId,
pub parent_id: Option<InstanceId>,
pub status: String,
pub state_config: Vec<String>,
pub esvs: BTreeMap<String, Value>,
pub external_source: BTreeMap<String, Value>,
pub queue: Vec<SnapEvent>,
pub deferred: Vec<SnapEvent>,
pub timers: Vec<SnapTimer>,
pub dead_letter: Vec<SnapDeadLetter>,
pub history: BTreeMap<String, SnapHist>,
pub spawn_counter: u64,
pub mode: String,
#[serde(default)]
pub clock: u64,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct SnapEvent {
pub etype: String,
pub payload: Value,
pub origin: Option<String>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct SnapTimer {
pub state: String,
pub after_idx: usize,
pub due: u64,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct SnapDeadLetter {
pub event: SnapEvent,
pub reason: String,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct SnapHist {
pub kind: String,
pub states: Vec<String>,
}
impl Snapshot {
fn from_instance(inst: &Instance, m: &Machine) -> Result<Snapshot, EngineError> {
let state_config: Vec<String> = active_leaves(inst, m)
.into_iter()
.map(|n| m.get(n).path.clone())
.collect();
let history = inst
.history
.iter()
.map(|(n, h)| {
let (kind, states) = match h {
HistRecord::Shallow(s) => ("shallow", s.clone()),
HistRecord::Deep(s) => ("deep", s.clone()),
};
(
m.get(*n).path.clone(),
SnapHist {
kind: kind.to_string(),
states: states.iter().map(|x| m.get(*x).path.clone()).collect(),
},
)
})
.collect();
Ok(Snapshot {
def_id: inst.def_id.clone(),
def_version: inst.def_version,
id: inst.id.clone(),
parent_id: inst.parent.clone(),
status: status_str(inst.status).to_string(),
state_config,
esvs: inst.esvs.clone(),
external_source: inst.external_source.clone(),
queue: snap_events(inst.queue.iter()),
deferred: snap_events(inst.deferred.iter()),
timers: inst
.timers
.iter()
.map(|t| SnapTimer {
state: m.get(t.state).path.clone(),
after_idx: t.after_idx,
due: t.due,
})
.collect(),
dead_letter: inst
.dead_letter
.iter()
.map(|d| SnapDeadLetter {
event: SnapEvent {
etype: d.event.etype.clone(),
payload: d.event.payload.clone(),
origin: d.event.origin.clone(),
},
reason: d.reason.clone(),
})
.collect(),
history,
spawn_counter: inst.spawn_counter,
mode: match inst.mode {
Mode::Auto => "auto",
Mode::Manual => "manual",
}
.into(),
clock: 0,
})
}
fn into_instance(self, m: &Machine) -> Instance {
let status = match self.status.as_str() {
"faulted" => Status::Faulted,
"terminated" => Status::Terminated,
_ => Status::Active,
};
let mode = match self.mode.as_str() {
"manual" => Mode::Manual,
_ => Mode::Auto,
};
let leaves: Vec<NodeId> = self
.state_config
.iter()
.filter_map(|p| m.by_path.get(p).copied())
.collect();
let mut inst = Instance {
id: self.id,
parent: self.parent_id,
def_id: self.def_id,
def_version: self.def_version,
status,
mode,
active: BTreeSet::new(),
esvs: self.esvs,
external_source: self.external_source,
queue: self.queue.into_iter().map(|e| QItem::Event(QueuedEvent {
etype: e.etype,
payload: e.payload,
origin: e.origin,
})).collect(),
deferred: self.deferred.into_iter().map(|e| QItem::Event(QueuedEvent {
etype: e.etype,
payload: e.payload,
origin: e.origin,
})).collect(),
timers: self.timers.into_iter().filter_map(|t| {
m.by_path.get(&t.state).copied().map(|state| ArmedTimer {
state,
after_idx: t.after_idx,
due: t.due,
})
}).collect(),
spawn_counter: self.spawn_counter,
dead_letter: self.dead_letter.into_iter().map(|d| DeadLetterRecord {
event: QueuedEvent {
etype: d.event.etype,
payload: d.event.payload,
origin: d.event.origin,
},
reason: d.reason,
}).collect(),
history: self.history.into_iter().filter_map(|(path, h)| {
let state = m.by_path.get(&path).copied()?;
let states: Vec<NodeId> = h
.states
.iter()
.filter_map(|p| m.by_path.get(p).copied())
.collect();
let rec = if h.kind == "shallow" {
HistRecord::Shallow(states)
} else {
HistRecord::Deep(states)
};
Some((state, rec))
}).collect(),
done_emitted: HashSet::new(),
};
for leaf in leaves {
for a in ancestors_inclusive(m, leaf) {
inst.active.insert(a);
}
}
inst
}
}
fn snap_events<'a>(items: impl Iterator<Item = &'a QItem>) -> Vec<SnapEvent> {
items
.filter_map(|q| match q {
QItem::Event(e) => Some(SnapEvent {
etype: e.etype.clone(),
payload: e.payload.clone(),
origin: e.origin.clone(),
}),
QItem::Timer { .. } => None,
})
.collect()
}
fn status_str(s: Status) -> &'static str {
match s {
Status::Active => "active",
Status::Faulted => "faulted",
Status::Terminated => "terminated",
}
}
fn new_instance(
id: &str,
parent: Option<InstanceId>,
def_id: String,
def_version: i64,
mode: Mode,
external: &BTreeMap<String, Value>,
) -> Instance {
Instance {
id: id.to_string(),
parent,
def_id,
def_version,
status: Status::Active,
mode,
active: BTreeSet::new(),
esvs: BTreeMap::new(),
external_source: external.clone(),
queue: VecDeque::new(),
deferred: Vec::new(),
timers: Vec::new(),
spawn_counter: 0,
dead_letter: Vec::new(),
history: BTreeMap::new(),
done_emitted: HashSet::new(),
}
}
impl Instance {
fn snapshot_state(&self) -> InstanceState {
InstanceState {
active: self.active.clone(),
esvs: self.esvs.clone(),
timers: self.timers.clone(),
history: self.history.clone(),
done_emitted: self.done_emitted.clone(),
deferred: self.deferred.clone(),
spawn_counter: self.spawn_counter,
external_source: self.external_source.clone(),
status: self.status,
queue: self.queue.clone(),
}
}
fn restore_state(&mut self, s: InstanceState) {
self.active = s.active;
self.esvs = s.esvs;
self.timers = s.timers;
self.history = s.history;
self.done_emitted = s.done_emitted;
self.deferred = s.deferred;
self.spawn_counter = s.spawn_counter;
self.external_source = s.external_source;
self.status = s.status;
self.queue = s.queue;
}
}
struct InstanceState {
active: BTreeSet<NodeId>,
esvs: BTreeMap<String, Value>,
timers: Vec<ArmedTimer>,
history: BTreeMap<NodeId, HistRecord>,
done_emitted: HashSet<NodeId>,
deferred: Vec<QItem>,
spawn_counter: u64,
external_source: BTreeMap<String, Value>,
status: Status,
queue: VecDeque<QItem>,
}
enum HKind {
Transition(usize),
Timer(usize),
}
struct Handler {
state: NodeId,
kind: HKind,
}
fn find_handlers(inst: &Instance, m: &Machine, etype: &str, event_value: &Value) -> Vec<Handler> {
let leaves = active_leaves(inst, m);
let mut found: Vec<Handler> = Vec::new();
let mut seen: HashSet<NodeId> = HashSet::new();
for leaf in leaves {
for s in ancestors_inclusive(m, leaf) {
if !inst.active.contains(&s) {
continue;
}
if let Some(list) = m.get(s).on_events.get(etype) {
let env = build_env(inst, m, s, event_value);
let mut matched = None;
for (i, t) in list.iter().enumerate() {
let pass = match &t.guard {
Some(g) => cel::eval_bool(g, &env).unwrap_or(false),
None => true,
};
if pass {
matched = Some(i);
break;
}
}
if let Some(i) = matched {
if seen.insert(s) {
found.push(Handler { state: s, kind: HKind::Transition(i) });
}
}
break; }
}
}
found
}
fn scope_has_error_handler(inst: &Instance, m: &Machine) -> bool {
inst.active
.iter()
.any(|&n| m.get(n).on_events.contains_key("error"))
}
fn execute_handler(
inst: &mut Instance,
m: &Machine,
h: &Handler,
event_value: &Value,
etype: Option<&str>,
buf: &mut StepBuf,
rec: &mut StepRecord,
) -> Result<(), CelError> {
match &h.kind {
HKind::Timer(idx) => {
let after = m.get(h.state).after[*idx].clone();
if let Some(g) = &after.guard {
let env = build_env(inst, m, h.state, event_value);
if !cel::eval_bool(g, &env)? {
return Ok(());
}
}
run_actions(inst, m, h.state, &after.action, event_value, etype, buf, rec)?;
if let Some(target) = after.target {
rec.transition = Some(m.get(target).id.clone());
external_transition(inst, m, h.state, target, &[], event_value, etype, buf, rec)?;
}
Ok(())
}
HKind::Transition(idx) => {
let list = m.get(h.state).on_events.get(etype.unwrap_or("")).cloned();
let list = match list {
Some(l) => l,
None => return Ok(()),
};
let t = list[*idx].clone();
if t.target.is_none() || t.internal {
run_actions(inst, m, h.state, &t.action, event_value, etype, buf, rec)?;
rec.transition = None;
return Ok(());
}
let target = t.target.unwrap();
if t.local {
rec.transition = Some(m.get(target).id.clone());
run_actions(inst, m, h.state, &t.action, event_value, etype, buf, rec)?;
local_transition(inst, m, h.state, target, event_value, etype, buf, rec)?;
return Ok(());
}
rec.transition = Some(m.get(target).id.clone());
let is_choice = m.get(target).kind == StateKind::Choice;
if is_choice {
run_actions(inst, m, h.state, &t.action, event_value, etype, buf, rec)?;
let final_target = resolve_choice(inst, m, h.state, target, event_value, etype, buf, rec)?;
external_transition(inst, m, h.state, final_target, &[], event_value, etype, buf, rec)?;
} else {
external_transition(inst, m, h.state, target, &t.action, event_value, etype, buf, rec)?;
}
Ok(())
}
}
}
fn resolve_choice(
inst: &mut Instance,
m: &Machine,
source: NodeId,
start: NodeId,
event_value: &Value,
etype: Option<&str>,
buf: &mut StepBuf,
rec: &mut StepRecord,
) -> Result<NodeId, CelError> {
let mut cur = start;
loop {
let branches = m.get(cur).choice.clone().unwrap_or_default();
let env = build_env(inst, m, source, event_value);
let mut chosen: Option<usize> = None;
for (i, b) in branches.iter().enumerate() {
let pass = match &b.guard {
Some(g) => cel::eval_bool(g, &env)?,
None => true,
};
if pass {
chosen = Some(i);
break;
}
}
let i = chosen.ok_or_else(|| CelError::Other("no choice branch matched".into()))?;
let b = branches[i].clone();
run_actions(inst, m, source, &b.action, event_value, etype, buf, rec)?;
if m.get(b.target).kind == StateKind::Choice {
cur = b.target;
continue;
}
return Ok(b.target);
}
}
fn external_transition(
inst: &mut Instance,
m: &Machine,
source: NodeId,
target: NodeId,
action: &[Action],
event_value: &Value,
etype: Option<&str>,
buf: &mut StepBuf,
rec: &mut StepRecord,
) -> Result<(), CelError> {
let lca = lca_external(m, source, target);
let anchor = child_below(m, lca, source);
let mut to_exit: Vec<NodeId> = inst
.active
.iter()
.copied()
.filter(|&x| is_strictly_below(m, lca, x) && is_ancestor_or_equal(m, anchor, x))
.collect();
to_exit.sort_by(|a, b| m.get(*b).depth.cmp(&m.get(*a).depth));
for &x in &to_exit {
if m.get(x).has_history() {
record_history(inst, m, x);
}
}
for x in to_exit {
exit_state(inst, m, x, buf, rec);
}
if !action.is_empty() {
run_actions(inst, m, source, action, event_value, etype, buf, rec)?;
}
let path = entry_path_below(m, lca, target);
for &s in &path {
enter(inst, m, s, 0, event_value, etype, buf, rec);
}
if let Some(&tgt) = path.last() {
descend(inst, m, tgt, 0, event_value, etype, buf, rec);
}
Ok(())
}
fn local_transition(
inst: &mut Instance,
m: &Machine,
source: NodeId,
target: NodeId,
event_value: &Value,
etype: Option<&str>,
buf: &mut StepBuf,
rec: &mut StepRecord,
) -> Result<(), CelError> {
let mut to_exit: Vec<NodeId> = inst
.active
.iter()
.copied()
.filter(|&x| is_strictly_below(m, source, x))
.collect();
to_exit.sort_by(|a, b| m.get(*b).depth.cmp(&m.get(*a).depth));
for &x in &to_exit {
if m.get(x).has_history() {
record_history(inst, m, x);
}
}
for x in to_exit {
exit_state(inst, m, x, buf, rec);
}
let path = entry_path_below(m, source, target);
for &s in &path {
enter(inst, m, s, 0, event_value, etype, buf, rec);
}
if let Some(&tgt) = path.last() {
descend(inst, m, tgt, 0, event_value, etype, buf, rec);
}
Ok(())
}
fn lca_external(m: &Machine, source: NodeId, target: NodeId) -> NodeId {
let sa = proper_ancestors(m, source);
let ta: HashSet<NodeId> = proper_ancestors(m, target).into_iter().collect();
for a in sa {
if ta.contains(&a) {
return a;
}
}
m.top
}
fn child_below(m: &Machine, anc: NodeId, n: NodeId) -> NodeId {
let mut cur = n;
while let Some(p) = m.get(cur).parent {
if p == anc {
return cur;
}
cur = p;
}
n
}
fn entry_path_below(m: &Machine, lca: NodeId, target: NodeId) -> Vec<NodeId> {
let chain = ancestors_inclusive(m, target); let mut path: Vec<NodeId> = chain.into_iter().rev().collect(); while path.first().map(|&x| x != lca).unwrap_or(false) {
path.remove(0);
}
if path.first() == Some(&lca) {
path.remove(0);
}
path
}
fn enter(
inst: &mut Instance,
m: &Machine,
n: NodeId,
clock: u64,
event_value: &Value,
etype: Option<&str>,
buf: &mut StepBuf,
rec: &mut StepRecord,
) {
inst.active.insert(n);
let sd = m.get(n);
if sd.is_sm_boundary && !sd.sm_with.is_empty() {
if let Some(parent) = sd.parent {
let env = build_env(inst, m, parent, event_value);
for (name, expr) in &sd.sm_with {
if let Ok(v) = crate::cel::eval(expr, &env) {
inst.external_source.insert(name.clone(), v);
}
}
}
}
for (name, decl) in sd.esvs.clone() {
let val = if decl.external {
inst.external_source.get(&name).cloned().unwrap_or(Value::Null)
} else {
decl.init.clone().unwrap_or(Value::Null)
};
inst.esvs.insert(esv_key(sd, &name), val);
}
let entry = sd.entry.clone();
let _ = run_actions(inst, m, n, &entry, event_value, etype, buf, rec);
let afters = sd.after.clone();
for (i, a) in afters.iter().enumerate() {
inst.timers.push(ArmedTimer {
state: n,
after_idx: i,
due: clock.saturating_add(a.duration_ms),
});
}
rec.entered.push(sd.id.clone());
}
fn descend(
inst: &mut Instance,
m: &Machine,
n: NodeId,
clock: u64,
event_value: &Value,
etype: Option<&str>,
buf: &mut StepBuf,
rec: &mut StepRecord,
) {
let sd = m.get(n);
match sd.kind {
StateKind::Composite => {
let has_hist = sd.has_history();
let hist = inst.history.get(&n).cloned();
if has_hist && hist.is_some() {
restore_history(inst, m, n, clock, event_value, etype, hist.unwrap(), buf, rec);
} else if let Some(init) = sd.initial.clone() {
for a in &init.action {
let _ = run_actions(inst, m, n, std::slice::from_ref(a), event_value, etype, buf, rec);
}
enter(inst, m, init.target, clock, event_value, etype, buf, rec);
descend(inst, m, init.target, clock, event_value, etype, buf, rec);
}
}
StateKind::Orthogonal => {
let has_hist = sd.has_history();
let hist = inst.history.get(&n).cloned();
if has_hist && hist.is_some() {
restore_history(inst, m, n, clock, event_value, etype, hist.unwrap(), buf, rec);
} else {
let regions = sd.regions.clone();
for r in ®ions {
for a in &r.initial.action {
let _ = run_actions(inst, m, n, std::slice::from_ref(a), event_value, etype, buf, rec);
}
enter(inst, m, r.initial.target, clock, event_value, etype, buf, rec);
descend(inst, m, r.initial.target, clock, event_value, etype, buf, rec);
}
}
}
_ => {}
}
}
fn restore_history(
inst: &mut Instance,
m: &Machine,
state: NodeId,
clock: u64,
event_value: &Value,
etype: Option<&str>,
hist: HistRecord,
buf: &mut StepBuf,
rec: &mut StepRecord,
) {
match hist {
HistRecord::Shallow(direct) => {
for child in direct {
enter(inst, m, child, clock, event_value, etype, buf, rec);
descend(inst, m, child, clock, event_value, etype, buf, rec);
}
}
HistRecord::Deep(leaves) => {
for leaf in leaves {
let path = entry_path_below(m, state, leaf);
for &s in &path {
enter(inst, m, s, clock, event_value, etype, buf, rec);
}
}
}
}
}
fn exit_state(
inst: &mut Instance,
m: &Machine,
n: NodeId,
buf: &mut StepBuf,
rec: &mut StepRecord,
) {
let sd = m.get(n);
inst.done_emitted.remove(&n);
let exit = sd.exit.clone();
let _ = run_actions(inst, m, n, &exit, &Value::Null, None, buf, rec);
for (name, _) in sd.esvs.clone() {
inst.esvs.remove(&esv_key(sd, &name));
}
inst.timers.retain(|t| t.state != n);
inst.active.remove(&n);
rec.exited.push(sd.id.clone());
}
fn exit_up_to(
inst: &mut Instance,
m: &Machine,
leaf: NodeId,
lca: Option<NodeId>,
buf: &mut StepBuf,
rec: &mut StepRecord,
) {
let mut cur = Some(leaf);
while let Some(c) = cur {
if lca == Some(c) {
break;
}
exit_state(inst, m, c, buf, rec);
cur = m.get(c).parent;
}
}
fn record_history(inst: &mut Instance, m: &Machine, state: NodeId) {
let sd = m.get(state);
match sd.history {
HistoryKind::Shallow => {
let direct: Vec<NodeId> = sd
.children
.iter()
.copied()
.filter(|&c| inst.active.contains(&c))
.collect();
inst.history.insert(state, HistRecord::Shallow(direct));
}
HistoryKind::Deep => {
let leaves: Vec<NodeId> = inst
.active
.iter()
.copied()
.filter(|&n| is_leaf(m, n) && is_strictly_below(m, state, n))
.collect();
inst.history.insert(state, HistRecord::Deep(leaves));
}
HistoryKind::None => {}
}
}
fn run_actions(
inst: &mut Instance,
m: &Machine,
scope: NodeId,
actions: &[Action],
event_value: &Value,
etype: Option<&str>,
buf: &mut StepBuf,
rec: &mut StepRecord,
) -> Result<(), CelError> {
for a in actions {
run_one(inst, m, scope, a, event_value, etype, buf, rec)?;
}
Ok(())
}
fn run_one(
inst: &mut Instance,
m: &Machine,
scope: NodeId,
action: &Action,
event_value: &Value,
etype: Option<&str>,
buf: &mut StepBuf,
_rec: &mut StepRecord,
) -> Result<(), CelError> {
let env = build_env(inst, m, scope, event_value);
match action {
Action::Assign(pairs) => {
for (name, expr) in pairs {
let val = cel::eval(expr, &env)?;
let decl_state = nearest_declaring(inst, m, scope, name)
.ok_or_else(|| CelError::Other(format!("assign to undeclared '{name}'")))?;
let sd = m.get(decl_state);
let decl = sd
.esvs
.iter()
.find(|(n, _)| n == name)
.map(|(_, d)| d.clone())
.unwrap();
if decl.external {
return Err(CelError::Other(format!("cannot assign to external esv '{name}'")));
}
let coerced = val
.coerce_to(&decl.ty)
.ok_or_else(|| CelError::Type(format!("'{name}' expects {}", decl.ty)))?;
inst.esvs.insert(esv_key(sd, name), coerced);
}
Ok(())
}
Action::Publish { event, to, payload } => {
let pmap = eval_payload(payload, &env)?;
buf.published.push(event.clone());
if let Some(to_expr) = to {
for t in eval_targets(to_expr, &env)? {
buf.deliveries.push((
t,
QueuedEvent {
etype: event.clone(),
payload: Value::Map(pmap.clone()),
origin: Some(inst.id.clone()),
},
));
}
} else {
buf.undirected.push((inst.id.clone(), event.clone(), Value::Map(pmap)));
}
Ok(())
}
Action::Refresh { only } => {
if etype != Some("env") {
return Err(CelError::Other("refresh only valid while handling env".into()));
}
if let Some(changed) = changed_of(event_value) {
let names: Vec<String> = match only {
Some(lst) => lst.clone(),
None => changed.keys().cloned().collect(),
};
for name in names {
if let Some(v) = changed.get(&name) {
if let Some(st) = nearest_declaring(inst, m, scope, &name) {
inst.esvs.insert(esv_key(m.get(st), &name), v.clone());
}
}
}
}
Ok(())
}
Action::Spawn { def, payload, result } => {
let pmap = eval_payload(payload, &env)?;
buf.spawns.push(SpawnReq {
parent: inst.id.clone(),
scope,
def_id: def.clone(),
payload: pmap,
result_var: result.clone(),
});
buf.spawned.push(def.clone());
buf.spawned_def.push(def.clone());
Ok(())
}
Action::Stop => {
buf.stop = true;
inst.queue.push_back(QItem::Event(QueuedEvent {
etype: "__stop__".into(),
payload: Value::Null,
origin: Some(inst.id.clone()),
}));
Ok(())
}
}
}
fn eval_payload(pairs: &[(String, String)], env: &Env) -> Result<BTreeMap<String, Value>, CelError> {
let mut out = BTreeMap::new();
for (k, expr) in pairs {
out.insert(k.clone(), cel::eval(expr, env)?);
}
Ok(out)
}
fn eval_targets(expr: &str, env: &Env) -> Result<Vec<InstanceId>, CelError> {
match cel::eval(expr, env)? {
Value::Str(s) => Ok(vec![s]),
Value::List(l) => Ok(l
.into_iter()
.filter_map(|x| x.as_str_value().map(|s| s.to_string()))
.collect()),
Value::Null => Ok(vec![]),
other => Err(CelError::Type(format!("publish.to must be string/list, got {}", other.type_name()))),
}
}
fn effective_defer(inst: &Instance, m: &Machine) -> HashSet<String> {
let mut out = HashSet::new();
for &n in &inst.active {
for d in &m.get(n).defer {
out.insert(d.clone());
}
}
out
}
fn replay_deferred(inst: &mut Instance, m: &Machine) {
let defer_set = effective_defer(inst, m);
let (mut move_front, mut remaining) = (Vec::new(), Vec::new());
for item in inst.deferred.drain(..) {
let still = match &item {
QItem::Event(e) => defer_set.contains(&e.etype),
QItem::Timer { .. } => false,
};
if still {
remaining.push(item);
} else {
move_front.push(item);
}
}
inst.deferred = remaining;
let mut new_q = VecDeque::new();
for item in move_front {
new_q.push_back(item);
}
while let Some(x) = inst.queue.pop_front() {
new_q.push_back(x);
}
inst.queue = new_q;
}
fn complete_states(inst: &Instance, m: &Machine) -> Vec<NodeId> {
let leaves = active_leaves(inst, m);
let mut out = Vec::new();
for &s in &inst.active {
let sd = m.get(s);
match sd.kind {
StateKind::Orthogonal => {
if regions_all_final(inst, m, s) {
out.push(s);
}
}
StateKind::Composite => {
if leaves.iter().any(|&l| {
is_ancestor_or_equal(m, s, l) && m.get(l).kind == StateKind::Final
}) {
out.push(s);
}
}
_ => {}
}
}
out
}
fn composite_leaf_name(inst: &Instance, m: &Machine, composite: NodeId) -> String {
for l in active_leaves(inst, m) {
if is_ancestor_or_equal(m, composite, l) && m.get(l).kind == StateKind::Final {
return m.get(l).id.clone();
}
}
String::new()
}
fn regions_all_final(inst: &Instance, m: &Machine, ortho: NodeId) -> bool {
let regions = m.get(ortho).regions.clone();
if regions.is_empty() {
return false;
}
for r in ®ions {
let any_final = r.states.iter().any(|&s| subtree_has_active_final(inst, m, s));
if !any_final {
return false;
}
}
true
}
fn subtree_has_active_final(inst: &Instance, m: &Machine, n: NodeId) -> bool {
if inst.active.contains(&n) && m.get(n).kind == StateKind::Final {
return true;
}
for &c in &m.get(n).children {
if subtree_has_active_final(inst, m, c) {
return true;
}
}
false
}
fn build_env(inst: &Instance, m: &Machine, scope: NodeId, event_value: &Value) -> Env {
let mut env = Env::new();
for (k, v) in resolve_visible(inst, m, scope) {
env = env.with(k, v);
}
env = env.with("id", Value::Str(inst.id.clone()));
env = env.with("parent", inst.parent.clone().map(Value::Str).unwrap_or(Value::Null));
env = env.with("event", event_value.clone());
env
}
fn event_binding(e: &QueuedEvent) -> Value {
let mut m = BTreeMap::new();
m.insert("type".to_string(), Value::Str(e.etype.clone()));
m.insert("payload".to_string(), e.payload.clone());
Value::Map(m)
}
fn changed_of(event_value: &Value) -> Option<BTreeMap<String, Value>> {
let payload = field(event_value, "payload")?;
let changed = field(&payload, "changed")?;
if let Value::Map(m) = changed {
Some(m)
} else {
None
}
}
fn field(v: &Value, name: &str) -> Option<Value> {
if let Value::Map(m) = v {
m.get(name).cloned()
} else {
None
}
}
fn error_payload(fault: &CelError) -> BTreeMap<String, Value> {
let mut m = BTreeMap::new();
m.insert("message".to_string(), Value::Str(fault.to_string()));
m
}
fn validate_event_payload(m: &Machine, etype: &str, payload: &Value) -> Result<(), String> {
if matches!(etype, "env" | "error" | "done" | "initial" | "entry" | "exit") {
return Ok(());
}
let decl = match m.events.get(etype) {
Some(d) => d,
None => return Err(format!("undeclared event '{etype}'")),
};
let pm = match payload {
Value::Map(m) => m.clone(),
Value::Null => BTreeMap::new(),
_ => return Err("payload must be a map".to_string()),
};
for (name, field) in &decl.payload {
match pm.get(name) {
None => {
if field.required {
return Err(format!("missing required field '{name}'"));
}
}
Some(v) => {
if !v.matches_type(&field.ty) {
return Err(format!("field '{name}' expects {}", field.ty));
}
}
}
}
for name in pm.keys() {
if !decl.payload.iter().any(|(n, _)| n == name) {
return Err(format!("unexpected field '{name}'"));
}
}
Ok(())
}
fn config_of(inst: &Instance, m: &Machine) -> Vec<String> {
let mut v: Vec<String> = active_leaves(inst, m).into_iter().map(|n| m.get(n).id.clone()).collect();
v.sort();
v
}
fn reported_esvs(inst: &Instance, m: &Machine) -> BTreeMap<String, Value> {
let mut leaves = active_leaves(inst, m);
leaves.sort_by_key(|&n| std::cmp::Reverse(m.get(n).depth));
let mut out = BTreeMap::new();
for leaf in leaves {
for (k, v) in resolve_visible(inst, m, leaf) {
out.entry(k).or_insert(v);
}
}
out
}
const RESERVED_LIFECYCLE_EVENTS: &[&str] = &["initial", "entry", "exit", "env", "error", "done"];
fn enabled_events(inst: &Instance, m: &Machine) -> Vec<String> {
let mut set = BTreeSet::new();
for leaf in active_leaves(inst, m) {
for s in ancestors_inclusive(m, leaf) {
for ev in m.get(s).on_events.keys() {
if !RESERVED_LIFECYCLE_EVENTS.contains(&ev.as_str()) {
set.insert(ev.clone());
}
}
}
}
set.into_iter().collect()
}