use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;
use vertep_core::compile::{CompiledChart, ROOT, StateIdx};
use vertep_core::datamodel::Value;
use vertep_core::ir::{
Block, Chart, Exec, ExecKind, Extension, ExtensionValue, Send as SendElem, State, StateElement,
Transition, VT_NS,
};
use vertep_core::kernel::History;
use vertep_core::session::Session;
use vertep_rdf::ChartVersion;
use crate::vocab;
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub enum UnhandledPolicy {
Discard,
DeadLetter,
Error,
}
impl UnhandledPolicy {
pub fn as_str(self) -> &'static str {
match self {
UnhandledPolicy::Discard => "discard",
UnhandledPolicy::DeadLetter => "dead-letter",
UnhandledPolicy::Error => "error",
}
}
pub fn parse(s: &str) -> Option<Self> {
match s.trim() {
"discard" => Some(UnhandledPolicy::Discard),
"dead-letter" => Some(UnhandledPolicy::DeadLetter),
"error" => Some(UnhandledPolicy::Error),
_ => None,
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum RestartMode {
Never,
OnFailure,
Always,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RestartPolicy {
pub mode: RestartMode,
pub limit: u32,
pub window_ms: i64,
pub backoff_ms: i64,
}
impl Default for RestartPolicy {
fn default() -> Self {
RestartPolicy {
mode: RestartMode::Never,
limit: 3,
window_ms: 60_000,
backoff_ms: 100,
}
}
}
impl RestartPolicy {
pub fn backoff(&self, n: u32) -> i64 {
self.backoff_ms
.saturating_mul(1i64 << n.saturating_sub(1).min(30))
}
}
pub fn parse_duration(text: &str) -> Option<i64> {
let t = text.trim();
let units: [(&str, f64); 5] = [
("ms", 1.0),
("min", 60_000.0),
("s", 1_000.0),
("h", 3_600_000.0),
("d", 86_400_000.0),
];
for (suffix, factor) in units {
if let Some(n) = t.strip_suffix(suffix) {
let v: f64 = n.trim().parse().ok()?;
return (v >= 0.0).then(|| (v * factor).round() as i64);
}
}
t.parse::<i64>().ok().filter(|v| *v >= 0)
}
pub fn element_attributes(xml: &str) -> BTreeMap<String, String> {
let mut out = BTreeMap::new();
let Some(start) = xml.find('<') else {
return out;
};
let rest = &xml[start + 1..];
let end = rest.find('>').unwrap_or(rest.len());
let tag = rest[..end].trim_end_matches('/');
let mut chars = tag.char_indices().peekable();
while let Some((_, c)) = chars.peek() {
if c.is_whitespace() {
break;
}
chars.next();
}
let bytes = tag;
let mut i = chars.peek().map_or(bytes.len(), |(i, _)| *i);
while i < bytes.len() {
let rest = &bytes[i..];
let trimmed = rest.trim_start();
i += rest.len() - trimmed.len();
let Some(eq) = trimmed.find('=') else { break };
let name = trimmed[..eq].trim().to_owned();
let after = trimmed[eq + 1..].trim_start();
let Some(q) = after.chars().next().filter(|c| *c == '"' || *c == '\'') else {
break;
};
let body = &after[1..];
let Some(close) = body.find(q) else { break };
out.insert(
name.rsplit(':').next().unwrap_or(&name).to_owned(),
unescape(&body[..close]),
);
let consumed = (trimmed.len() - after.len()) + 1 + close + 1;
i += consumed;
}
out
}
fn unescape(s: &str) -> String {
s.replace("<", "<")
.replace(">", ">")
.replace(""", "\"")
.replace("'", "'")
.replace("&", "&")
}
fn vt_attr<'a>(exts: &'a [Extension], local: &str) -> Option<&'a str> {
exts.iter().find_map(|e| match &e.value {
ExtensionValue::Attribute(v) if e.ns == VT_NS && e.local == local => Some(v.as_str()),
_ => None,
})
}
fn vt_elements<'a>(
exts: &'a [Extension],
local: &'a str,
) -> impl Iterator<Item = BTreeMap<String, String>> + 'a {
exts.iter().filter_map(move |e| match &e.value {
ExtensionValue::Element(xml) if e.ns == VT_NS && e.local == local => {
Some(element_attributes(xml))
}
_ => None,
})
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LowerError(pub String);
impl core::fmt::Display for LowerError {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
write!(f, "invalid vt: extension: {}", self.0)
}
}
pub fn request_type(timeout_ms: Option<i64>) -> String {
match timeout_ms {
Some(t) => format!("{};timeout={t}", vocab::REQUEST_SEND_TYPE),
None => vocab::REQUEST_SEND_TYPE.to_owned(),
}
}
pub fn split_type(type_: &str) -> (&str, BTreeMap<String, String>) {
let mut parts = type_.split(';');
let base = parts.next().unwrap_or_default();
let params = parts
.map(|p| {
let (k, v) = p.split_once('=').unwrap_or((p, ""));
(k.to_owned(), v.to_owned())
})
.collect();
(base, params)
}
pub fn lower(chart: &Chart) -> Result<(Chart, BTreeSet<String>), LowerError> {
let mut out = chart.clone();
let mut types = BTreeSet::new();
for s in &mut out.states {
lower_state(s, &mut types)?;
}
Ok((out, types))
}
fn lower_state(s: &mut State, types: &mut BTreeSet<String>) -> Result<(), LowerError> {
let requests: Vec<_> = vt_elements(&s.extensions, "request").collect();
for (k, r) in requests.iter().enumerate() {
let event = r
.get("event")
.ok_or_else(|| LowerError(format!("vt:request in state {} has no event", s.id)))?;
let timeout = match r.get("timeout") {
Some(t) => Some(parse_duration(t).ok_or_else(|| {
LowerError(format!("vt:request in state {}: bad timeout {t:?}", s.id))
})?),
None => None,
};
let ty = request_type(timeout);
types.insert(ty.clone());
let send = SendElem {
event: Some(event.clone()),
target: r.get("target").cloned(),
type_: Some(ty),
..SendElem::default()
};
s.onentry
.push(vec![Exec::new(ExecKind::Send(Box::new(send)))]);
if k == 0 && s.element == StateElement::State && s.children.is_empty() {
s.children.push(State::new(
format!("{}.vt-waiting", s.id),
StateElement::State,
));
}
if let Some(target) = r.get("reply") {
s.transitions.push(Transition {
events: vec![vocab::REPLY_EVENT.to_owned()],
targets: vec![target.clone()],
..Transition::default()
});
}
if let Some(target) = r.get("ontimeout") {
s.transitions.push(Transition {
events: vec![vocab::TIMEOUT_EVENT.to_owned()],
targets: vec![target.clone()],
..Transition::default()
});
}
}
for b in s.onentry.iter_mut().chain(s.onexit.iter_mut()) {
lower_block(b, types)?;
}
for t in &mut s.transitions {
lower_block(&mut t.body, types)?;
}
for c in &mut s.children {
lower_state(c, types)?;
}
Ok(())
}
fn lower_block(b: &mut Block, types: &mut BTreeSet<String>) -> Result<(), LowerError> {
for e in b.iter_mut() {
match &mut e.kind {
ExecKind::If { branches } => {
for br in branches {
lower_block(&mut br.body, types)?;
}
}
ExecKind::Foreach { body, .. } => lower_block(body, types)?,
ExecKind::Extension(x) if x.ns == VT_NS => {
let ExtensionValue::Element(xml) = &x.value else {
continue;
};
let a = element_attributes(xml);
let send = match x.local.as_str() {
"reminder" => {
let name = a
.get("name")
.ok_or_else(|| LowerError("vt:reminder has no name".into()))?;
let ty = if a.get("cancel").is_some_and(|c| c == "true") {
format!("{};name={name};cancel=true", vocab::REMINDER_SEND_TYPE)
} else {
let period = a
.get("period")
.and_then(|p| parse_duration(p))
.filter(|p| *p > 0)
.ok_or_else(|| {
LowerError(format!("vt:reminder {name}: bad period"))
})?;
format!("{};name={name};period={period}", vocab::REMINDER_SEND_TYPE)
};
SendElem {
event: Some(a.get("event").cloned().unwrap_or_else(|| name.clone())),
type_: Some(ty),
..SendElem::default()
}
}
"migrate" => {
let to = a
.get("to")
.ok_or_else(|| LowerError("vt:migrate has no `to`".into()))?;
let m = a.get("migration").cloned().unwrap_or_default();
SendElem {
event: Some("vt:migrate".into()),
type_: Some(format!(
"{};to={to};migration={m}",
vocab::MIGRATE_SEND_TYPE
)),
..SendElem::default()
}
}
_ => continue,
};
if let Some(t) = &send.type_ {
types.insert(t.clone());
}
e.kind = ExecKind::Send(Box::new(send));
}
_ => {}
}
}
Ok(())
}
pub struct ChartInfo {
pub version: ChartVersion,
pub chart_type: Option<String>,
pub ir: Chart,
pub compiled: CompiledChart,
pub vt_send_types: BTreeSet<String>,
pub defers: BTreeMap<StateIdx, Vec<String>>,
pub unhandled: Option<UnhandledPolicy>,
}
impl std::fmt::Debug for ChartInfo {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ChartInfo")
.field("version", &self.version.as_str())
.field("chart_type", &self.chart_type)
.finish_non_exhaustive()
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RegisterError {
BadType(String),
Lower(LowerError),
Invalid(Vec<String>),
InvokesVirtualActor {
state: String,
src: String,
},
Refused {
datamodel: String,
problems: Vec<String>,
},
}
impl core::fmt::Display for RegisterError {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
RegisterError::BadType(t) => write!(f, "chart type {t:?} is not an absolute IRI"),
RegisterError::Lower(e) => write!(f, "{e}"),
RegisterError::Invalid(v) => write!(f, "chart is invalid: {}", v.join("; ")),
RegisterError::InvokesVirtualActor { state, src } => write!(
f,
"state {state} invokes the virtual actor {src}; virtual actors are not invokable"
),
RegisterError::Refused {
datamodel,
problems,
} => write!(
f,
"the {datamodel} datamodel refuses the chart: {}",
problems.join("; ")
),
}
}
}
impl std::error::Error for RegisterError {}
impl ChartInfo {
pub fn new(chart: Chart, chart_type: Option<String>) -> Result<Self, RegisterError> {
let version = vertep_rdf::chart_version(&chart);
let (lowered, vt_send_types) = lower(&chart).map_err(RegisterError::Lower)?;
let compiled = CompiledChart::new(lowered)
.map_err(|v| RegisterError::Invalid(v.iter().map(|x| format!("{x:?}")).collect()))?;
let mut defers = BTreeMap::new();
collect_defers(&compiled.chart().states, &compiled, &mut defers);
let unhandled = vt_attr(&chart.extensions, "unhandled").and_then(UnhandledPolicy::parse);
Ok(ChartInfo {
version,
chart_type,
ir: chart,
compiled,
vt_send_types,
defers,
unhandled,
})
}
pub fn deferred(&self, session: &Session) -> Vec<String> {
let mut out = Vec::new();
for (idx, ds) in &self.defers {
if session.configuration.contains(*idx) {
out.extend(ds.iter().cloned());
}
}
out
}
pub fn restart_policy(&self, state: &str, index: usize) -> RestartPolicy {
let mut p = RestartPolicy::default();
let Some(inv) = find_state(&self.ir.states, state).and_then(|s| s.invokes.get(index))
else {
return p;
};
let x = &inv.extensions;
p.mode = match vt_attr(x, "restart") {
Some("on-failure") => RestartMode::OnFailure,
Some("always") => RestartMode::Always,
_ => RestartMode::Never,
};
if let Some(l) = vt_attr(x, "restart-limit").and_then(|v| v.trim().parse().ok()) {
p.limit = l;
}
if let Some(w) = vt_attr(x, "restart-window").and_then(parse_duration) {
p.window_ms = w;
}
if let Some(b) = vt_attr(x, "restart-backoff").and_then(parse_duration) {
p.backoff_ms = b;
}
p
}
pub fn active(&self, session: &Session) -> Vec<String> {
session.active(&self.compiled)
}
}
fn collect_defers(
states: &[State],
chart: &CompiledChart,
out: &mut BTreeMap<StateIdx, Vec<String>>,
) {
for s in states {
if let Some(d) = vt_attr(&s.extensions, "defer")
&& let Some(i) = chart.index_of(&s.id)
{
out.insert(i, d.split_whitespace().map(str::to_owned).collect());
}
collect_defers(&s.children, chart, out);
}
}
pub fn find_state<'a>(states: &'a [State], id: &str) -> Option<&'a State> {
for s in states {
if s.id == id {
return Some(s);
}
if let Some(f) = find_state(&s.children, id) {
return Some(f);
}
}
None
}
pub fn static_invoke_sources(states: &[State], out: &mut Vec<(String, String)>) {
for s in states {
for inv in &s.invokes {
if let Some(src) = &inv.src {
out.push((s.id.clone(), src.clone()));
}
}
static_invoke_sources(&s.children, out);
}
}
pub type Transform = Arc<dyn Fn(Value) -> Result<Value, String> + Send + Sync>;
#[derive(Clone)]
pub struct Migration {
pub iri: String,
pub from: ChartVersion,
pub to: ChartVersion,
pub configurations: Vec<(BTreeSet<String>, BTreeSet<String>)>,
pub states: BTreeMap<String, String>,
pub transform: Transform,
}
impl std::fmt::Debug for Migration {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Migration")
.field("iri", &self.iri)
.field("from", &self.from.as_str())
.field("to", &self.to.as_str())
.finish_non_exhaustive()
}
}
impl Migration {
pub fn identity(iri: impl Into<String>, from: ChartVersion, to: ChartVersion) -> Self {
Migration {
iri: iri.into(),
from,
to,
configurations: Vec::new(),
states: BTreeMap::new(),
transform: Arc::new(Ok),
}
}
pub fn map_configuration<I: IntoIterator<Item = S>, S: Into<String>>(
mut self,
from: I,
to: I,
) -> Self {
self.configurations.push((
from.into_iter().map(Into::into).collect(),
to.into_iter().map(Into::into).collect(),
));
self
}
pub fn rename_state(mut self, from: impl Into<String>, to: impl Into<String>) -> Self {
self.states.insert(from.into(), to.into());
self
}
pub fn with_transform(
mut self,
f: impl Fn(Value) -> Result<Value, String> + Send + Sync + 'static,
) -> Self {
self.transform = Arc::new(f);
self
}
pub fn map(&self, old: &BTreeSet<String>, to: &ChartInfo) -> Option<BTreeSet<String>> {
let core: BTreeSet<String> = old.iter().filter(|s| !is_generated(s)).cloned().collect();
if let Some((_, n)) = self.configurations.iter().find(|(o, _)| *o == core) {
return Some(n.clone());
}
let mut out = BTreeSet::new();
for s in &core {
let n = self.states.get(s).unwrap_or(s);
to.compiled.index_of(n)?;
out.insert(n.clone());
}
Some(out)
}
}
fn is_generated(id: &str) -> bool {
id.ends_with(".vt-waiting")
}
pub fn reachable_configurations(
chart: &Chart,
limit: usize,
) -> Result<Vec<BTreeSet<String>>, usize> {
let mut all = Vec::new();
for s in &chart.states {
if matches!(s.element, StateElement::History(_)) {
continue;
}
all.extend(configs(s, limit)?);
if all.len() > limit {
return Err(all.len());
}
}
Ok(all)
}
fn configs(s: &State, limit: usize) -> Result<Vec<BTreeSet<String>>, usize> {
let kids: Vec<&State> = s
.children
.iter()
.filter(|c| !matches!(c.element, StateElement::History(_)))
.collect();
let mut out: Vec<BTreeSet<String>> = Vec::new();
if kids.is_empty() {
out.push(BTreeSet::from([s.id.clone()]));
} else if s.element == StateElement::Parallel {
out.push(BTreeSet::from([s.id.clone()]));
for k in kids {
let sub = configs(k, limit)?;
let mut next = Vec::with_capacity(out.len() * sub.len());
for a in &out {
for b in &sub {
let mut c = a.clone();
c.extend(b.iter().cloned());
next.push(c);
}
if next.len() > limit {
return Err(next.len());
}
}
out = next;
}
} else {
for k in kids {
for mut c in configs(k, limit)? {
c.insert(s.id.clone());
out.push(c);
}
if out.len() > limit {
return Err(out.len());
}
}
}
Ok(out)
}
pub fn unmapped_configurations(
m: &Migration,
from: &ChartInfo,
to: &ChartInfo,
) -> Result<Vec<BTreeSet<String>>, usize> {
Ok(reachable_configurations(&from.ir, 100_000)?
.into_iter()
.filter(|c| m.map(c, to).is_none())
.collect())
}
pub fn migrated_session(old: &Session, ids: &BTreeSet<String>, to: &ChartInfo) -> Session {
let c = &to.compiled;
let mut config = c.empty_set();
for id in ids {
if let Some(i) = c.index_of(id) {
config.insert(i);
let mut p = c.state(i).parent;
while let Some(a) = p {
if a == ROOT {
break;
}
config.insert(a);
p = c.state(a).parent;
}
}
}
complete(c, ROOT, &mut config);
let mut data = config.clone();
data.insert(ROOT);
let mut options = old.options.clone();
for t in &to.vt_send_types {
if !options.send_types.contains(t) {
options.send_types.push(t.clone());
}
}
Session {
sessionid: old.sessionid.clone(),
status: old.status,
configuration: config,
history: History::new(c.len()),
internal_queue: Default::default(),
invocations: Vec::new(),
data_initialized: data,
send_counter: old.send_counter,
invoke_counter: old.invoke_counter,
macrosteps: old.macrosteps,
options,
}
}
fn complete(c: &CompiledChart, s: StateIdx, config: &mut vertep_core::bitset::StateSet) {
let st = c.state(s);
if st.children.is_empty() {
return;
}
let parallel = st.kind == vertep_core::ir::StateKind::Parallel;
let active: Vec<StateIdx> = st
.children
.iter()
.copied()
.filter(|k| config.contains(*k))
.collect();
if parallel {
for k in st.children.clone() {
config.insert(k);
complete(c, k, config);
}
} else if active.is_empty() {
if let Some(&k) = c.default_entry(s).first().or(st.children.first()) {
config.insert(k);
let mut p = c.state(k).parent;
while let Some(a) = p {
if a == s {
break;
}
config.insert(a);
p = c.state(a).parent;
}
complete(c, k, config);
}
} else {
for k in active {
complete(c, k, config);
}
}
}
#[derive(Default)]
pub struct ChartRegistry {
versions: BTreeMap<String, Arc<ChartInfo>>,
types: BTreeMap<String, Vec<String>>,
migrations: BTreeMap<String, Migration>,
bindings: BTreeMap<String, (String, String)>,
}
impl std::fmt::Debug for ChartRegistry {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ChartRegistry")
.field("versions", &self.versions.keys().collect::<Vec<_>>())
.field("types", &self.types)
.finish_non_exhaustive()
}
}
impl ChartRegistry {
pub fn insert(&mut self, info: ChartInfo) -> Arc<ChartInfo> {
let key = info.version.as_str().to_owned();
let ty = info.chart_type.clone();
let info = self
.versions
.entry(key.clone())
.or_insert_with(|| Arc::new(info))
.clone();
if let Some(t) = ty {
let list = self.types.entry(t).or_default();
list.retain(|v| *v != key);
list.push(key);
}
info
}
pub fn get(&self, iri: &str) -> Option<Arc<ChartInfo>> {
self.versions.get(iri).cloned()
}
pub fn latest(&self, ty: &str) -> Option<Arc<ChartInfo>> {
self.types
.get(ty)
.and_then(|l| l.last())
.and_then(|v| self.get(v))
}
pub fn versions_of(&self, ty: &str) -> Vec<String> {
self.types.get(ty).cloned().unwrap_or_default()
}
pub fn types(&self) -> impl Iterator<Item = &String> {
self.types.keys()
}
pub fn type_of(&self, address: &str) -> Option<(String, String)> {
if let Some(b) = self.bindings.get(address) {
return Some(b.clone());
}
self.types
.keys()
.filter_map(|t| {
let rest = address.strip_prefix(t.as_str())?.strip_prefix('/')?;
Some((t, rest))
})
.max_by_key(|(t, _)| t.len())
.and_then(|(t, k)| Some((t.clone(), crate::address::percent_decode(k)?)))
}
pub fn bind(&mut self, address: &str, chart_type: &str, key: &str) {
self.bindings
.insert(address.to_owned(), (chart_type.to_owned(), key.to_owned()));
}
pub fn remove(&mut self, iri: &str) -> bool {
for l in self.types.values_mut() {
l.retain(|v| v != iri);
}
self.versions.remove(iri).is_some()
}
pub fn add_migration(&mut self, m: Migration) {
self.migrations.insert(m.iri.clone(), m);
}
pub fn migration(&self, iri: &str) -> Option<Migration> {
self.migrations.get(iri).cloned()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn durations() {
assert_eq!(parse_duration("100ms"), Some(100));
assert_eq!(parse_duration("5s"), Some(5_000));
assert_eq!(parse_duration("10min"), Some(600_000));
assert_eq!(parse_duration("1h"), Some(3_600_000));
assert_eq!(parse_duration("250"), Some(250));
assert_eq!(parse_duration("x"), None);
}
#[test]
fn attributes_of_an_extension_element() {
let a = element_attributes(
r#"<vt:request xmlns:vt="https://w3id.org/vertep/vt#" target="urn:b" event='price.get' timeout="5s"/>"#,
);
assert_eq!(a.get("target").map(String::as_str), Some("urn:b"));
assert_eq!(a.get("event").map(String::as_str), Some("price.get"));
assert_eq!(a.get("timeout").map(String::as_str), Some("5s"));
assert_eq!(
split_type("x;a=1;b=2").1.get("b").map(String::as_str),
Some("2")
);
}
#[test]
fn backoff_doubles() {
let p = RestartPolicy {
backoff_ms: 100,
..RestartPolicy::default()
};
assert_eq!((p.backoff(1), p.backoff(2), p.backoff(3)), (100, 200, 400));
}
}