use crate::engine::Engine;
use crate::error::{Error, Result};
use crate::runtime::RunSummary;
use crate::{Failure, StartOptions};
use fusee_core::record::Status;
use serde_json::{Value, json};
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::oneshot;
pub const MAX_STREAM_WAIT_MS: u64 = 60_000;
pub type CallFuture<'a> = Pin<Box<dyn Future<Output = Result<Value>> + Send + 'a>>;
pub trait Calls: Send + Sync + 'static {
fn call<'a>(&'a self, method: &'a str, args: Value) -> CallFuture<'a>;
}
impl Calls for Engine {
fn call<'a>(&'a self, method: &'a str, args: Value) -> CallFuture<'a> {
Box::pin(async move { run_call(self, method, &args).await })
}
}
pub fn error_kind(e: &Error) -> &'static str {
match e {
Error::NotFound(_) => "not_found",
Error::AlreadyExists(_) => "already_exists",
Error::UnknownWorkflow(_) => "unknown_workflow",
Error::Rejected(_) => "rejected",
Error::Cancelled(_) => "cancelled",
Error::TimedOut => "timed_out",
Error::Terminated => "terminated",
Error::Busy => "busy",
Error::Closed => "closed",
Error::Codec(_) => "codec",
Error::Connection(_) => "connection",
Error::Io(_) => "io",
Error::Determinism(_) => "determinism",
_ => "failed",
}
}
pub fn error_json(e: &Error) -> Value {
let message = match e.failure() {
Some(_) => String::from_utf8_lossy(&e.to_failure().to_payload()).into_owned(),
None => e.to_string(),
};
json!({"error": message, "kind": error_kind(e), "code": e.code()})
}
pub fn decode_error(kind: Option<&str>, msg: &str) -> Error {
match kind {
Some("not_found") => Error::NotFound(msg.to_owned()),
Some("already_exists") => Error::AlreadyExists(msg.to_owned()),
Some("unknown_workflow") => Error::UnknownWorkflow(msg.to_owned()),
Some("rejected") => Error::Rejected(msg.to_owned()),
Some("cancelled") => Error::Cancelled(msg.trim_start_matches("cancelled: ").to_owned()),
Some("timed_out") => Error::TimedOut,
Some("terminated") => Error::Terminated,
Some("busy") => Error::Busy,
Some("closed") => Error::Closed,
Some("codec") => Error::Codec(msg.to_owned()),
Some("connection") => Error::Connection(msg.to_owned()),
Some("io") => Error::Io(msg.to_owned()),
Some("determinism") => Error::Determinism(msg.to_owned()),
_ => Error::from_failure(Failure::from_payload(msg.as_bytes())),
}
}
pub fn decode_answer(answer: &Value) -> Result<Value> {
match answer.get("error") {
Some(e) => Err(decode_error(
answer.get("kind").and_then(Value::as_str),
e.as_str().unwrap_or_default(),
)),
None => Ok(answer.get("value").cloned().unwrap_or(Value::Null)),
}
}
pub fn answer(r: Result<Value>) -> Value {
match r {
Ok(v) => json!({ "value": v }),
Err(e) => error_json(&e),
}
}
fn bytes(v: &Value) -> Arc<[u8]> {
Arc::from(v.to_string().into_bytes())
}
fn parse(p: &[u8]) -> Value {
serde_json::from_slice(p).unwrap_or_else(|_| crate::journal::payload(p))
}
pub fn wire_options(options: &StartOptions) -> Value {
json!({"timeout":options.timeout.map(|t| u64::try_from(t.as_millis()).unwrap_or(u64::MAX)),
"idPolicy":options.id_policy})
}
pub fn start_options(args: &Value) -> Result<StartOptions> {
let opts = &args["options"];
let version = serde_json::from_value(args["version"].clone())?;
let timeout: Option<u64> = serde_json::from_value(opts["timeout"].clone())?;
let id_policy = if opts["idPolicy"].is_null() {
Default::default()
} else {
serde_json::from_value(opts["idPolicy"].clone())?
};
let delay: Option<u64> = serde_json::from_value(opts["delay"].clone())?;
let at: Option<u64> = serde_json::from_value(opts["at"].clone())?;
let trace = opts["trace"]
.as_str()
.filter(|t| is_traceparent(t))
.map(str::to_owned);
Ok(StartOptions {
timeout: timeout.map(Duration::from_millis),
version,
id_policy,
at_ms: at.or(delay.map(|d| crate::local::unix_ms() + d)),
trace,
concurrency_key: opts["concurrencyKey"].as_str().map(str::to_owned),
})
}
pub fn is_traceparent(t: &str) -> bool {
let p: Vec<&str> = t.split('-').collect();
let hex = |s: &str, n: usize| s.len() == n && s.bytes().all(|b| b.is_ascii_hexdigit());
p.len() == 4 && hex(p[0], 2) && hex(p[1], 32) && hex(p[2], 16) && hex(p[3], 2)
}
fn update_result((status, payload): (Status, Arc<[u8]>)) -> Result<Value> {
if status == Status::Ok {
Ok(parse(&payload))
} else {
Err(Error::from_failure(Failure::from_payload(&payload)))
}
}
pub fn summary_json(r: &RunSummary) -> Value {
json!({"id":r.id,"type":r.wtype,"version":r.version, "status":r.status,"shard":r.shard,
"run":r.run,"startedMs":r.started_ms,"finishedMs":r.finished_ms, "error":r.error,
"attributes":serde_json::from_str::<Value>(&r.attributes).unwrap_or(Value::Null),
"object": r.object})
}
pub fn base64(b: &[u8]) -> String {
const T: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
let mut out = String::with_capacity(b.len().div_ceil(3) * 4);
for c in b.chunks(3) {
let n = (u32::from(c[0]) << 16)
| (u32::from(*c.get(1).unwrap_or(&0)) << 8)
| u32::from(*c.get(2).unwrap_or(&0));
out.push(T[(n >> 18) as usize & 63] as char);
out.push(T[(n >> 12) as usize & 63] as char);
out.push(if c.len() > 1 {
T[(n >> 6) as usize & 63] as char
} else {
'='
});
out.push(if c.len() > 2 {
T[n as usize & 63] as char
} else {
'='
});
}
out
}
pub fn unbase64(s: &str) -> Option<Vec<u8>> {
let val = |c: u8| -> Option<u32> {
Some(match c {
b'A'..=b'Z' => c - b'A',
b'a'..=b'z' => c - b'a' + 26,
b'0'..=b'9' => c - b'0' + 52,
b'+' => 62,
b'/' => 63,
_ => return None,
} as u32)
};
let s: Vec<u8> = s.bytes().filter(|b| !b.is_ascii_whitespace()).collect();
let mut out = Vec::with_capacity(s.len() / 4 * 3);
for c in s.chunks(4) {
let pad = c.iter().filter(|&&b| b == b'=').count();
let mut n = 0u32;
for (i, &b) in c.iter().enumerate() {
n |= if b == b'=' { 0 } else { val(b)? } << (18 - 6 * i);
}
out.push((n >> 16) as u8);
if pad < 2 {
out.push((n >> 8) as u8);
}
if pad < 1 {
out.push(n as u8);
}
}
Some(out)
}
pub async fn run_call(engine: &Engine, method: &str, a: &Value) -> Result<Value> {
match a.get("by").and_then(Value::as_str) {
Some(by) if AUDITED.contains(&method) && !by.is_empty() => {
crate::audit::as_requester(by, run_call_as(engine, method, a)).await
}
_ => run_call_as(engine, method, a).await,
}
}
async fn run_call_as(engine: &Engine, method: &str, a: &Value) -> Result<Value> {
let s = |k: &str| {
a.get(k)
.and_then(Value::as_str)
.unwrap_or_default()
.to_owned()
};
let v = |k: &str| a.get(k).cloned().unwrap_or(Value::Null);
match method {
"complete" => engine
.complete(&s("token"), &v("value"))
.await
.map(|()| Value::Null),
"fail" => {
let error = match a.get("error") {
Some(Value::String(t)) => t.clone(),
Some(other) => other.to_string(),
None => String::new(),
};
engine.fail(&s("token"), &error).await.map(|()| Value::Null)
}
"start" => {
let mut opts = start_options(a)?;
match opts.at_ms.take() {
Some(at) => engine
.start_delayed(&s("type"), &s("id"), bytes(&v("input")), at, opts)
.await
.map(|()| Value::Null),
None => engine
.start_raw(&s("type"), &s("id"), bytes(&v("input")), opts)
.await
.map(|()| Value::Null),
}
}
"signal" => engine
.signal_raw(&s("id"), &s("name"), bytes(&v("value")))
.await
.map(|()| Value::Null),
"signalWithStart" => engine
.signal_with_start_raw(
&s("type"),
&s("id"),
bytes(&v("input")),
start_options(a)?,
&s("name"),
bytes(&v("value")),
)
.await
.map(|()| Value::Null),
"updateWithStart" => update_result(
engine
.update_with_start_raw(
&s("type"),
&s("id"),
bytes(&v("input")),
start_options(a)?,
&s("name"),
&s("updateId"),
bytes(&v("arg")),
)
.await?,
),
"update" => update_result(
engine
.update_raw(&s("id"), &s("name"), &s("updateId"), bytes(&v("arg")))
.await?,
),
"reset" => engine
.reset(
&s("id"),
&s("beforeStep"),
a["reapplySignals"].as_bool().unwrap_or(true),
)
.await
.map(|_| Value::Null),
"list" => {
let limit = serde_json::from_value::<Option<usize>>(a["limit"].clone())?.unwrap_or(100);
let filter = a.get("where").cloned().unwrap_or_else(|| json!({}));
let results = match serde_json::from_value::<Option<Vec<u32>>>(a["shards"].clone())? {
Some(shards) => engine.list_where_in(&shards, &filter, limit).await?,
None => engine.list_where(&filter, limit).await?,
};
Ok(Value::Array(results.iter().map(summary_json).collect()))
}
"describe" => match engine.describe(&s("id")).await? {
None => Err(Error::NotFound(s("id"))),
Some(d) => {
let mut out = summary_json(&d.summary);
out["journal"] = Value::Array(
d.journal
.iter()
.map(crate::journal::normalize_record)
.collect(),
);
if let Some(t) = crate::journal::step_times(&d.journal) {
out["stepTimes"] = t;
}
if let Some(o) = &d.object {
out["objectDetail"] = json!({"state": o.state, "version": o.version,
"running": o.running, "waiting": o.waiting});
}
Ok(out)
}
},
"query" => engine
.query_raw(&s("id"), &s("name"), bytes(&v("arg")))
.await
.map(|p| parse(&p)),
"result" => {
let (status, p) = engine.result_raw(&s("id")).await?;
match status {
Status::Ok => Ok(parse(&p)),
Status::Cancelled => Err(Error::Cancelled(String::from_utf8_lossy(&p).into())),
Status::TimedOut => Err(Error::TimedOut),
Status::Terminated => Err(Error::Terminated),
_ => Err(Error::from_failure(Failure::from_payload(&p))),
}
}
"cancel" => engine
.run::<Value>(&s("id"))
.cancel(&s("reason"))
.await
.map(|()| Value::Null),
"resume" => engine
.run::<Value>(&s("id"))
.resume()
.await
.map(|()| Value::Null),
"editState" => Box::pin(run_call(engine, "updateWithStart", &edit_state_call(a)?)).await,
"terminate" => engine
.run::<Value>(&s("id"))
.terminate(&s("reason"))
.await
.map(|()| Value::Null),
"streamRead" => {
let wait = a["waitMs"]
.as_u64()
.unwrap_or(crate::stream::WAIT_MS)
.min(MAX_STREAM_WAIT_MS);
let p = engine
.stream_read_raw(
&s("id"),
a["run"].as_u64(),
&s("name"),
a["from"].as_u64().unwrap_or(0),
Duration::from_millis(wait),
)
.await?;
Ok(serde_json::from_slice(&p)?)
}
"types" => Ok(engine.types_json()),
"metrics" => {
let shards = engine.shard_metrics().await?;
let total = engine.metrics().await?;
let mut total = serde_json::to_value(total)?;
let mut shards = shards
.iter()
.map(serde_json::to_value)
.collect::<std::result::Result<Vec<_>, _>>()?;
for (status, key) in [
("stuck", "runs_stuck"),
("paused", "runs_paused"),
("waiting_for_code", "runs_waiting_for_code"),
] {
let mut per = vec![0u64; shards.len()];
for r in engine
.list_where(&json!({ "status": status }), usize::MAX)
.await?
{
if let Some(n) = per.get_mut(r.shard as usize) {
*n += 1;
}
}
total[key] = json!(per.iter().sum::<u64>());
for (s, n) in shards.iter_mut().zip(per) {
s[key] = json!(n);
}
}
Ok(json!({ "total": total, "shards": shards }))
}
"info" => Ok(engine.info_json()),
"limits" => Ok(serde_json::to_value(engine.limits())?),
"export" => match engine.describe(&s("id")).await? {
None => Err(Error::NotFound(s("id"))),
Some(d) => Ok(json!({
"file": base64(&crate::export::ExportedJournal::from_detail(&d).encode()),
"name": crate::export::file_name(&s("id")),
})),
},
other => Err(Error::Rejected(format!("unknown call '{other}'"))),
}
}
pub const AUDITED: &[&str] = &[
"signal",
"signalWithStart",
"cancel",
"resume",
"terminate",
"reset",
"editState",
];
pub fn edit_state_call(a: &Value) -> Result<Value> {
let id = a["id"].as_str().unwrap_or_default();
let Some((wtype, key)) = id
.split_once('/')
.filter(|(t, k)| !t.is_empty() && !k.is_empty())
else {
return Err(Error::Rejected(format!(
"editState takes the key's workflow ID, <type>/<key>, not '{id}'"
)));
};
let mut edit = a["edit"].clone();
if !edit.is_object() {
return Err(Error::Rejected("editState's edit is a JSON object".into()));
}
if let Some(by) = a["by"].as_str().filter(|b| !b.is_empty()) {
edit["by"] = json!(by);
}
Ok(json!({
"type": wtype,
"id": crate::object::object_id(wtype, key),
"input": null,
"name": crate::object::EDIT_HANDLER,
"arg": edit,
"updateId": a["updateId"].as_str().unwrap_or_default(),
"version": null,
"options": {"timeout": null, "idPolicy": {"running": "use-existing", "finished": "allow"}},
}))
}
pub struct Stamped {
inner: Arc<dyn Calls>,
by: String,
trace: Option<String>,
}
impl Stamped {
pub fn new(inner: Arc<dyn Calls>, by: String, trace: Option<String>) -> Self {
Self { inner, by, trace }
}
}
impl Calls for Stamped {
fn call<'a>(&'a self, method: &'a str, mut args: Value) -> CallFuture<'a> {
if AUDITED.contains(&method) && args.is_object() {
args["by"] = json!(self.by);
}
if let Some(t) = &self.trace
&& matches!(method, "start" | "signalWithStart" | "updateWithStart")
&& args.is_object()
{
if !args["options"].is_object() {
args["options"] = json!({});
}
if args["options"]["trace"].is_null() {
args["options"]["trace"] = json!(t);
}
}
self.inner.call(method, args)
}
}
pub const OUTSIDE_NAMESPACE: &str = "this token reaches namespace";
pub struct Namespaced {
inner: Arc<dyn Calls>,
namespace: String,
prefix: String,
}
impl Namespaced {
pub fn new(inner: Arc<dyn Calls>, namespace: &str) -> Self {
let namespace = namespace.trim_end_matches('/').to_owned();
let prefix = format!("{namespace}/");
Self {
inner,
namespace,
prefix,
}
}
fn owns(&self, id: &str) -> bool {
id.len() > self.prefix.len() && id.starts_with(&self.prefix)
}
fn refuse(&self, why: &str) -> Error {
Error::Rejected(format!(
"{OUTSIDE_NAMESPACE} '{}' only: {why}",
self.namespace
))
}
}
impl Calls for Namespaced {
fn call<'a>(&'a self, method: &'a str, mut args: Value) -> CallFuture<'a> {
Box::pin(async move {
match method {
"start" | "signal" | "signalWithStart" | "updateWithStart" | "update" | "reset"
| "describe" | "query" | "result" | "cancel" | "resume" | "terminate"
| "streamRead" | "export" | "editState" => {
let id = args.get("id").and_then(Value::as_str).unwrap_or_default();
if !self.owns(id) {
return Err(self.refuse(&format!(
"workflow ID '{id}' does not start with '{}'",
self.prefix
)));
}
if matches!(method, "start" | "signalWithStart" | "updateWithStart")
&& let Some(t) = args.get("type").and_then(Value::as_str)
&& t.starts_with("fusee.")
{
return Err(self
.refuse(&format!("the engine's own type '{t}' needs the full token")));
}
}
"complete" | "fail" => {
let token = args
.get("token")
.and_then(Value::as_str)
.unwrap_or_default();
match crate::runtime::parse_token(token) {
Some((id, ..)) if self.owns(&id) => {}
Some(_) => {
return Err(self.refuse("the completion is for a run outside it"));
}
None => return Err(Error::Rejected("not a completion token".into())),
}
}
"list" => {
if !args.is_object() {
args = json!({});
}
let mut filter = match args.get("where") {
Some(Value::Object(m)) => m.clone(),
_ => Default::default(),
};
filter.entry("id").or_insert_with(
|| json!({"gte": self.prefix, "lt": format!("{}0", self.namespace)}),
);
args["where"] = Value::Object(filter);
let rows = self.inner.call(method, args).await?;
return Ok(Value::Array(
rows.as_array()
.into_iter()
.flatten()
.filter(|r| r["id"].as_str().is_some_and(|id| self.owns(id)))
.cloned()
.collect(),
));
}
"info" => {
let mut info = self.inner.call(method, args).await?;
info["namespace"] = json!(self.namespace);
return Ok(info);
}
"types" | "limits" => {}
other => return Err(self.refuse(&format!("'{other}' needs the full token"))),
}
self.inner.call(method, args).await
})
}
}
pub struct RelayedCall {
pub id: u64,
pub method: String,
pub args: Value,
}
struct RelayState {
next: u64,
queue: std::collections::VecDeque<RelayedCall>,
waiting: rustc_hash::FxHashMap<u64, oneshot::Sender<Value>>,
closed: bool,
}
#[derive(Clone)]
pub struct Relay {
state: Arc<Mutex<RelayState>>,
wake: Arc<dyn Fn() + Send + Sync>,
}
pub const RELAY_BASE: u64 = 1 << 62;
impl Relay {
pub fn new(wake: Arc<dyn Fn() + Send + Sync>) -> Self {
Self {
state: Arc::new(Mutex::new(RelayState {
next: RELAY_BASE,
queue: Default::default(),
waiting: Default::default(),
closed: false,
})),
wake,
}
}
fn lock(&self) -> std::sync::MutexGuard<'_, RelayState> {
self.state.lock().unwrap_or_else(|e| e.into_inner())
}
pub fn next(&self) -> Option<RelayedCall> {
self.lock().queue.pop_front()
}
pub fn reply(&self, id: u64, answer: Value) -> bool {
match self.lock().waiting.remove(&id) {
Some(tx) => tx.send(answer).is_ok(),
None => false,
}
}
pub fn owns(id: u64) -> bool {
id >= RELAY_BASE
}
pub fn close(&self) {
let mut st = self.lock();
st.closed = true;
st.queue.clear();
st.waiting.clear();
}
}
impl Calls for Relay {
fn call<'a>(&'a self, method: &'a str, args: Value) -> CallFuture<'a> {
Box::pin(async move {
let (method, args) = if method == "editState" {
("updateWithStart", edit_state_call(&args)?)
} else {
(method, args)
};
let rx = {
let mut st = self.lock();
if st.closed {
return Err(Error::Closed);
}
let id = st.next;
st.next += 1;
let (tx, rx) = oneshot::channel();
st.waiting.insert(id, tx);
st.queue.push_back(RelayedCall {
id,
method: method.to_owned(),
args,
});
rx
};
(self.wake)();
match rx.await {
Ok(answer) => decode_answer(&answer),
Err(_) => Err(Error::Closed),
}
})
}
}