#[cfg(any(feature = "rusqlite", feature = "postgres"))]
pub mod backtrace;
pub mod cas;
pub use deployment_config::component_id;
mod error_conversions;
#[cfg(feature = "postgres")]
mod postgres_ext;
#[cfg(feature = "rusqlite")]
mod rusqlite_ext;
pub mod storage;
pub mod time;
use ::serde::{Deserialize, Serialize};
pub use deployment_config::component_id::{
ComponentId, ComponentType, ContentDigest, InvalidNameError, check_name,
};
use indexmap::IndexMap;
use opentelemetry::propagation::{Extractor, Injector};
pub use prefixed_ulid::ExecutionId;
use serde_json::Value;
use std::collections::hash_map::DefaultHasher;
use std::hash::BuildHasherDefault;
use std::{
fmt::{Debug, Display},
hash::Hash,
str::FromStr,
sync::Arc,
time::Duration,
};
use storage::{PendingStateFinishedError, PendingStateFinishedResultKind};
use tracing::{Span, error};
use val_json::{
type_wrapper::{TypeConversionError, TypeKey, TypeWrapper},
wast_val::{ValKey, WastVal, WastValWithType},
wast_val_ser::params,
};
use wasmtime::component::{Type, Val};
pub use deployment_config::naming::{
FnMarker, FnName, FunctionFqn, FunctionFqnParseError, IfcFqnMarker, IfcFqnName,
IfcFqnParseError, NAMESPACE_OBELISK, Name, PackageExtension, PkgFqn, SUFFIX_FN_CANCELLABLE,
SUFFIX_PKG_EXT, SUFFIX_PKG_SCHEDULE, SUFFIX_PKG_STUB, StrVariant,
};
#[derive(
thiserror::Error, Clone, Debug, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
#[error("{kind}")]
pub struct FinishedExecutionFailure {
pub kind: ExecutionFailureKind,
#[serde(skip_serializing_if = "Option::is_none")]
pub reason: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub detail: Option<String>,
}
impl FinishedExecutionFailure {
#[must_use]
pub fn as_pending_state_finished_error(&self) -> PendingStateFinishedError {
PendingStateFinishedError::ExecutionFailure(self.kind)
}
}
#[derive(
Debug,
Clone,
Copy,
derive_more::Display,
PartialEq,
Eq,
Serialize,
Deserialize,
schemars::JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum ExecutionFailureKind {
TimedOut,
NondeterminismDetected,
OutOfFuel,
Cancelled,
Uncategorized,
}
#[derive(
Debug,
Clone,
Copy,
derive_more::Display,
PartialEq,
Eq,
Serialize,
Deserialize,
schemars::JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum TrapKind {
#[display("trap")]
Trap,
#[display("out of fuel")]
OutOfFuel,
#[display("host function error")]
HostFunctionError,
}
#[derive(
Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, schemars::JsonSchema,
)]
pub struct TypeWrapperTopLevel {
pub ok: Option<Box<TypeWrapper>>,
pub err: Option<Box<TypeWrapper>>,
}
impl TypeWrapperTopLevel {
#[must_use]
pub fn is_result_of_units(&self) -> bool {
self.ok.is_none() && self.err.is_none()
}
}
impl From<TypeWrapperTopLevel> for TypeWrapper {
fn from(value: TypeWrapperTopLevel) -> TypeWrapper {
TypeWrapper::Result {
ok: value.ok,
err: value.err,
}
}
}
#[derive(
Clone, derive_more::Debug, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum SupportedFunctionReturnValue {
Ok(Option<WastValWithType>),
Err(Option<WastValWithType>),
ExecutionFailure(FinishedExecutionFailure),
}
impl Display for SupportedFunctionReturnValue {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self.as_pending_state_finished_result() {
PendingStateFinishedResultKind::Ok => write!(f, "completed successfully"),
PendingStateFinishedResultKind::Err(err) => write!(f, "{err}"),
}
}
}
impl SupportedFunctionReturnValue {
#[must_use]
pub fn is_permanent_variant(&self) -> bool {
match self {
SupportedFunctionReturnValue::Err(Some(WastValWithType {
value: val_json::wast_val::WastVal::Variant(key, _),
..
})) => key.as_snake_str().contains("permanent"),
_ => false,
}
}
}
pub const SUPPORTED_RETURN_VALUE_OK_EMPTY: SupportedFunctionReturnValue =
SupportedFunctionReturnValue::Ok(None);
#[derive(Debug, thiserror::Error)]
pub enum ResultParsingError {
#[error("return value must not be empty")]
NoValue,
#[error("return value cannot be parsed, multi-value results are not supported")]
MultiValue,
#[error("return value cannot be parsed, {0}")]
TypeConversionError(val_json::type_wrapper::TypeConversionError),
#[error(transparent)]
ResultParsingErrorFromVal(ResultParsingErrorFromVal),
}
#[derive(Debug, thiserror::Error)]
pub enum ResultParsingErrorFromVal {
#[error("return value cannot be parsed, {0}")]
WastValConversionError(val_json::wast_val::WastValConversionError),
#[error("top level type must be a result")]
TopLevelTypeMustBeAResult,
#[error("value does not type check - {0}")]
TypeCheckError(String),
}
impl SupportedFunctionReturnValue {
pub fn new_from_iterator<
I: ExactSizeIterator<Item = (wasmtime::component::Val, wasmtime::component::Type)>,
>(
mut iter: I,
) -> Result<Self, ResultParsingError> {
if iter.len() == 0 {
Err(ResultParsingError::NoValue)
} else if iter.len() == 1 {
let (val, r#type) = iter.next().unwrap();
let r#type =
TypeWrapper::try_from(r#type).map_err(ResultParsingError::TypeConversionError)?;
Self::from_val_and_type_wrapper(val, r#type)
.map_err(ResultParsingError::ResultParsingErrorFromVal)
} else {
Err(ResultParsingError::MultiValue)
}
}
pub fn new(
val: wasmtime::component::Val,
r#type: wasmtime::component::Type,
) -> Result<Self, ResultParsingError> {
let r#type =
TypeWrapper::try_from(r#type).map_err(ResultParsingError::TypeConversionError)?;
Self::from_val_and_type_wrapper(val, r#type)
.map_err(ResultParsingError::ResultParsingErrorFromVal)
}
#[expect(clippy::result_unit_err)]
pub fn from_wast_val_with_type(
value: WastValWithType,
) -> Result<SupportedFunctionReturnValue, ()> {
match value {
WastValWithType {
r#type: TypeWrapper::Result { ok: None, err: _ },
value: WastVal::Result(Ok(None)),
} => Ok(SupportedFunctionReturnValue::Ok(None)),
WastValWithType {
r#type:
TypeWrapper::Result {
ok: Some(ok),
err: _,
},
value: WastVal::Result(Ok(Some(value))),
} => Ok(SupportedFunctionReturnValue::Ok(Some(WastValWithType {
r#type: *ok,
value: *value,
}))),
WastValWithType {
r#type: TypeWrapper::Result { ok: _, err: None },
value: WastVal::Result(Err(None)),
} => Ok(SupportedFunctionReturnValue::Err(None)),
WastValWithType {
r#type:
TypeWrapper::Result {
ok: _,
err: Some(err),
},
value: WastVal::Result(Err(Some(value))),
} => Ok(SupportedFunctionReturnValue::Err(Some(WastValWithType {
r#type: *err,
value: *value,
}))),
_ => Err(()),
}
}
pub fn from_val_and_type_wrapper(
value: wasmtime::component::Val,
ty: TypeWrapper,
) -> Result<Self, ResultParsingErrorFromVal> {
let TypeWrapper::Result { ok, err } = ty else {
return Err(ResultParsingErrorFromVal::TopLevelTypeMustBeAResult);
};
let ty = TypeWrapperTopLevel { ok, err };
Self::from_val_and_type_wrapper_tl(value, ty)
}
pub fn from_val_and_type_wrapper_tl(
value: wasmtime::component::Val,
ty: TypeWrapperTopLevel,
) -> Result<Self, ResultParsingErrorFromVal> {
let wasmtime::component::Val::Result(value) = value else {
return Err(ResultParsingErrorFromVal::TopLevelTypeMustBeAResult);
};
match (ty.ok, ty.err, value) {
(None, _, Ok(None)) => Ok(SupportedFunctionReturnValue::Ok(None)),
(Some(ok_type), _, Ok(Some(value))) => {
Ok(SupportedFunctionReturnValue::Ok(Some(WastValWithType {
r#type: *ok_type,
value: WastVal::try_from(*value)
.map_err(ResultParsingErrorFromVal::WastValConversionError)?,
})))
}
(_, None, Err(None)) => Ok(SupportedFunctionReturnValue::Err(None)),
(_, Some(err_type), Err(Some(value))) => {
Ok(SupportedFunctionReturnValue::Err(Some(WastValWithType {
r#type: *err_type,
value: WastVal::try_from(*value)
.map_err(ResultParsingErrorFromVal::WastValConversionError)?,
})))
}
(ok_type, err_type, value) => Err(ResultParsingErrorFromVal::TypeCheckError(format!(
"invalid combination - ok type: {ok_type:?}, err type: {err_type:?}, value: {value:?}"
))),
}
}
#[must_use]
pub fn into_wast_val(self, get_return_type: impl FnOnce() -> TypeWrapperTopLevel) -> WastVal {
WastVal::Result(self.into_wast_val_res(get_return_type))
}
pub fn into_wast_val_res(
self,
get_return_type: impl FnOnce() -> TypeWrapperTopLevel,
) -> Result<Option<Box<WastVal>>, Option<Box<WastVal>>> {
match self {
SupportedFunctionReturnValue::Ok(None) => Ok(None),
SupportedFunctionReturnValue::Ok(Some(v)) => Ok(Some(Box::new(v.value))),
SupportedFunctionReturnValue::Err(None) => Err(None),
SupportedFunctionReturnValue::Err(Some(v)) => Err(Some(Box::new(v.value))),
SupportedFunctionReturnValue::ExecutionFailure(_) => {
Err(Self::execution_error_to_wast_val_err(&get_return_type()))
}
}
}
fn execution_error_to_wast_val_err(ret_type: &TypeWrapperTopLevel) -> Option<Box<WastVal>> {
match ret_type {
TypeWrapperTopLevel { ok: _, err: None } => None,
TypeWrapperTopLevel {
ok: _,
err: Some(inner),
} => match inner.as_ref() {
TypeWrapper::String => Some(Box::new(WastVal::String(
EXECUTION_FAILED_JSON_STRING.to_string(),
))),
TypeWrapper::Variant(variants)
if variants.get(&TypeKey::new_kebab(EXECUTION_FAILED_WIT_CASE))
== Some(&None) =>
{
Some(Box::new(WastVal::Variant(
ValKey::from_kebab(EXECUTION_FAILED_WIT_CASE),
None,
)))
}
_ => {
unreachable!(
"unexpected top-level return type {ret_type:?} cannot be `ReturnTypeExtendable`"
)
}
},
}
}
#[must_use]
pub fn as_pending_state_finished_result(&self) -> PendingStateFinishedResultKind {
match self {
SupportedFunctionReturnValue::Ok(_) => PendingStateFinishedResultKind::Ok,
SupportedFunctionReturnValue::Err(_) => {
PendingStateFinishedResultKind::Err(PendingStateFinishedError::Error)
}
SupportedFunctionReturnValue::ExecutionFailure(err) => {
PendingStateFinishedResultKind::Err(err.as_pending_state_finished_error())
}
}
}
}
#[derive(Debug, Clone, schemars::JsonSchema)]
#[schemars(with = "Vec<serde_json::Value>")]
pub struct Params(ParamsInternal);
#[derive(derive_more::Debug, Clone)]
enum ParamsInternal {
JsonValues(Arc<[Value]>),
Vals {
vals: Arc<[wasmtime::component::Val]>,
#[debug(skip)]
json_vals_cache: Arc<std::sync::RwLock<Option<Arc<[Value]>>>>, },
Empty,
}
pub const SUFFIX_FN_SUBMIT: &str = "-submit";
pub const SUFFIX_FN_AWAIT_NEXT: &str = "-await-next";
pub const SUFFIX_FN_SCHEDULE: &str = "-schedule";
pub const SUFFIX_FN_STUB: &str = "-stub";
pub const SUFFIX_FN_GET: &str = "-get";
#[derive(
Debug,
Clone,
Copy,
serde::Serialize,
serde::Deserialize,
PartialEq,
Eq,
strum::EnumIter,
schemars::JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum FunctionExtension {
Submit,
AwaitNext,
Schedule,
Stub,
Get,
}
impl FunctionExtension {
#[must_use]
pub fn suffix(&self) -> &'static str {
match self {
FunctionExtension::Submit => SUFFIX_FN_SUBMIT,
FunctionExtension::AwaitNext => SUFFIX_FN_AWAIT_NEXT,
FunctionExtension::Schedule => SUFFIX_FN_SCHEDULE,
FunctionExtension::Stub => SUFFIX_FN_STUB,
FunctionExtension::Get => SUFFIX_FN_GET,
}
}
#[must_use]
pub fn belongs_to(&self, pkg_ext: PackageExtension) -> bool {
matches!(
(pkg_ext, self),
(
PackageExtension::ObeliskExt,
FunctionExtension::Submit | FunctionExtension::AwaitNext | FunctionExtension::Get
) | (
PackageExtension::ObeliskSchedule,
FunctionExtension::Schedule
) | (PackageExtension::ObeliskStub, FunctionExtension::Stub)
)
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct FunctionMetadata {
pub ffqn: FunctionFqn,
pub parameter_types: ParameterTypes,
pub return_type: ReturnType,
pub extension: Option<FunctionExtension>,
pub submittable: bool,
}
impl FunctionMetadata {
#[must_use]
pub fn split_extension(&self) -> Option<(&str, FunctionExtension)> {
self.extension.map(|extension| {
let prefix = self
.ffqn
.function_name
.value
.strip_suffix(extension.suffix())
.unwrap_or_else(|| {
panic!(
"extension function {} must end with expected suffix {}",
self.ffqn.function_name,
extension.suffix()
)
});
(prefix, extension)
})
}
}
impl Display for FunctionMetadata {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"{ffqn}: func{params} -> {return_type}",
ffqn = self.ffqn,
params = self.parameter_types,
return_type = self.return_type,
)
}
}
pub mod serde_params {
use crate::{Params, ParamsInternal};
use serde::de::{SeqAccess, Visitor};
use serde::ser::SerializeSeq;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::sync::Arc;
use val_json::wast_val::WastVal;
impl Serialize for Params {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: ::serde::Serializer,
{
let serialize_json_values = |slice: &[serde_json::Value], serializer: S| {
let mut seq = serializer.serialize_seq(Some(slice.len()))?;
for item in slice {
seq.serialize_element(item)?;
}
seq.end()
};
match &self.0 {
ParamsInternal::Vals {
vals,
json_vals_cache,
} => {
let guard = json_vals_cache.read().unwrap();
if let Some(slice) = &*guard {
serialize_json_values(slice, serializer)
} else {
drop(guard);
let mut json_vals = Vec::with_capacity(vals.len());
for val in vals.iter() {
let value = WastVal::try_from(val.clone())
.map_err(|err| serde::ser::Error::custom(err.to_string()))?;
let value = serde_json::to_value(&value)
.map_err(|err| serde::ser::Error::custom(err.to_string()))?;
json_vals.push(value);
}
let res = serialize_json_values(&json_vals, serializer);
*json_vals_cache.write().unwrap() = Some(Arc::from(json_vals));
res
}
}
ParamsInternal::Empty => serializer.serialize_seq(Some(0))?.end(),
ParamsInternal::JsonValues(vec) => serialize_json_values(vec, serializer),
}
}
}
pub struct VecVisitor;
impl<'de> Visitor<'de> for VecVisitor {
type Value = Vec<Value>;
fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
formatter.write_str("a sequence of `Value`")
}
#[inline]
fn visit_seq<V>(self, mut visitor: V) -> Result<Self::Value, V::Error>
where
V: SeqAccess<'de>,
{
let mut vec = Vec::new();
while let Some(elem) = visitor.next_element()? {
vec.push(elem);
}
Ok(vec)
}
}
impl<'de> Deserialize<'de> for Params {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let vec: Vec<Value> = deserializer.deserialize_seq(VecVisitor)?;
if vec.is_empty() {
Ok(Self(ParamsInternal::Empty))
} else {
Ok(Self(ParamsInternal::JsonValues(Arc::from(vec))))
}
}
}
}
#[derive(Debug, thiserror::Error)]
pub enum ParamsParsingError {
#[error("parameters cannot be parsed, cannot convert type of {idx}-th parameter")]
ParameterTypeError {
idx: usize,
err: TypeConversionError,
},
#[error("parameters cannot be deserialized: {0}")]
ParamsDeserializationError(#[source] serde_json::Error),
#[error("parameter cardinality mismatch, expected: {expected}, specified: {specified}")]
ParameterCardinalityMismatch { expected: usize, specified: usize },
}
impl ParamsParsingError {
#[must_use]
pub fn detail(&self) -> Option<String> {
match self {
ParamsParsingError::ParameterTypeError { err, .. } => Some(format!("{err:?}")),
ParamsParsingError::ParamsDeserializationError(err) => Some(format!("{err:?}")),
ParamsParsingError::ParameterCardinalityMismatch { .. } => None,
}
}
}
impl Params {
#[must_use]
pub const fn empty() -> Self {
Self(ParamsInternal::Empty)
}
#[must_use]
pub fn from_wasmtime(vals: Arc<[wasmtime::component::Val]>) -> Self {
if vals.is_empty() {
Self::empty()
} else {
Self(ParamsInternal::Vals {
vals,
json_vals_cache: Arc::default(),
})
}
}
#[cfg(any(test, feature = "test"))]
#[must_use]
pub fn from_json_values_test(vec: Vec<Value>) -> Self {
if vec.is_empty() {
Self::empty()
} else {
Self(ParamsInternal::JsonValues(Arc::from(vec)))
}
}
pub fn from_json_values<'a>(
values: Arc<[Value]>,
param_types: impl ExactSizeIterator<Item = &'a TypeWrapper>,
) -> Result<Self, ParamsParsingError> {
if param_types.len() != values.len() {
return Err(ParamsParsingError::ParameterCardinalityMismatch {
expected: param_types.len(),
specified: values.len(),
});
}
if values.is_empty() {
Ok(Self::empty())
} else {
params::deserialize_values(&values, param_types)
.map_err(ParamsParsingError::ParamsDeserializationError)?;
Ok(Self(ParamsInternal::JsonValues(values)))
}
}
pub fn as_vals<'a>(
&self,
param_types: impl ExactSizeIterator<Item = (&'a str, Type)>,
) -> Result<Arc<[wasmtime::component::Val]>, ParamsParsingError> {
if param_types.len() != self.len() {
return Err(ParamsParsingError::ParameterCardinalityMismatch {
expected: param_types.len(),
specified: self.len(),
});
}
match &self.0 {
ParamsInternal::JsonValues(json_vec) => {
let param_types = param_types
.enumerate()
.map(|(idx, (_param_name, ty))| {
TypeWrapper::try_from(ty).map_err(|err| (idx, err))
})
.collect::<Result<Vec<_>, _>>()
.map_err(|(idx, err)| ParamsParsingError::ParameterTypeError { idx, err })?;
Ok(params::deserialize_values(json_vec, param_types.iter())
.map_err(ParamsParsingError::ParamsDeserializationError)?
.into_iter()
.map(Val::from)
.collect())
}
ParamsInternal::Vals { vals, .. } => Ok(vals.clone()),
ParamsInternal::Empty => Ok(Arc::from([])),
}
}
#[must_use]
pub fn len(&self) -> usize {
match &self.0 {
ParamsInternal::JsonValues(vec) => vec.len(),
ParamsInternal::Vals { vals, .. } => vals.len(),
ParamsInternal::Empty => 0,
}
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.len() == 0
}
#[must_use]
pub fn as_json_values(&self) -> Option<&[serde_json::Value]> {
match &self.0 {
ParamsInternal::Vals { .. } => None,
ParamsInternal::Empty => Some(&[]),
ParamsInternal::JsonValues(vec) => Some(vec),
}
}
}
impl PartialEq for Params {
fn eq(&self, other: &Self) -> bool {
if self.is_empty() && other.is_empty() {
return true;
}
if self.len() != other.len() {
return false;
}
match (&self.0, &other.0) {
(ParamsInternal::JsonValues(l), ParamsInternal::JsonValues(r)) => l == r,
(
ParamsInternal::Vals {
vals: l,
json_vals_cache: _,
},
ParamsInternal::Vals {
vals: r,
json_vals_cache: _,
},
) => l == r,
(
ParamsInternal::JsonValues(json_vals),
ParamsInternal::Vals {
vals,
json_vals_cache,
},
)
| (
ParamsInternal::Vals {
vals,
json_vals_cache,
},
ParamsInternal::JsonValues(json_vals),
) => {
let Ok(vec) = to_json(vals) else { return false };
let equals = *json_vals == vec;
*json_vals_cache.write().unwrap() = Some(vec);
equals
}
(ParamsInternal::Empty, _) | (_, ParamsInternal::Empty) => {
unreachable!("zero length and different lengths handled earlier")
}
}
}
}
impl Eq for Params {}
fn to_json(vals: &[wasmtime::component::Val]) -> Result<Arc<[serde_json::Value]>, ()> {
let mut vec = Vec::with_capacity(vals.len());
for val in vals {
let value = match WastVal::try_from(val.clone()) {
Ok(ok) => ok,
Err(err) => {
error!("cannot compare Params, cannot convert to WastVal: {err:?}");
return Err(());
}
};
let value = match serde_json::to_value(&value) {
Ok(ok) => ok,
Err(err) => {
error!("cannot compare Params, cannot convert to JSON: {err:?}");
return Err(());
}
};
vec.push(value);
}
Ok(Arc::from(vec))
}
impl Display for Params {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let render_json_vals =
|f: &mut std::fmt::Formatter<'_>, json_vals: &[Value]| -> std::fmt::Result {
for (i, json_value) in json_vals.iter().enumerate() {
if i > 0 {
write!(f, ", ")?;
}
write!(f, "{json_value}")?;
}
Ok(())
};
write!(f, "[")?;
match &self.0 {
ParamsInternal::Empty => {}
ParamsInternal::JsonValues(json_vals) => {
render_json_vals(f, json_vals)?;
}
ParamsInternal::Vals {
vals,
json_vals_cache,
} => {
let guard = json_vals_cache.read().unwrap();
if let Some(json_vals) = &*guard {
render_json_vals(f, json_vals)?;
} else {
drop(guard);
let Ok(json_vals) = to_json(vals) else {
return write!(f, "<serialization error>]"); };
render_json_vals(f, &json_vals)?;
*json_vals_cache.write().unwrap() = Some(json_vals);
}
}
}
write!(f, "]")
}
}
pub mod prefixed_ulid {
use crate::{
EXECUTION_ID_INFIX, EXECUTION_ID_INFIX_BETWEEN_JOIN_SET_AND_INDEX, JoinSetId,
JoinSetIdParseError,
};
use serde_with::{DeserializeFromStr, SerializeDisplay};
use sha2::{Digest, Sha256};
use std::{
fmt::{Debug, Display},
marker::PhantomData,
num::ParseIntError,
str::FromStr,
sync::Arc,
};
use ulid::Ulid;
#[derive(derive_more::Display, SerializeDisplay, DeserializeFromStr, schemars::JsonSchema)]
#[schemars(with = "String")]
#[derive_where::derive_where(Clone, Copy)]
#[display("{}_{ulid}", Self::prefix())]
pub struct PrefixedUlid<T: 'static> {
ulid: Ulid,
phantom_data: PhantomData<fn(T) -> T>,
}
impl<T> PrefixedUlid<T> {
const fn new(ulid: Ulid) -> Self {
Self {
ulid,
phantom_data: PhantomData,
}
}
fn prefix() -> &'static str {
std::any::type_name::<T>().rsplit("::").next().unwrap()
}
}
impl<T> PrefixedUlid<T> {
#[must_use]
pub fn generate() -> Self {
Self::new(Ulid::new())
}
#[must_use]
pub const fn from_parts(timestamp_ms: u64, random: u128) -> Self {
Self::new(Ulid::from_parts(timestamp_ms, random))
}
#[must_use]
pub fn timestamp_part(&self) -> u64 {
self.ulid.timestamp_ms()
}
#[must_use]
pub fn random_part(&self) -> u128 {
self.ulid.random()
}
#[must_use]
pub fn ulid(&self) -> Ulid {
self.ulid
}
}
#[derive(Debug, thiserror::Error)]
pub enum PrefixedUlidParseError {
#[error("wrong prefix in `{input}`, expected prefix `{expected}`")]
WrongPrefix { input: String, expected: String },
#[error("cannot parse ULID suffix from `{input}`")]
CannotParseUlid { input: String },
}
mod impls {
use super::{PrefixedUlid, PrefixedUlidParseError, Ulid};
use std::{fmt::Debug, fmt::Display, hash::Hash, marker::PhantomData, str::FromStr};
impl<T> FromStr for PrefixedUlid<T> {
type Err = PrefixedUlidParseError;
fn from_str(input: &str) -> Result<Self, Self::Err> {
let prefix = Self::prefix();
let mut input_chars = input.chars();
for exp in prefix.chars() {
if input_chars.next() != Some(exp) {
return Err(PrefixedUlidParseError::WrongPrefix {
input: input.to_string(),
expected: format!("{prefix}_"),
});
}
}
if input_chars.next() != Some('_') {
return Err(PrefixedUlidParseError::WrongPrefix {
input: input.to_string(),
expected: format!("{prefix}_"),
});
}
let Ok(ulid) = Ulid::from_string(input_chars.as_str()) else {
return Err(PrefixedUlidParseError::CannotParseUlid {
input: input.to_string(),
});
};
Ok(Self {
ulid,
phantom_data: PhantomData,
})
}
}
impl<T> Debug for PrefixedUlid<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
Display::fmt(&self, f)
}
}
impl<T> Hash for PrefixedUlid<T> {
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
Self::prefix().hash(state);
self.ulid.hash(state);
self.phantom_data.hash(state);
}
}
impl<T> PartialEq for PrefixedUlid<T> {
fn eq(&self, other: &Self) -> bool {
self.ulid == other.ulid
}
}
impl<T> Eq for PrefixedUlid<T> {}
impl<T> PartialOrd for PrefixedUlid<T> {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl<T> Ord for PrefixedUlid<T> {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
self.ulid.cmp(&other.ulid)
}
}
}
pub mod prefix {
pub struct E;
pub struct Exr;
pub struct Run;
pub struct Delay;
pub struct Dep;
}
pub type ExecutorId = PrefixedUlid<prefix::Exr>;
pub type ExecutionIdTopLevel = PrefixedUlid<prefix::E>;
pub type RunId = PrefixedUlid<prefix::Run>;
pub type DelayIdTopLevel = PrefixedUlid<prefix::Delay>; pub type DeploymentId = PrefixedUlid<prefix::Dep>;
#[cfg(any(test, feature = "test"))]
pub const DEPLOYMENT_ID_DUMMY: DeploymentId = DeploymentId::from_parts(0, 0);
#[cfg(any(test, feature = "test"))]
pub const EXECUTION_ID_DUMMY: ExecutionId = ExecutionId::from_parts(0, 0);
#[cfg(any(test, feature = "test"))]
impl<'a, T> arbitrary::Arbitrary<'a> for PrefixedUlid<T> {
fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
Ok(Self::new(ulid::Ulid::from_parts(
u.arbitrary()?,
u.arbitrary()?,
)))
}
}
#[derive(
Hash,
PartialEq,
Eq,
PartialOrd,
Ord,
SerializeDisplay,
DeserializeFromStr,
Clone,
schemars::JsonSchema,
)]
#[schemars(with = "String")]
pub enum ExecutionId {
TopLevel(ExecutionIdTopLevel),
Derived(ExecutionIdDerived),
}
#[derive(
Hash,
PartialEq,
Eq,
PartialOrd,
Ord,
Clone,
SerializeDisplay,
DeserializeFromStr,
schemars::JsonSchema,
)]
#[schemars(with = "String")]
pub struct ExecutionIdDerived {
top_level: ExecutionIdTopLevel,
infix: Arc<str>,
idx: u64,
}
impl ExecutionIdDerived {
#[must_use]
pub fn get_incremented(&self) -> Self {
self.get_incremented_by(1)
}
#[must_use]
pub fn get_incremented_by(&self, count: u64) -> Self {
ExecutionIdDerived {
top_level: self.top_level,
infix: self.infix.clone(),
idx: self.idx + count,
}
}
#[must_use]
pub fn next_level(&self, join_set_id: &JoinSetId) -> ExecutionIdDerived {
let ExecutionIdDerived {
top_level,
infix,
idx,
} = self;
let infix = Arc::from(format!(
"{infix}{EXECUTION_ID_INFIX_BETWEEN_JOIN_SET_AND_INDEX}{idx}{EXECUTION_ID_INFIX}{join_set_id}"
));
ExecutionIdDerived {
top_level: *top_level,
infix,
idx: EXECUTION_ID_START_IDX,
}
}
fn display_or_debug(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let ExecutionIdDerived {
top_level,
infix,
idx,
} = self;
write!(
f,
"{top_level}{EXECUTION_ID_INFIX}{infix}{EXECUTION_ID_INFIX_BETWEEN_JOIN_SET_AND_INDEX}{idx}"
)
}
#[must_use]
pub fn split_to_parts(&self) -> (ExecutionId, JoinSetId) {
self.split_to_parts_res().expect("verified in from_str")
}
fn split_to_parts_res(&self) -> Result<(ExecutionId, JoinSetId), DerivedIdSplitError> {
if let Some((old_infix_and_index, join_set_id)) =
self.infix.rsplit_once(EXECUTION_ID_INFIX)
{
let join_set_id = JoinSetId::from_str(join_set_id)
.map_err(DerivedIdSplitError::JoinSetIdParseError)?;
let Some((old_infix, old_idx)) =
old_infix_and_index.rsplit_once(EXECUTION_ID_INFIX_BETWEEN_JOIN_SET_AND_INDEX)
else {
return Err(DerivedIdSplitError::CannotFindJoinSetDelimiter);
};
let parent = ExecutionIdDerived {
top_level: self.top_level,
infix: Arc::from(old_infix),
idx: old_idx
.parse()
.map_err(DerivedIdSplitError::CannotParseOldIndex)?,
};
Ok((ExecutionId::Derived(parent), join_set_id))
} else {
Ok((
ExecutionId::TopLevel(self.top_level),
JoinSetId::from_str(&self.infix)
.map_err(DerivedIdSplitError::JoinSetIdParseError)?,
))
}
}
}
#[derive(Debug, thiserror::Error)]
pub enum DerivedIdSplitError {
#[error(transparent)]
JoinSetIdParseError(JoinSetIdParseError),
#[error("cannot parse index of parent execution - {0}")]
CannotParseOldIndex(ParseIntError),
#[error("cannot find join set delimiter")]
CannotFindJoinSetDelimiter,
}
impl Debug for ExecutionIdDerived {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.display_or_debug(f)
}
}
impl Display for ExecutionIdDerived {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.display_or_debug(f)
}
}
impl FromStr for ExecutionIdDerived {
type Err = DerivedIdParseError;
fn from_str(input: &str) -> Result<Self, Self::Err> {
let (top_level, infix, idx) = derived_from_str(input)?;
let id = ExecutionIdDerived {
top_level,
infix,
idx,
};
Ok(id)
}
}
fn derived_from_str<T: 'static>(
input: &str,
) -> Result<(PrefixedUlid<T>, Arc<str>, u64), DerivedIdParseError> {
let (prefix, full_suffix) = input
.split_once(EXECUTION_ID_INFIX)
.ok_or(DerivedIdParseError::FirstDelimiterNotFound)?;
let top_level =
PrefixedUlid::from_str(prefix).map_err(DerivedIdParseError::PrefixedUlidParseError)?;
let mut last_idx = None;
for segment in full_suffix.split(EXECUTION_ID_INFIX) {
if segment.is_empty() {
return Err(DerivedIdParseError::EmptySegment);
}
let (join_set_str, idx_str) = segment
.split_once(EXECUTION_ID_INFIX_BETWEEN_JOIN_SET_AND_INDEX)
.ok_or(DerivedIdParseError::SecondDelimiterNotFound)?;
JoinSetId::from_str(join_set_str).map_err(DerivedIdParseError::JoinSetIdParseError)?;
let idx = u64::from_str(idx_str).map_err(DerivedIdParseError::ParseIndexError)?;
last_idx = Some((idx, idx_str));
}
let (idx, idx_str) = last_idx.expect("split must have returned at least one element");
let infix = full_suffix
.strip_suffix(idx_str)
.expect("must have ended with index");
let infix = infix
.strip_suffix(EXECUTION_ID_INFIX_BETWEEN_JOIN_SET_AND_INDEX)
.expect("must have ended with `_index`");
Ok((top_level, Arc::from(infix), idx))
}
#[cfg(any(test, feature = "test"))]
impl<'a> arbitrary::Arbitrary<'a> for ExecutionIdDerived {
fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
let top_level = ExecutionId::TopLevel(ExecutionIdTopLevel::arbitrary(u)?);
let join_set_id = JoinSetId::arbitrary(u)?;
Ok(top_level.next_level(&join_set_id))
}
}
#[derive(Debug, thiserror::Error)]
pub enum DerivedIdParseError {
#[error(transparent)]
PrefixedUlidParseError(PrefixedUlidParseError),
#[error("cannot parse derived id - delimiter `{EXECUTION_ID_INFIX}` not found")]
FirstDelimiterNotFound,
#[error(
"cannot parse derived id - delimiter `{EXECUTION_ID_INFIX_BETWEEN_JOIN_SET_AND_INDEX}` not found"
)]
SecondDelimiterNotFound,
#[error(
"cannot parse derived id - suffix after `{EXECUTION_ID_INFIX_BETWEEN_JOIN_SET_AND_INDEX}` must be a number"
)]
ParseIndexError(ParseIntError),
#[error(transparent)]
DerivedIdSplitError(DerivedIdSplitError),
#[error(transparent)]
JoinSetIdParseError(JoinSetIdParseError),
#[error("empty segment")]
EmptySegment,
}
impl ExecutionId {
#[must_use]
pub fn generate() -> Self {
ExecutionId::TopLevel(PrefixedUlid::generate())
}
#[must_use]
pub fn deterministic_at_unix_epoch(hash: &[u8; 32]) -> Self {
let mut ts_bytes = [0u8; 8];
ts_bytes[4..8].copy_from_slice(&hash[..4]); let timestamp_ms = u64::from_be_bytes(ts_bytes);
const RAND_LEN: usize = 10;
const U128_BYTES_LEN: usize = std::mem::size_of::<u128>(); const START: usize = U128_BYTES_LEN - RAND_LEN;
let mut rand_bytes = [0u8; U128_BYTES_LEN];
rand_bytes[START..].copy_from_slice(&hash[4..4 + RAND_LEN]);
let random = u128::from_be_bytes(rand_bytes);
ExecutionId::TopLevel(PrefixedUlid::new(Ulid::from_parts(timestamp_ms, random)))
}
#[must_use]
pub fn get_top_level(&self) -> ExecutionIdTopLevel {
match &self {
ExecutionId::TopLevel(prefixed_ulid) => *prefixed_ulid,
ExecutionId::Derived(ExecutionIdDerived { top_level, .. }) => *top_level,
}
}
#[must_use]
pub fn is_top_level(&self) -> bool {
matches!(self, ExecutionId::TopLevel(_))
}
#[must_use]
pub fn random_seed(&self) -> u64 {
let mut hasher = Sha256::new();
hasher.update(self.get_top_level().ulid.0.to_le_bytes());
if let ExecutionId::Derived(ExecutionIdDerived {
top_level: _,
infix,
idx,
}) = self
{
hasher.update(infix.as_bytes());
hasher.update(idx.to_le_bytes());
}
let hash = hasher.finalize();
u64::from_le_bytes(hash[..8].try_into().unwrap())
}
#[must_use]
pub const fn from_parts(timestamp_ms: u64, random_part: u128) -> Self {
ExecutionId::TopLevel(ExecutionIdTopLevel::from_parts(timestamp_ms, random_part))
}
#[must_use]
pub fn next_level(&self, join_set_id: &JoinSetId) -> ExecutionIdDerived {
match &self {
ExecutionId::TopLevel(top_level) => ExecutionIdDerived {
top_level: *top_level,
infix: Arc::from(join_set_id.to_string()),
idx: EXECUTION_ID_START_IDX,
},
ExecutionId::Derived(derived) => derived.next_level(join_set_id),
}
}
fn display_or_debug(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match &self {
ExecutionId::TopLevel(top_level) => Display::fmt(top_level, f),
ExecutionId::Derived(derived) => Display::fmt(derived, f),
}
}
}
const EXECUTION_ID_START_IDX: u64 = 1;
pub const JOIN_SET_START_IDX: u64 = 1;
const DELAY_ID_START_IDX: u64 = 1;
#[derive(Debug, thiserror::Error)]
pub enum ExecutionIdParseError {
#[error(transparent)]
PrefixedUlidParseError(PrefixedUlidParseError),
#[error(transparent)]
DerivedIdParseError(DerivedIdParseError),
}
impl FromStr for ExecutionId {
type Err = ExecutionIdParseError;
fn from_str(input: &str) -> Result<Self, Self::Err> {
if input.contains(EXECUTION_ID_INFIX) {
ExecutionIdDerived::from_str(input)
.map(ExecutionId::Derived)
.map_err(ExecutionIdParseError::DerivedIdParseError)
} else {
Ok(ExecutionId::TopLevel(
PrefixedUlid::from_str(input)
.map_err(ExecutionIdParseError::PrefixedUlidParseError)?,
))
}
}
}
#[derive(Debug, thiserror::Error)]
pub enum ExecutionIdStructuralParseError {
#[error(transparent)]
ExecutionIdParseError(#[from] ExecutionIdParseError),
#[error("execution-id must be a record with `id` field of type string")]
TypeError,
}
impl TryFrom<&wasmtime::component::Val> for ExecutionId {
type Error = ExecutionIdStructuralParseError;
fn try_from(execution_id: &wasmtime::component::Val) -> Result<Self, Self::Error> {
if let wasmtime::component::Val::Record(key_vals) = execution_id
&& key_vals.len() == 1
&& let Some((key, execution_id)) = key_vals.first()
&& key == "id"
&& let wasmtime::component::Val::String(execution_id) = execution_id
{
ExecutionId::from_str(execution_id)
.map_err(ExecutionIdStructuralParseError::ExecutionIdParseError)
} else {
Err(ExecutionIdStructuralParseError::TypeError)
}
}
}
impl Debug for ExecutionId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.display_or_debug(f)
}
}
impl Display for ExecutionId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.display_or_debug(f)
}
}
#[cfg(any(test, feature = "test"))]
impl<'a> arbitrary::Arbitrary<'a> for ExecutionId {
fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
Ok(ExecutionId::TopLevel(PrefixedUlid::arbitrary(u)?))
}
}
#[derive(
Hash,
PartialEq,
Eq,
PartialOrd,
Ord,
Clone,
SerializeDisplay,
DeserializeFromStr,
schemars::JsonSchema,
)]
#[schemars(with = "String")]
pub struct DelayId {
top_level: DelayIdTopLevel,
infix: Arc<str>,
idx: u64,
}
impl DelayId {
#[must_use]
pub fn new(execution_id: &ExecutionId, join_set_id: &JoinSetId) -> DelayId {
Self::from_index(execution_id, join_set_id, DELAY_ID_START_IDX)
}
#[cfg(any(test, feature = "test"))]
#[must_use]
pub fn new_with_index(
execution_id: &ExecutionId,
join_set_id: &JoinSetId,
idx: u64,
) -> DelayId {
Self::from_index(execution_id, join_set_id, idx)
}
fn from_index(execution_id: &ExecutionId, join_set_id: &JoinSetId, idx: u64) -> DelayId {
let ExecutionIdDerived {
top_level: PrefixedUlid { ulid, .. },
infix,
idx: _,
} = execution_id.next_level(join_set_id);
let top_level = DelayIdTopLevel::new(ulid);
DelayId {
top_level,
infix,
idx,
}
}
#[must_use]
pub fn index(&self) -> u64 {
self.idx
}
#[must_use]
pub fn get_incremented(&self) -> Self {
self.get_incremented_by(1)
}
#[must_use]
pub fn get_incremented_by(&self, count: u64) -> Self {
Self {
top_level: self.top_level,
infix: self.infix.clone(),
idx: self.idx + count,
}
}
#[must_use]
pub fn split_to_parts(&self) -> (ExecutionId, JoinSetId) {
self.split_to_parts_res().expect("verified in from_str")
}
fn split_to_parts_res(&self) -> Result<(ExecutionId, JoinSetId), DerivedIdSplitError> {
if let Some((old_infix_and_index, join_set_id)) =
self.infix.rsplit_once(EXECUTION_ID_INFIX)
{
let join_set_id = JoinSetId::from_str(join_set_id)
.map_err(DerivedIdSplitError::JoinSetIdParseError)?;
let Some((old_infix, old_idx)) =
old_infix_and_index.rsplit_once(EXECUTION_ID_INFIX_BETWEEN_JOIN_SET_AND_INDEX)
else {
return Err(DerivedIdSplitError::CannotFindJoinSetDelimiter);
};
let parent = ExecutionIdDerived {
top_level: ExecutionIdTopLevel::new(self.top_level.ulid),
infix: Arc::from(old_infix),
idx: old_idx
.parse()
.map_err(DerivedIdSplitError::CannotParseOldIndex)?,
};
Ok((ExecutionId::Derived(parent), join_set_id))
} else {
Ok((
ExecutionId::TopLevel(ExecutionIdTopLevel::new(self.top_level.ulid)),
JoinSetId::from_str(&self.infix)
.map_err(DerivedIdSplitError::JoinSetIdParseError)?,
))
}
}
fn display_or_debug(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let DelayId {
top_level,
infix,
idx,
} = self;
write!(
f,
"{top_level}{EXECUTION_ID_INFIX}{infix}{EXECUTION_ID_INFIX_BETWEEN_JOIN_SET_AND_INDEX}{idx}"
)
}
}
pub mod delay_impl {
use super::{DelayId, DerivedIdParseError, derived_from_str};
use std::{
fmt::{Debug, Display},
str::FromStr,
};
impl Debug for DelayId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.display_or_debug(f)
}
}
impl Display for DelayId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.display_or_debug(f)
}
}
impl FromStr for DelayId {
type Err = DerivedIdParseError;
fn from_str(input: &str) -> Result<Self, Self::Err> {
let (top_level, infix, idx) = derived_from_str(input)?;
let id = DelayId {
top_level,
infix,
idx,
};
Ok(id)
}
}
#[cfg(any(test, feature = "test"))]
impl<'a> arbitrary::Arbitrary<'a> for DelayId {
fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
use super::{ExecutionId, JoinSetId};
let execution_id = ExecutionId::arbitrary(u)?;
let mut join_set_id = JoinSetId::arbitrary(u)?;
join_set_id.kind = crate::JoinSetKind::OneOff;
Ok(DelayId::new(&execution_id, &join_set_id))
}
}
}
}
#[derive(
Debug,
Clone,
PartialEq,
Eq,
Hash,
derive_more::Display,
serde_with::SerializeDisplay,
serde_with::DeserializeFromStr,
schemars::JsonSchema,
)]
#[schemars(with = "String")]
#[non_exhaustive] #[display("{kind}{JOIN_SET_ID_INFIX}{name}")]
pub struct JoinSetId {
pub kind: JoinSetKind,
pub name: StrVariant,
}
impl JoinSetId {
pub fn new(kind: JoinSetKind, name: StrVariant) -> Result<Self, InvalidNameError<JoinSetId>> {
Ok(Self {
kind,
name: check_name(name, CHARSET_EXTRA_JSON_SET)?,
})
}
}
pub const CHARSET_ALPHANUMERIC: &str =
"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
#[derive(
Debug,
Clone,
Copy,
PartialEq,
Eq,
Hash,
derive_more::Display,
strum::EnumIter,
schemars::JsonSchema,
)]
#[cfg_attr(any(test, feature = "test"), derive(arbitrary::Arbitrary))]
#[display("{}", self.as_code())]
pub enum JoinSetKind {
OneOff,
Named,
Generated,
}
impl JoinSetKind {
fn as_code(&self) -> &'static str {
match self {
JoinSetKind::OneOff => "o",
JoinSetKind::Named => "n",
JoinSetKind::Generated => "g",
}
}
}
impl FromStr for JoinSetKind {
type Err = &'static str;
fn from_str(s: &str) -> Result<Self, Self::Err> {
use strum::IntoEnumIterator;
Self::iter()
.find(|variant| s == variant.as_code())
.ok_or("unknown join set kind")
}
}
pub(crate) const EXECUTION_ID_INFIX: char = '.';
pub(crate) const EXECUTION_ID_INFIX_BETWEEN_JOIN_SET_AND_INDEX: char = '_';
const JOIN_SET_ID_INFIX: char = ':';
const CHARSET_EXTRA_JSON_SET: &str = "-/";
impl FromStr for JoinSetId {
type Err = JoinSetIdParseError;
fn from_str(input: &str) -> Result<Self, Self::Err> {
let Some((kind, name)) = input.split_once(JOIN_SET_ID_INFIX) else {
return Err(JoinSetIdParseError::WrongParts);
};
let kind = kind
.parse()
.map_err(JoinSetIdParseError::JoinSetKindParseError)?;
JoinSetId::new(kind, StrVariant::from(name.to_string()))
.map_err(JoinSetIdParseError::InvalidName)
}
}
#[derive(Debug, thiserror::Error)]
pub enum JoinSetIdParseError {
#[error("join set must consist of three parts separated by {JOIN_SET_ID_INFIX} ")]
WrongParts,
#[error("cannot parse join set kind - {0}")]
JoinSetKindParseError(&'static str),
#[error("cannot parse join set id - {0}")]
InvalidName(InvalidNameError<JoinSetId>),
}
#[cfg(any(test, feature = "test"))]
const CHARSET_JOIN_SET_NAME: &str =
const_format::concatcp!(CHARSET_ALPHANUMERIC, CHARSET_EXTRA_JSON_SET);
#[cfg(any(test, feature = "test"))]
impl<'a> arbitrary::Arbitrary<'a> for JoinSetId {
fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
let name: String = {
let length_inclusive = u.int_in_range(0..=10).unwrap();
(0..=length_inclusive)
.map(|_| {
let idx = u.choose_index(CHARSET_JOIN_SET_NAME.len()).unwrap();
CHARSET_JOIN_SET_NAME
.chars()
.nth(idx)
.expect("idx is < charset.len()")
})
.collect()
};
Ok(JoinSetId::new(JoinSetKind::Named, StrVariant::from(name)).unwrap())
}
}
const EXECUTION_FAILED_WIT_CASE: &str = "execution-failed";
const EXECUTION_FAILED_JSON_STRING: &str = "execution_failed";
#[derive(
Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, derive_more::Display,
)]
#[serde(rename_all = "snake_case")]
pub enum ReturnType {
Extendable(ReturnTypeExtendable), NonExtendable(ReturnTypeNonExtendable), }
impl ReturnType {
#[must_use]
pub fn detect(type_wrapper: TypeWrapper, wit_type: StrVariant) -> ReturnType {
if let TypeWrapper::Result { ok, err: None } = type_wrapper {
return ReturnType::Extendable(ReturnTypeExtendable {
type_wrapper_tl: TypeWrapperTopLevel { ok, err: None },
wit_type,
});
} else if let TypeWrapper::Result { ok, err: Some(err) } = type_wrapper {
if let TypeWrapper::String = err.as_ref() {
return ReturnType::Extendable(ReturnTypeExtendable {
type_wrapper_tl: TypeWrapperTopLevel { ok, err: Some(err) },
wit_type,
});
} else if let TypeWrapper::Variant(fields) = err.as_ref()
&& let Some(None) = fields.get(&TypeKey::new_kebab(EXECUTION_FAILED_WIT_CASE))
{
return ReturnType::Extendable(ReturnTypeExtendable {
type_wrapper_tl: TypeWrapperTopLevel { ok, err: Some(err) },
wit_type,
});
}
return ReturnType::NonExtendable(ReturnTypeNonExtendable {
type_wrapper: TypeWrapper::Result { ok, err: Some(err) },
wit_type,
});
}
ReturnType::NonExtendable(ReturnTypeNonExtendable {
type_wrapper: type_wrapper.clone(),
wit_type,
})
}
#[must_use]
pub fn wit_type(&self) -> &str {
match self {
ReturnType::Extendable(compatible) => compatible.wit_type.as_ref(),
ReturnType::NonExtendable(incompatible) => incompatible.wit_type.as_ref(),
}
}
#[must_use]
pub fn type_wrapper(&self) -> TypeWrapper {
match self {
ReturnType::Extendable(compatible) => {
TypeWrapper::from(compatible.type_wrapper_tl.clone())
}
ReturnType::NonExtendable(incompatible) => incompatible.type_wrapper.clone(),
}
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Eq, derive_more::Display)]
#[derive_where::derive_where(PartialEq)]
#[display("{wit_type}")]
pub struct ReturnTypeNonExtendable {
pub type_wrapper: TypeWrapper,
#[derive_where(skip)]
pub wit_type: StrVariant,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Eq, derive_more::Display)]
#[derive_where::derive_where(PartialEq)]
#[display("{wit_type}")]
pub struct ReturnTypeExtendable {
pub type_wrapper_tl: TypeWrapperTopLevel,
#[derive_where(skip)]
pub wit_type: StrVariant,
}
pub const RETURN_TYPE_DUMMY: ReturnType = ReturnType::Extendable(ReturnTypeExtendable {
type_wrapper_tl: TypeWrapperTopLevel {
ok: None,
err: None,
},
wit_type: StrVariant::Static("result"),
});
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Eq, derive_more::Display)]
#[derive_where::derive_where(PartialEq)]
#[display("{name}: {wit_type}")]
pub struct ParameterType {
pub type_wrapper: TypeWrapper,
#[derive_where(skip)]
pub name: StrVariant,
#[derive_where(skip)]
pub wit_type: StrVariant,
}
#[derive(
Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Default, derive_more::Deref,
)]
pub struct ParameterTypes(pub Vec<ParameterType>);
impl Debug for ParameterTypes {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "(")?;
let mut iter = self.0.iter().peekable();
while let Some(p) = iter.next() {
write!(f, "{p:?}")?;
if iter.peek().is_some() {
write!(f, ", ")?;
}
}
write!(f, ")")
}
}
impl Display for ParameterTypes {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "(")?;
let mut iter = self.0.iter().peekable();
while let Some(p) = iter.next() {
write!(f, "{p}")?;
if iter.peek().is_some() {
write!(f, ", ")?;
}
}
write!(f, ")")
}
}
#[derive(Debug, Clone)]
pub struct PackageIfcFns {
pub ifc_fqn: IfcFqnName,
pub extension: bool, pub fns: IndexMap<FnName, FunctionMetadata>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ComponentRetryConfig {
pub max_retries: Option<u32>,
pub retry_exp_backoff: Duration,
}
impl ComponentRetryConfig {
pub const ZERO: ComponentRetryConfig = ComponentRetryConfig {
max_retries: Some(0),
retry_exp_backoff: Duration::ZERO,
};
pub const WORKFLOW: ComponentRetryConfig = ComponentRetryConfig {
max_retries: None,
retry_exp_backoff: Duration::ZERO,
};
pub const CRON: ComponentRetryConfig = ComponentRetryConfig {
max_retries: None,
retry_exp_backoff: Duration::ZERO,
};
}
pub trait FunctionRegistry: Send + Sync {
fn get_by_exported_function(
&self,
ffqn: &FunctionFqn,
) -> Option<(FunctionMetadata, ComponentId)>;
fn get_ret_type(&self, ffqn: &FunctionFqn) -> Option<TypeWrapperTopLevel> {
self.get_by_exported_function(ffqn)
.and_then(|(fn_meta, _)| {
if let ReturnType::Extendable(ReturnTypeExtendable {
type_wrapper_tl: type_wrapper,
wit_type: _,
}) = fn_meta.return_type
{
Some(type_wrapper)
} else {
None
}
})
}
fn all_exports(&self) -> &[PackageIfcFns];
}
type MetadataMap = hashbrown::HashMap<String, String, BuildHasherDefault<DefaultHasher>>;
#[derive(
Debug,
Default,
Clone,
Serialize,
Deserialize,
derive_more::Display,
PartialEq,
Eq,
schemars::JsonSchema,
)]
#[schemars(with = "std::collections::HashMap<String, String>")]
#[display("{_0:?}")]
pub struct ExecutionMetadata(MetadataMap);
impl ExecutionMetadata {
const LINKED_KEY: &str = "obelisk-tracing-linked";
const EMPTY_MAP: MetadataMap = hashbrown::HashMap::with_hasher(BuildHasherDefault::new());
#[must_use]
pub const fn empty() -> Self {
Self(Self::EMPTY_MAP)
}
#[must_use]
pub fn from_parent_span(less_specific: &Span) -> Self {
ExecutionMetadata::create(less_specific, false)
}
#[must_use]
pub fn from_linked_span(less_specific: &Span) -> Self {
ExecutionMetadata::create(less_specific, true)
}
#[must_use]
fn create(span: &Span, link_marker: bool) -> Self {
use tracing_opentelemetry::OpenTelemetrySpanExt as _;
let mut metadata = Self(hashbrown::HashMap::default());
let mut metadata_view = ExecutionMetadataInjectorView {
metadata: &mut metadata,
};
fn inject(s: &Span, metadata_view: &mut ExecutionMetadataInjectorView) {
opentelemetry::global::get_text_map_propagator(|propagator| {
propagator.inject_context(&s.context(), metadata_view);
});
}
inject(&Span::current(), &mut metadata_view);
if metadata_view.is_empty() {
inject(span, &mut metadata_view);
}
if link_marker {
metadata_view.set(Self::LINKED_KEY, String::new());
}
metadata
}
pub fn enrich(&self, span: &Span) {
use opentelemetry::trace::TraceContextExt as _;
use tracing_opentelemetry::OpenTelemetrySpanExt as _;
let metadata_view = ExecutionMetadataExtractorView { metadata: self };
let otel_context = opentelemetry::global::get_text_map_propagator(|propagator| {
propagator.extract(&metadata_view)
});
if metadata_view.get(Self::LINKED_KEY).is_some() {
let linked_span_context = otel_context.span().span_context().clone();
span.add_link(linked_span_context);
} else {
let _ = span.set_parent(otel_context);
}
}
}
struct ExecutionMetadataInjectorView<'a> {
metadata: &'a mut ExecutionMetadata,
}
impl ExecutionMetadataInjectorView<'_> {
fn is_empty(&self) -> bool {
self.metadata.0.is_empty()
}
}
impl opentelemetry::propagation::Injector for ExecutionMetadataInjectorView<'_> {
fn set(&mut self, key: &str, value: String) {
let key = format!("tracing:{key}");
self.metadata.0.insert(key, value);
}
}
struct ExecutionMetadataExtractorView<'a> {
metadata: &'a ExecutionMetadata,
}
impl opentelemetry::propagation::Extractor for ExecutionMetadataExtractorView<'_> {
fn get(&self, key: &str) -> Option<&str> {
self.metadata
.0
.get(&format!("tracing:{key}"))
.map(std::string::String::as_str)
}
fn keys(&self) -> Vec<&str> {
self.metadata
.0
.keys()
.filter_map(|key| key.strip_prefix("tracing:"))
.collect()
}
}
#[cfg(test)]
mod tests {
use rstest::rstest;
use crate::{
ExecutionFailureKind, ExecutionId, FinishedExecutionFailure, FunctionFqn, JoinSetId,
JoinSetKind, StrVariant, SupportedFunctionReturnValue, TypeWrapperTopLevel,
prefixed_ulid::ExecutorId,
};
use std::{
hash::{DefaultHasher, Hash, Hasher},
str::FromStr,
sync::Arc,
};
use val_json::{type_wrapper::TypeWrapper, wast_val::WastVal};
#[test]
fn execution_failure_projects_to_snake_case_string() {
let failure = SupportedFunctionReturnValue::ExecutionFailure(FinishedExecutionFailure {
kind: ExecutionFailureKind::TimedOut,
reason: None,
detail: None,
});
assert_eq!(
failure.into_wast_val_res(|| TypeWrapperTopLevel {
ok: None,
err: Some(Box::new(TypeWrapper::String)),
}),
Err(Some(Box::new(WastVal::String(
"execution_failed".to_string()
))))
);
}
#[test]
fn ulid_parsing() {
let generated = ExecutorId::generate();
let str = generated.to_string();
let parsed = str.parse().unwrap();
assert_eq!(generated, parsed);
}
#[test]
fn execution_id_parsing_top_level() {
let generated = ExecutionId::generate();
let str = generated.to_string();
let parsed = str.parse().unwrap();
assert_eq!(generated, parsed);
}
#[test]
fn execution_id_with_one_level_should_parse() {
let top_level = ExecutionId::generate();
let join_set_id = JoinSetId::new(JoinSetKind::Named, StrVariant::Static("name")).unwrap();
let first_child = ExecutionId::Derived(top_level.next_level(&join_set_id));
let ser = first_child.to_string();
assert_eq!(format!("{top_level}.n:name_1"), ser);
let parsed = ExecutionId::from_str(&ser).unwrap();
assert_eq!(first_child, parsed);
}
#[test]
fn execution_id_increment_twice() {
let top_level = ExecutionId::generate();
let join_set_id = JoinSetId::new(JoinSetKind::Named, StrVariant::Static("name")).unwrap();
let first_child = top_level.next_level(&join_set_id);
let second_child = ExecutionId::Derived(first_child.get_incremented());
let ser = second_child.to_string();
assert_eq!(format!("{top_level}.n:name_2"), ser);
let parsed = ExecutionId::from_str(&ser).unwrap();
assert_eq!(second_child, parsed);
}
#[test]
fn execution_id_next_level_twice() {
let top_level = ExecutionId::generate();
let join_set_id_outer =
JoinSetId::new(JoinSetKind::Generated, StrVariant::Static("gg")).unwrap();
let join_set_id_inner =
JoinSetId::new(JoinSetKind::OneOff, StrVariant::Static("oo")).unwrap();
let execution_id = ExecutionId::Derived(
top_level
.next_level(&join_set_id_outer)
.get_incremented()
.next_level(&join_set_id_inner)
.get_incremented(),
);
let ser = execution_id.to_string();
assert_eq!(format!("{top_level}.g:gg_2.o:oo_2"), ser);
let parsed = ExecutionId::from_str(&ser).unwrap();
assert_eq!(execution_id, parsed);
}
#[test]
fn execution_id_split_first_level() {
let top_level = ExecutionId::generate();
let join_set_id =
JoinSetId::new(JoinSetKind::Generated, StrVariant::Static("some")).unwrap();
let execution_id = top_level.next_level(&join_set_id);
let (actual_top_level, actual_join_set) = execution_id.split_to_parts();
assert_eq!(top_level, actual_top_level);
assert_eq!(join_set_id, actual_join_set);
}
#[rstest]
fn execution_id_split_second_level(#[values(0, 1)] outer_idx: u64) {
let top_level = ExecutionId::generate();
let join_set_id_outer =
JoinSetId::new(JoinSetKind::Generated, StrVariant::Static("some")).unwrap();
let first_level = top_level
.next_level(&join_set_id_outer)
.get_incremented_by(outer_idx);
let join_set_id_inner =
JoinSetId::new(JoinSetKind::Generated, StrVariant::Static("other")).unwrap();
let second_level = first_level.next_level(&join_set_id_inner);
let (actual_first_level, actual_join_set) = second_level.split_to_parts();
assert_eq!(ExecutionId::Derived(first_level), actual_first_level);
assert_eq!(join_set_id_inner, actual_join_set);
}
#[test]
fn invalid_execution_id_should_fail_to_parse() {
ExecutionId::from_str("E_01KBNHM5FQW81KP5Y3XMNX31K4.g:gg._2.o:oo_2").unwrap_err();
}
#[test]
fn execution_id_hash_should_be_stable() {
let parent = ExecutionId::from_parts(1, 2);
let join_set_id = JoinSetId::new(JoinSetKind::Named, StrVariant::Static("name")).unwrap();
let sibling_1 = parent.next_level(&join_set_id);
let sibling_2 = ExecutionId::Derived(sibling_1.get_incremented());
let sibling_1 = ExecutionId::Derived(sibling_1);
let join_set_id_inner =
JoinSetId::new(JoinSetKind::OneOff, StrVariant::Static("oo")).unwrap();
let child =
ExecutionId::Derived(sibling_1.next_level(&join_set_id_inner).get_incremented());
let parent = parent.random_seed();
let sibling_1 = sibling_1.random_seed();
let sibling_2 = sibling_2.random_seed();
let child = child.random_seed();
let vec = vec![parent, sibling_1, sibling_2, child];
insta::assert_debug_snapshot!(vec);
let set: hashbrown::HashSet<_> = vec.into_iter().collect();
assert_eq!(4, set.len());
}
#[test]
fn hash_of_str_variants_should_be_equal() {
let input = "foo";
let left = StrVariant::Arc(Arc::from(input));
let right = StrVariant::Static(input);
assert_eq!(left, right);
let mut left_hasher = DefaultHasher::new();
left.hash(&mut left_hasher);
let mut right_hasher = DefaultHasher::new();
right.hash(&mut right_hasher);
let left_hasher = left_hasher.finish();
let right_hasher = right_hasher.finish();
println!("left: {left_hasher:x}, right: {right_hasher:x}");
assert_eq!(left_hasher, right_hasher);
}
#[test]
fn ffqn_from_tuple_with_version_should_work() {
let ffqn = FunctionFqn::try_from_tuple("wasi:cli/run@0.2.0", "run").unwrap();
assert_eq!(FunctionFqn::new_static("wasi:cli/run@0.2.0", "run"), ffqn);
}
#[test]
fn ffqn_from_str_with_version_should_work() {
let ffqn = FunctionFqn::from_str("wasi:cli/run@0.2.0.run").unwrap();
assert_eq!(FunctionFqn::new_static("wasi:cli/run@0.2.0", "run"), ffqn);
}
#[tokio::test]
async fn join_set_serde_should_be_consistent() {
use crate::{JoinSetId, JoinSetKind};
use strum::IntoEnumIterator;
for kind in JoinSetKind::iter() {
let join_set_id = JoinSetId::new(kind, StrVariant::from("name")).unwrap();
let ser = serde_json::to_string(&join_set_id).unwrap();
let deser = serde_json::from_str(&ser).unwrap();
assert_eq!(join_set_id, deser);
}
}
mod ifc_fqn {
use crate::{IfcFqnName, IfcFqnParseError};
use std::str::FromStr;
#[test]
fn parses_valid_without_version() {
let fqn = IfcFqnName::from_str("ns:pkg/ifc").unwrap();
assert_eq!(fqn.namespace(), "ns");
assert_eq!(fqn.package_name(), "pkg");
assert_eq!(fqn.ifc_name(), "ifc");
assert_eq!(fqn.version(), None);
}
#[test]
fn parses_valid_with_version() {
let fqn = IfcFqnName::from_str("ns:pkg/ifc@1.2.3").unwrap();
assert_eq!(fqn.version(), Some("1.2.3"));
}
#[test]
fn fails_missing_colon() {
let err = IfcFqnName::from_str("pkg/ifc").unwrap_err();
assert!(matches!(err, IfcFqnParseError::MissingNamespace(_)));
}
#[test]
fn fails_empty_namespace() {
let err = IfcFqnName::from_str(":pkg/ifc").unwrap_err();
assert!(matches!(err, IfcFqnParseError::EmptyNamespace(_)));
}
#[test]
fn fails_missing_slash() {
let err = IfcFqnName::from_str("ns:pkgifc").unwrap_err();
assert!(matches!(err, IfcFqnParseError::MissingPackageOrIfc(_)));
}
#[test]
fn fails_empty_package() {
let err = IfcFqnName::from_str("ns:/ifc").unwrap_err();
assert!(matches!(err, IfcFqnParseError::EmptyPackageName(_)));
}
#[test]
fn fails_empty_ifc() {
let err = IfcFqnName::from_str("ns:pkg/").unwrap_err();
assert!(matches!(err, IfcFqnParseError::EmptyIfcName(_)));
}
#[test]
fn fails_empty_version() {
let err = IfcFqnName::from_str("ns:pkg/ifc@").unwrap_err();
assert!(matches!(err, IfcFqnParseError::EmptyVersion(_)));
}
}
}