mod compfn;
use std::fmt::{Debug, Display, Formatter};
use std::future::Future;
use std::pin::Pin;
use std::rc::Rc;
pub use compfn::*;
pub type LocalBoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + 'a>>;
#[inline]
pub fn boxed<'a, T>(fut: impl Future<Output = T> + 'a) -> LocalBoxFuture<'a, T> {
Box::pin(fut)
}
#[allow(async_fn_in_trait)]
pub trait Operation: Clone {
type In: Clone;
type Out: Clone;
type Err: Clone;
async fn execute(&self, input: Self::In) -> Result<Self::Out, Self::Err>;
async fn compensate(&self, input: Self::In, result: Self::Out) -> Result<(), Self::Err>;
}
#[allow(clippy::type_complexity)]
pub fn operation<In: Clone, Out: Clone, Err: Clone>(
execute_fn: impl Fn(In) -> Pin<Box<dyn Future<Output = Result<Out, Err>>>> + 'static,
compensate_fn: impl Fn(In, Out) -> Pin<Box<dyn Future<Output = Result<(), Err>>>> + 'static,
) -> impl Operation<In = In, Out = Out, Err = Err> {
FnOperation {
execute_fn: Rc::new(execute_fn),
compensate_fn: Rc::new(compensate_fn),
}
}
pub fn sync_operation<In: Clone + 'static, Out: Clone + 'static, Err: Clone + 'static>(
execute_fn: impl Fn(In) -> Result<Out, Err> + 'static,
compensate_fn: impl Fn(In, Out) -> Result<(), Err> + 'static,
) -> impl Operation<In = In, Out = Out, Err = Err> {
let execute_fn = Rc::new(execute_fn);
let compensate_fn = Rc::new(compensate_fn);
operation(
move |input| {
let f = execute_fn.clone();
Box::pin(async move { f(input) })
},
move |input, output| {
let f = compensate_fn.clone();
Box::pin(async move { f(input, output) })
},
)
}
#[allow(clippy::type_complexity)]
struct FnOperation<In, Out, Err> {
execute_fn: Rc<dyn Fn(In) -> Pin<Box<dyn Future<Output = Result<Out, Err>>>>>,
compensate_fn: Rc<dyn Fn(In, Out) -> Pin<Box<dyn Future<Output = Result<(), Err>>>>>,
}
impl<In, Out, Err> Clone for FnOperation<In, Out, Err> {
fn clone(&self) -> Self {
Self {
execute_fn: self.execute_fn.clone(),
compensate_fn: self.compensate_fn.clone(),
}
}
}
impl<In: Clone, Out: Clone, Err: Clone> Operation for FnOperation<In, Out, Err> {
type In = In;
type Out = Out;
type Err = Err;
async fn execute(&self, input: In) -> Result<Out, Err> {
(self.execute_fn)(input).await
}
async fn compensate(&self, input: In, result: Out) -> Result<(), Err> {
(self.compensate_fn)(input, result).await
}
}
pub type TransactionResult<Out, Err> = Result<Out, TransactionFailure<Err>>;
#[derive(Debug)]
pub enum TransactionFailure<Err> {
RetryExhausted(Err),
RetryPolicyError { failure: Err, message: String },
FailedAndRolledBackCompletely(Err),
FailedAndRolledBackPartially {
failure: Err,
compensation_failure: Err,
},
}
impl<Err: Display> Display for TransactionFailure<Err> {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
match self {
TransactionFailure::RetryExhausted(err) => {
write!(f, "Transaction retry policy exhausted: {err}")
}
TransactionFailure::RetryPolicyError { failure, message } => write!(
f,
"Transaction failed with {failure}; retry policy error: {message}"
),
TransactionFailure::FailedAndRolledBackCompletely(err) => {
write!(
f,
"Transaction failed with {err} and rolled back completely."
)
}
TransactionFailure::FailedAndRolledBackPartially {
failure,
compensation_failure,
} => write!(
f,
"Transaction failed with {failure} and rolled back partially; compensation failed with: {compensation_failure}."
),
}
}
}
pub async fn fallible_transaction<Out, Err>(
f: impl for<'a> FnOnce(&'a mut FallibleTransaction<Err>) -> LocalBoxFuture<'a, Result<Out, Err>>,
) -> TransactionResult<Out, Err>
where
Err: Clone + 'static,
{
let mut transaction = FallibleTransaction::new();
match f(&mut transaction).await {
Ok(output) => Ok(output),
Err(error) => Err(transaction.on_fail(error).await),
}
}
pub async fn infallible_transaction<Out, Err: Clone + 'static>(
mut f: impl for<'a> FnMut(&'a mut FallibleTransaction<Err>) -> LocalBoxFuture<'a, Result<Out, Err>>,
) -> TransactionResult<Out, Err> {
let started = saga_host::now();
let mut state = Vec::new();
let mut attempt = 0_i64;
loop {
let mut transaction = FallibleTransaction::new();
match f(&mut transaction).await {
Ok(output) => return Ok(output),
Err(error) => {
let failure = transaction.on_fail(error).await;
let TransactionFailure::FailedAndRolledBackCompletely(error) = failure else {
return Err(failure);
};
attempt = attempt.saturating_add(1);
match saga_next(&state, saga_host::now().saturating_sub(started), attempt) {
Ok(decision) => {
state = decision.state;
match decision.delay {
Some(delay) => saga_host::wait(delay).await,
None => return Err(TransactionFailure::RetryExhausted(error)),
}
}
Err(message) => {
return Err(TransactionFailure::RetryPolicyError {
failure: error,
message,
});
}
}
}
}
}
}
fn saga_next(
state: &[u8],
elapsed: u64,
attempt: i64,
) -> Result<crate::bindings::golem::api::retry::RetryDecision, String> {
use crate::bindings::golem::api::retry::PredicateValue;
let noun = saga_host::noun()?;
let properties = vec![
("attempt".to_string(), PredicateValue::Integer(attempt)),
(
"error-kind".to_string(),
PredicateValue::Text("operation-error".to_string()),
),
];
let policy = saga_host::resolve("saga", &noun, &properties)?;
saga_host::next(policy.as_ref(), state, elapsed, "saga", &noun, &properties)
}
#[cfg(test)]
mod saga_host {
use crate::bindings::golem::api::retry::{PredicateValue, RetryDecision, RetryPolicy};
use std::cell::Cell;
thread_local! {
pub static CLOCK: Cell<u64> = const { Cell::new(0) };
pub static ELAPSED: Cell<u64> = const { Cell::new(0) };
pub static POLICY_ERROR: Cell<bool> = const { Cell::new(false) };
pub static RESOLVE_ERROR: Cell<bool> = const { Cell::new(false) };
}
pub fn now() -> u64 {
CLOCK.get()
}
pub async fn wait(delay: u64) {
CLOCK.set(CLOCK.get() + delay);
}
pub fn noun() -> Result<String, String> {
Ok("urn:worker:component/agent".into())
}
pub fn resolve(
verb: &str,
noun: &str,
properties: &[(String, PredicateValue)],
) -> Result<Option<RetryPolicy>, String> {
assert_eq!(verb, "saga");
assert_eq!(noun, "urn:worker:component/agent");
assert_eq!(properties[1].0, "error-kind");
assert!(
matches!(&properties[1].1, PredicateValue::Text(value) if value == "operation-error")
);
if RESOLVE_ERROR.replace(false) {
return Err("invalid resolver predicate".into());
}
Ok(None)
}
pub fn next(
policy: Option<&RetryPolicy>,
state: &[u8],
elapsed: u64,
verb: &str,
noun: &str,
properties: &[(String, PredicateValue)],
) -> Result<RetryDecision, String> {
assert!(policy.is_none());
ELAPSED.set(elapsed);
resolve(verb, noun, properties)?;
assert_eq!(properties[0].0, "attempt");
let PredicateValue::Integer(attempt) = properties[0].1 else {
panic!("missing attempt")
};
if POLICY_ERROR.replace(false) {
return Err("invalid policy".into());
}
let expected = if attempt == 1 {
vec![]
} else {
vec![attempt as u8 - 1]
};
assert_eq!(state, expected);
assert!(elapsed >= (attempt as u64 - 1) * 10);
Ok(RetryDecision {
state: vec![attempt as u8],
delay: (attempt < 3).then_some(10),
})
}
}
#[cfg(not(test))]
mod saga_host {
pub use crate::bindings::golem::api::retry::next_retry_delay as next;
pub use crate::retry::resolve_retry_policy as resolve;
pub fn now() -> u64 {
crate::wasip3::clocks::monotonic_clock::now()
}
pub async fn wait(delay: u64) {
crate::wasip3::clocks::monotonic_clock::wait_for(delay).await;
}
pub fn noun() -> Result<String, String> {
let metadata = crate::get_self_metadata().map_err(|e| format!("{e:?}"))?;
let id = metadata.agent_id;
Ok(format!(
"urn:worker:{}/{}",
id.component_id.uuid, id.agent_id
))
}
}
pub async fn transaction<Out: 'static, Err, T>(
f: impl for<'a> FnOnce(&'a mut T) -> LocalBoxFuture<'a, Result<Out, Err>>,
) -> TransactionResult<Out, Err>
where
T: Transaction<Err>,
{
T::run(f).await
}
#[allow(clippy::type_complexity)]
struct CompensationAction<Err> {
action: Box<dyn FnOnce() -> Pin<Box<dyn Future<Output = Result<(), Err>>>>>,
}
impl<Err> CompensationAction<Err> {
pub async fn execute(self) -> Result<(), Err> {
(self.action)().await
}
}
pub struct FallibleTransaction<Err> {
compensations: Vec<CompensationAction<Err>>,
}
impl<Err: Clone + 'static> FallibleTransaction<Err> {
fn new() -> Self {
Self {
compensations: Vec::new(),
}
}
pub async fn execute<OpIn: Clone + 'static, OpOut: Clone + 'static>(
&mut self,
operation: impl Operation<In = OpIn, Out = OpOut, Err = Err> + 'static,
input: OpIn,
) -> Result<OpOut, Err> {
let result = operation.execute(input.clone()).await;
if let Ok(output) = &result {
let cloned_op = operation.clone();
let cloned_out = output.clone();
self.compensations.push(CompensationAction {
action: Box::new(move || {
Box::pin(async move {
cloned_op
.compensate(input.clone(), cloned_out.clone())
.await
})
}),
});
}
result
}
async fn on_fail(&mut self, failure: Err) -> TransactionFailure<Err> {
for compensation_action in self.compensations.drain(..).rev() {
if let Err(compensation_failure) = compensation_action.execute().await {
return TransactionFailure::FailedAndRolledBackPartially {
failure,
compensation_failure,
};
}
}
TransactionFailure::FailedAndRolledBackCompletely(failure)
}
}
#[allow(async_fn_in_trait)]
pub trait Transaction<Err> {
async fn execute<OpIn: Clone + 'static, OpOut: Clone + 'static>(
&mut self,
operation: impl Operation<In = OpIn, Out = OpOut, Err = Err> + 'static,
input: OpIn,
) -> Result<OpOut, Err>;
async fn fail(&mut self, error: Err) -> Result<(), Err>;
async fn run<Out: 'static>(
f: impl for<'a> FnOnce(&'a mut Self) -> LocalBoxFuture<'a, Result<Out, Err>>,
) -> TransactionResult<Out, Err>;
}
impl<Err: Clone + 'static> Transaction<Err> for FallibleTransaction<Err> {
async fn execute<OpIn: Clone + 'static, OpOut: Clone + 'static>(
&mut self,
operation: impl Operation<In = OpIn, Out = OpOut, Err = Err> + 'static,
input: OpIn,
) -> Result<OpOut, Err> {
FallibleTransaction::execute(self, operation, input).await
}
async fn fail(&mut self, error: Err) -> Result<(), Err> {
Err(error)
}
async fn run<Out: 'static>(
f: impl for<'a> FnOnce(&'a mut Self) -> LocalBoxFuture<'a, Result<Out, Err>>,
) -> TransactionResult<Out, Err> {
fallible_transaction(f).await
}
}
#[cfg(test)]
mod tests {
use std::cell::RefCell;
use std::rc::Rc;
use test_r::test;
use crate::{boxed, fallible_transaction, infallible_transaction, sync_operation};
#[test]
async fn bounded_retries_wait_and_exhaust() {
super::saga_host::CLOCK.set(0);
let op = sync_operation(
|()| {
super::saga_host::CLOCK.set(super::saga_host::CLOCK.get() + 2);
Ok::<_, &str>(())
},
|(), ()| {
super::saga_host::CLOCK.set(super::saga_host::CLOCK.get() + 3);
Ok(())
},
);
let mut attempts = 0;
let result = infallible_transaction(|tx| {
attempts += 1;
let op = op.clone();
boxed(async move {
tx.execute(op, ()).await?;
Err::<(), _>("operation")
})
})
.await;
assert!(matches!(
result,
Err(crate::TransactionFailure::RetryExhausted("operation"))
));
assert_eq!(attempts, 3);
assert_eq!(super::saga_host::CLOCK.get(), 35);
assert_eq!(super::saga_host::ELAPSED.get(), 35);
}
#[test]
async fn resolver_error_preserves_original_failure() {
super::saga_host::RESOLVE_ERROR.set(true);
let mut attempts = 0;
let result = infallible_transaction(|_| {
attempts += 1;
boxed(async { Err::<(), _>("original") })
})
.await;
assert!(matches!(
result,
Err(crate::TransactionFailure::RetryPolicyError { failure: "original", message })
if message == "invalid resolver predicate"
));
assert_eq!(attempts, 1);
}
#[test]
async fn invalid_policy_never_retries() {
super::saga_host::POLICY_ERROR.set(true);
let mut attempts = 0;
let result = infallible_transaction(|_| {
attempts += 1;
boxed(async { Err::<(), _>("operation") })
})
.await;
assert!(matches!(
result,
Err(crate::TransactionFailure::RetryPolicyError {
failure: "operation",
..
})
));
assert_eq!(attempts, 1);
}
#[test]
async fn compensation_order_and_partial_failure() {
let log = Rc::new(RefCell::new(Vec::new()));
let compensation_log = log.clone();
let op = sync_operation(
|input: u8| Ok::<_, &str>(input),
move |input, output| {
assert_eq!(input, output);
compensation_log.borrow_mut().push(input);
if input == 2 {
Err("compensation")
} else {
Ok(())
}
},
);
let result = fallible_transaction(|tx| {
boxed(async move {
tx.execute(op.clone(), 1).await?;
tx.execute(op.clone(), 2).await?;
tx.execute(op, 3).await?;
Err::<(), _>("body")
})
})
.await;
assert!(matches!(
result,
Err(crate::TransactionFailure::FailedAndRolledBackPartially {
failure: "body",
compensation_failure: "compensation"
})
));
assert_eq!(*log.borrow(), [3, 2]);
}
#[test]
async fn infallible_stops_on_compensation_failure() {
let op = sync_operation(|()| Ok::<_, &str>(()), |(), ()| Err("compensation"));
let mut attempts = 0;
let result = infallible_transaction(|tx| {
attempts += 1;
assert_eq!(attempts, 1);
let op = op.clone();
boxed(async move {
tx.execute(op, ()).await?;
Err::<(), _>("body")
})
})
.await;
assert!(matches!(
result,
Err(crate::TransactionFailure::FailedAndRolledBackPartially {
failure: "body",
compensation_failure: "compensation"
})
));
}
#[test]
async fn fallible_compensates_successful_operations_only() {
let log = Rc::new(RefCell::new(Vec::new()));
let log1 = log.clone();
let log2 = log.clone();
let log3 = log.clone();
let log4 = log.clone();
let op1 = sync_operation(
move |input: String| {
log1.borrow_mut().push(format!("op1 execute {input}"));
Ok(())
},
move |input: String, _| {
log2.borrow_mut().push(format!("op1 rollback {input}"));
Ok(())
},
);
let op2 = sync_operation(
move |_: ()| {
log3.clone().borrow_mut().push("op2 execute".to_string());
Err::<(), &str>("op2 error")
},
move |_: (), _| {
log4.clone().borrow_mut().push("op2 rollback".to_string());
Ok(())
},
);
let result = fallible_transaction(|tx| {
boxed(async move {
println!("First we execute op1");
tx.execute(op1, "hello".to_string()).await?;
println!("Then execute op2");
tx.execute(op2, ()).await?;
println!("Finally compute a result");
Ok(11)
})
})
.await;
assert!(matches!(
result,
Err(crate::TransactionFailure::FailedAndRolledBackCompletely(
"op2 error"
))
));
assert_eq!(
*log.borrow(),
["op1 execute hello", "op2 execute", "op1 rollback hello"]
);
}
#[test]
async fn infallible_retries_with_fresh_compensations() {
let log = Rc::new(RefCell::new(Vec::new()));
let log1 = log.clone();
let log2 = log.clone();
let log3 = log.clone();
let log4 = log.clone();
let op1 = sync_operation(
move |input: String| {
log1.borrow_mut().push(format!("op1 execute {input}"));
Ok::<(), &str>(())
},
move |input: String, _| {
log2.borrow_mut().push(format!("op1 rollback {input}"));
Ok(())
},
);
let attempts = Rc::new(std::cell::Cell::new(0));
let count = attempts.clone();
let op2 = sync_operation(
move |_: ()| {
log3.clone().borrow_mut().push("op2 execute".to_string());
count.set(count.get() + 1);
if count.get() < 3 {
Err("op2 error")
} else {
Ok(())
}
},
move |_: (), r| {
log4.clone()
.borrow_mut()
.push(format!("op2 rollback {r:?}"));
Ok(())
},
);
let result = infallible_transaction(|tx| {
let op1 = op1.clone();
let op2 = op2.clone();
boxed(async move {
println!("First we execute op1");
tx.execute(op1.clone(), "hello".to_string()).await?;
tx.execute(op1, "world".to_string()).await?;
println!("Then execute op2");
tx.execute(op2, ()).await?;
println!("Finally compute a result");
Ok(11)
})
})
.await;
assert_eq!(result.unwrap(), 11);
assert_eq!(attempts.get(), 3);
assert_eq!(
*log.borrow(),
[
"op1 execute hello",
"op1 execute world",
"op2 execute",
"op1 rollback world",
"op1 rollback hello",
"op1 execute hello",
"op1 execute world",
"op2 execute",
"op1 rollback world",
"op1 rollback hello",
"op1 execute hello",
"op1 execute world",
"op2 execute"
]
);
}
}
#[cfg(test)]
#[cfg(feature = "macro")]
mod macro_tests {
use crate::{boxed, fallible_transaction, infallible_transaction};
use golem_rust_macro::golem_operation;
use test_r::test;
mod golem_rust {
pub use crate::*;
}
#[golem_operation(compensation=test_compensation)]
fn test_operation(input1: u64, input2: f32) -> Result<bool, String> {
println!("Op input: {input1}, {input2}");
Ok(true)
}
fn test_compensation(_: bool, input1: u64, input2: f32) -> Result<(), String> {
println!("Compensation input: {input1}, {input2}");
Ok(())
}
#[golem_operation(compensation=test_compensation_2)]
fn test_operation_2(input1: u64, input2: f32) -> Result<bool, String> {
println!("Op input: {input1}, {input2}");
Ok(true)
}
fn test_compensation_2(result: bool) -> Result<(), String> {
println!("Compensation for operation result {result:?}");
Ok(())
}
#[golem_operation(compensation=test_compensation_3)]
fn test_operation_3(input: String) -> Result<(), String> {
println!("Op input: {input}");
Ok(())
}
fn test_compensation_3() -> Result<(), String> {
println!("Compensation for operation, not using any input");
Ok(())
}
#[golem_operation(compensation=test_compensation_4)]
fn test_operation_4(input: u64) -> Result<(), String> {
println!("Op input: {input}");
Ok(())
}
fn test_compensation_4(_: (), input: u64) -> Result<(), String> {
println!("Compensation for operation with single input {input}");
Ok(())
}
#[test]
#[ignore]
async fn tx_test_1() {
let result = fallible_transaction(|tx| {
boxed(async move {
println!("Executing the annotated function as an operation directly");
tx.test_operation(1, 0.1).await?;
tx.test_operation_2(1, 0.1).await?;
tx.test_operation_3("test".to_string()).await?;
tx.test_operation_4(1).await?;
Ok(11)
})
})
.await;
println!("{result:?}");
}
#[test]
#[ignore]
async fn tx_test_2() {
let result = infallible_transaction(|tx| {
boxed(async move {
println!("Executing the annotated function as an operation directly");
tx.test_operation(1, 0.1).await?;
Ok(11)
})
})
.await;
println!("{result:?}");
}
}