diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index 99d53e030..b1babe65c 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -35,6 +35,10 @@ const W_ID: usize = 36; const W_ROLE: usize = 25; const W_STATUS: usize = 15; +/// Delay before retrying a transiently failed state transition, so a +/// misbehaving directory or relay is not hammered in a tight loop. +const TRANSIENT_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(5); + #[derive(Clone)] pub(crate) struct App { config: Config, @@ -697,27 +701,37 @@ impl App { async fn post_original_proposal( &self, - sender: Sender, + mut sender: Sender, persister: &SenderPersister, ) -> Result<()> { - let (response, ctx) = - self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await?; - let sender = sender.process_response(&response.bytes().await?, ctx).save(persister)?; - println!("Posted Original PSBT..."); - self.get_proposed_payjoin_psbt(sender, persister).await + loop { + let (response, ctx) = + self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await?; + match sender.process_response(&response.bytes().await?, ctx).save(persister) { + Ok(sender) => { + println!("Posted Original PSBT..."); + return self.get_proposed_payjoin_psbt(sender, persister).await; + } + Err(e) if e.is_transient() => { + tracing::debug!("Transient error posting original proposal, retrying: {e:?}"); + sender = e.transient_state().expect("transient error carries current state"); + tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; + } + Err(e) => return Err(e.into()), + } + } } async fn get_proposed_payjoin_psbt( &self, - sender: Sender, + mut sender: Sender, persister: &SenderPersister, ) -> Result<()> { - let mut session = sender.clone(); // Long poll until we get a response loop { let (response, ctx) = - self.post_via_relay(|relay| session.create_poll_request(relay)).await?; - let res = session.process_response(&response.bytes().await?, ctx).save(persister); + self.post_via_relay(|relay| sender.create_poll_request(relay)).await?; + let res = sender.process_response(&response.bytes().await?, ctx).save(persister); match res { Ok(OptionalTransitionOutcome::Progress(psbt)) => { println!("Proposal received. Processing..."); @@ -726,8 +740,12 @@ impl App { } Ok(OptionalTransitionOutcome::Stasis(current_state)) => { println!("No response yet."); - session = current_state; - continue; + sender = current_state; + } + Err(e) if e.is_transient() => { + tracing::debug!("Transient error polling for proposal, retrying: {e:?}"); + sender = e.transient_state().expect("transient error carries current state"); + tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; } Err(re) => { println!("{re}"); @@ -740,10 +758,9 @@ impl App { async fn long_poll_fallback( &self, - session: Receiver, + mut session: Receiver, persister: &ReceiverPersister, ) -> Result> { - let mut session = session; loop { println!("Polling receive request..."); let (ohttp_response, context) = @@ -758,7 +775,11 @@ impl App { } Ok(OptionalTransitionOutcome::Stasis(current_state)) => { session = current_state; - continue; + } + Err(e) if e.is_transient() => { + tracing::debug!("Transient error polling for request, retrying: {e:?}"); + session = e.transient_state().expect("transient error carries current state"); + tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; } Err(e) => return Err(e.into()), } @@ -946,24 +967,34 @@ impl App { async fn send_payjoin_proposal( &self, - proposal: Receiver, + mut proposal: Receiver, persister: &ReceiverPersister, ) -> Result<()> { - let (res, ohttp_ctx) = self - .post_via_relay(|relay| { - proposal - .create_post_request(relay) - .map_err(|e| anyhow!("v2 req extraction failed {}", e)) - }) - .await?; - let payjoin_psbt = proposal.psbt().clone(); - let session = proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister)?; - println!( - "Response successful. Watch mempool for successful Payjoin. TXID: {}", - payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid() - ); - - return self.monitor_payjoin_proposal(session, persister).await; + loop { + let (res, ohttp_ctx) = self + .post_via_relay(|relay| { + proposal + .create_post_request(relay) + .map_err(|e| anyhow!("v2 req extraction failed {}", e)) + }) + .await?; + let payjoin_psbt = proposal.psbt().clone(); + match proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister) { + Ok(session) => { + println!( + "Response successful. Watch mempool for successful Payjoin. TXID: {}", + payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid() + ); + return self.monitor_payjoin_proposal(session, persister).await; + } + Err(e) if e.is_transient() => { + tracing::debug!("Transient error sending payjoin proposal, retrying: {e:?}"); + proposal = e.transient_state().expect("transient error carries current state"); + tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; + } + Err(e) => return Err(e.into()), + } + } } async fn monitor_payjoin_proposal( @@ -1014,38 +1045,47 @@ impl App { /// Handle error by attempting to send an error response over the directory async fn handle_error( &self, - session: Receiver, + mut session: Receiver, persister: &ReceiverPersister, ) -> Result<()> { - let (err_response, err_ctx) = self - .post_via_relay(|relay| { - session - .create_error_request(relay) - .map_err(|e| anyhow!("Failed to post error request: {}", e)) - }) - .await?; - - let err_bytes = match err_response.bytes().await { - Ok(bytes) => bytes, - Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)), - }; + loop { + let (err_response, err_ctx) = self + .post_via_relay(|relay| { + session + .create_error_request(relay) + .map_err(|e| anyhow!("Failed to post error request: {}", e)) + }) + .await?; + + let err_bytes = match err_response.bytes().await { + Ok(bytes) => bytes, + Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)), + }; - if let Err(e) = session.process_error_response(&err_bytes, err_ctx).save(persister) { - if let Some(api_err) = e.api_error_ref() { - tracing::warn!("Failed to confirm error response delivery: {api_err}"); - } - match e.error_state() { - Some(_) => { - let id = persister.session_id(); - println!( - "Session {id} failed. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction." - ); + match session.process_error_response(&err_bytes, err_ctx).save(persister) { + Ok(_) => return Ok(()), + Err(e) if e.is_transient() => { + tracing::debug!("Transient error posting error response, retrying: {e:?}"); + session = e.transient_state().expect("transient error carries current state"); + tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; + } + Err(e) => { + if let Some(api_err) = e.api_error_ref() { + tracing::warn!("Failed to confirm error response delivery: {api_err}"); + } + match e.error_state() { + Some(_) => { + let id = persister.session_id(); + println!( + "Session {id} failed. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction." + ); + return Ok(()); + } + None => return Err(anyhow!("Failed to process error response")), + } } - None => return Err(anyhow!("Failed to process error response")), } } - - Ok(()) } async fn post_request(&self, req: payjoin::Request) -> Result { diff --git a/payjoin-ffi/src/receive/error.rs b/payjoin-ffi/src/receive/error.rs index c3fbf8419..1425a36bb 100644 --- a/payjoin-ffi/src/receive/error.rs +++ b/payjoin-ffi/src/receive/error.rs @@ -59,9 +59,13 @@ impl ReceiverCreateRequestError { #[derive(Debug, thiserror::Error, uniffi::Error)] #[error(transparent)] pub enum ReceiverPersistedError { - /// rust-payjoin receiver error + /// A transient error: nothing was persisted, and the failed transition + /// can be retried by calling it again on the same receiver object #[error(transparent)] - Receiver(ReceiverError), + Transient(ReceiverError), + /// A fatal error: the session is closed or has moved to an error state + #[error(transparent)] + Fatal(ReceiverError), /// Storage error that could occur at application storage layer #[error(transparent)] Storage(Arc), @@ -76,23 +80,30 @@ macro_rules! impl_persisted_error_from { $api_error_ty:ty, $receiver_arm:expr ) => { - impl From> + impl From> for ReceiverPersistedError where S: std::error::Error + Send + Sync + 'static, E: std::fmt::Debug, + C: std::fmt::Debug, { - fn from(err: payjoin::persist::PersistedError<$api_error_ty, S, E>) -> Self { + fn from(err: payjoin::persist::PersistedError<$api_error_ty, S, E, C>) -> Self { if err.storage_error_ref().is_some() { if let Some(storage_err) = err.storage_error() { return ReceiverPersistedError::from(ImplementationError::new(storage_err)); } - return ReceiverPersistedError::Receiver(ReceiverError::Unexpected); + return ReceiverPersistedError::Fatal(ReceiverError::Unexpected); } + let is_transient = err.is_transient(); if let Some(api_err) = err.api_error() { - return ReceiverPersistedError::Receiver($receiver_arm(api_err)); + let receiver_err = $receiver_arm(api_err); + return if is_transient { + ReceiverPersistedError::Transient(receiver_err) + } else { + ReceiverPersistedError::Fatal(receiver_err) + }; } - ReceiverPersistedError::Receiver(ReceiverError::Unexpected) + ReceiverPersistedError::Fatal(ReceiverError::Unexpected) } } }; @@ -320,4 +331,50 @@ mod tests { receiver.create_poll_request(EXAMPLE_URL).map(|_| ()).expect_err("session is expired"); assert!(ReceiverCreateRequestError::from(expired).is_expired()); } + + #[cfg(feature = "_test-utils")] + #[test] + fn persisted_error_classifies_transient_and_fatal() { + use std::str::FromStr; + + use payjoin::bitcoin::Address; + use payjoin::directory::ENCAPSULATED_MESSAGE_BYTES; + use payjoin::persist::InMemoryPersister; + use payjoin::receive::v2::{ReceiverBuilder, SessionEvent}; + use payjoin::OhttpKeys; + use payjoin_test_utils::EXAMPLE_URL; + + let address = Address::from_str("tb1q6d3a2w975yny0asuvd9a67ner4nks58ff0q8g4") + .expect("valid address") + .assume_checked(); + let ohttp_keys = OhttpKeys::decode(&payjoin_test_utils::ohttp_key_config_bytes()) + .expect("valid ohttp keys"); + let persister = InMemoryPersister::::default(); + let receiver = ReceiverBuilder::new(address, EXAMPLE_URL, ohttp_keys) + .expect("valid builder") + .build() + .save(&persister) + .expect("in-memory persister is infallible"); + + // An undersized directory response fails the size check, which is + // retryable, so the binding error classifies it as transient. + let (_req, ctx) = receiver.create_poll_request(EXAMPLE_URL).expect("valid poll request"); + let err = receiver + .clone() + .process_response(&[0u8; 1], ctx) + .save(&persister) + .expect_err("undersized response should fail"); + assert!(err.is_transient()); + assert!(matches!(ReceiverPersistedError::from(err), ReceiverPersistedError::Transient(_))); + + // A right-sized garbage body fails OHTTP decapsulation, which is + // fatal, so the binding error classifies it as fatal. + let (_req, ctx) = receiver.create_poll_request(EXAMPLE_URL).expect("valid poll request"); + let err = receiver + .process_response(&[0u8; ENCAPSULATED_MESSAGE_BYTES], ctx) + .save(&persister) + .expect_err("garbage response should fail"); + assert!(err.is_fatal()); + assert!(matches!(ReceiverPersistedError::from(err), ReceiverPersistedError::Fatal(_))); + } } diff --git a/payjoin-ffi/src/receive/mod.rs b/payjoin-ffi/src/receive/mod.rs index 8c746ef54..017d829f1 100644 --- a/payjoin-ffi/src/receive/mod.rs +++ b/payjoin-ffi/src/receive/mod.rs @@ -748,6 +748,7 @@ pub struct UncheckedOriginalPayloadTransition( payjoin::receive::v2::Receiver, payjoin::receive::Error, payjoin::receive::v2::Receiver, + payjoin::receive::v2::Receiver, >, >, >, @@ -829,6 +830,7 @@ pub struct MaybeInputsOwnedTransition( payjoin::receive::v2::Receiver, payjoin::receive::Error, payjoin::receive::v2::Receiver, + payjoin::receive::v2::Receiver, >, >, >, @@ -884,6 +886,7 @@ pub struct MaybeInputsSeenTransition( payjoin::receive::v2::Receiver, payjoin::receive::Error, payjoin::receive::v2::Receiver, + payjoin::receive::v2::Receiver, >, >, >, @@ -937,6 +940,7 @@ pub struct OutputsUnknownTransition( payjoin::receive::v2::Receiver, payjoin::receive::Error, payjoin::receive::v2::Receiver, + payjoin::receive::v2::Receiver, >, >, >, @@ -1137,6 +1141,8 @@ pub struct WantsFeeRangeTransition( payjoin::receive::v2::SessionEvent, payjoin::receive::v2::Receiver, payjoin::receive::ProtocolError, + (), + payjoin::receive::v2::Receiver, >, >, >, @@ -1217,6 +1223,7 @@ pub struct ProvisionalProposalTransition( payjoin::receive::v2::SessionEvent, payjoin::receive::v2::Receiver, payjoin::ImplementationError, + payjoin::receive::v2::Receiver, >, >, >, @@ -1279,6 +1286,7 @@ pub struct PayjoinProposalTransition( payjoin::receive::v2::Receiver, payjoin::receive::ProtocolError, payjoin::receive::v2::Receiver, + payjoin::receive::v2::Receiver, >, >, >, @@ -1368,6 +1376,7 @@ pub struct HasReplyableErrorTransition( payjoin::receive::v2::SessionEvent, payjoin::receive::v2::Receiver, payjoin::receive::ProtocolError, + payjoin::receive::v2::Receiver, >, >, >, diff --git a/payjoin-ffi/src/send/error.rs b/payjoin-ffi/src/send/error.rs index 96fdec825..0685a34b4 100644 --- a/payjoin-ffi/src/send/error.rs +++ b/payjoin-ffi/src/send/error.rs @@ -11,16 +11,16 @@ use crate::error::{FfiValidationError, ImplementationError}; #[derive(Debug, PartialEq, Eq, thiserror::Error, uniffi::Object)] #[uniffi::export(Debug, Display, Eq)] #[error("Error initializing the sender: {msg}")] -pub struct BuildSenderError { +pub struct SenderBuilderError { msg: String, } -impl From for BuildSenderError { - fn from(value: PsbtParseError) -> Self { BuildSenderError { msg: value.to_string() } } +impl From for SenderBuilderError { + fn from(value: PsbtParseError) -> Self { SenderBuilderError { msg: value.to_string() } } } -impl From for BuildSenderError { - fn from(value: send::BuildSenderError) -> Self { BuildSenderError { msg: value.to_string() } } +impl From for SenderBuilderError { + fn from(value: send::BuildSenderError) -> Self { SenderBuilderError { msg: value.to_string() } } } /// FFI-visible PSBT parsing error surfaced at the sender boundary. @@ -41,7 +41,7 @@ pub enum SenderInputError { #[error(transparent)] Psbt(PsbtParseError), #[error(transparent)] - Build(Arc), + Build(Arc), #[error(transparent)] FfiValidation(FfiValidationError), } @@ -177,85 +177,89 @@ impl SenderReplayError { pub fn is_expired(&self) -> bool { self.0.is_expired() } } -/// Error that may occur during state machine transitions +/// Error raised by a sender state machine transition #[derive(Debug, thiserror::Error, uniffi::Error)] #[error(transparent)] -pub enum SenderPersistedError { +pub enum SenderError { /// rust-payjoin sender Decapsulation error #[error(transparent)] - DecapsulationError(Arc), + Decapsulation(Arc), /// rust-payjoin sender response error #[error(transparent)] - ResponseError(ResponseError), + Response(ResponseError), /// Sender Build error #[error(transparent)] - BuildSenderError(Arc), - /// Storage error that could occur at application storage layer - #[error(transparent)] - Storage(Arc), + Build(Arc), /// Unexpected error #[error("An unexpected error occurred")] Unexpected, } +/// Error that may occur during state machine transitions +#[derive(Debug, thiserror::Error, uniffi::Error)] +#[error(transparent)] +pub enum SenderPersistedError { + /// A transient error: nothing was persisted, and the failed transition + /// can be retried by calling it again on the same sender object + #[error(transparent)] + Transient(SenderError), + /// A fatal error: the session is closed or has moved to an error state + #[error(transparent)] + Fatal(SenderError), + /// Storage error that could occur at application storage layer + #[error(transparent)] + Storage(Arc), +} + impl From for SenderPersistedError { fn from(value: ImplementationError) -> Self { SenderPersistedError::Storage(Arc::new(value)) } } -impl From> - for SenderPersistedError -where - S: std::error::Error + Send + Sync + 'static, -{ - fn from(err: payjoin::persist::PersistedError) -> Self { - if err.storage_error_ref().is_some() { - if let Some(storage_err) = err.storage_error() { - return SenderPersistedError::from(ImplementationError::new(storage_err)); +macro_rules! impl_sender_persisted_error_from { + ( + $api_error_ty:ty, + $sender_arm:expr + ) => { + impl From> + for SenderPersistedError + where + S: std::error::Error + Send + Sync + 'static, + C: std::fmt::Debug, + { + fn from(err: payjoin::persist::PersistedError<$api_error_ty, S, (), C>) -> Self { + if err.storage_error_ref().is_some() { + if let Some(storage_err) = err.storage_error() { + return SenderPersistedError::from(ImplementationError::new(storage_err)); + } + return SenderPersistedError::Fatal(SenderError::Unexpected); + } + let is_transient = err.is_transient(); + if let Some(api_err) = err.api_error() { + let sender_err = $sender_arm(api_err); + return if is_transient { + SenderPersistedError::Transient(sender_err) + } else { + SenderPersistedError::Fatal(sender_err) + }; + } + SenderPersistedError::Fatal(SenderError::Unexpected) } - return SenderPersistedError::Unexpected; - } - if let Some(api_err) = err.api_error() { - return SenderPersistedError::DecapsulationError(Arc::new(api_err.into())); } - SenderPersistedError::Unexpected - } + }; } -impl From> for SenderPersistedError -where - S: std::error::Error + Send + Sync + 'static, -{ - fn from(err: payjoin::persist::PersistedError) -> Self { - if err.storage_error_ref().is_some() { - if let Some(storage_err) = err.storage_error() { - return SenderPersistedError::from(ImplementationError::new(storage_err)); - } - return SenderPersistedError::Unexpected; - } - if let Some(api_err) = err.api_error() { - return SenderPersistedError::ResponseError(api_err.into()); - } - SenderPersistedError::Unexpected - } -} +impl_sender_persisted_error_from!( + send::v2::DecapsulationError, + |api_err: send::v2::DecapsulationError| SenderError::Decapsulation(Arc::new(api_err.into())) +); -impl From> for SenderPersistedError -where - S: std::error::Error + Send + Sync + 'static, -{ - fn from(err: payjoin::persist::PersistedError) -> Self { - if err.storage_error_ref().is_some() { - if let Some(storage_err) = err.storage_error() { - return SenderPersistedError::from(ImplementationError::new(storage_err)); - } - return SenderPersistedError::Unexpected; - } - if let Some(api_err) = err.api_error() { - return SenderPersistedError::BuildSenderError(Arc::new(api_err.into())); - } - SenderPersistedError::Unexpected - } -} +impl_sender_persisted_error_from!(send::ResponseError, |api_err: send::ResponseError| { + SenderError::Response(api_err.into()) +}); + +impl_sender_persisted_error_from!(send::BuildSenderError, |api_err: send::BuildSenderError| { + SenderError::Build(Arc::new(api_err.into())) +}); #[cfg(test)] mod tests { diff --git a/payjoin-ffi/src/send/mod.rs b/payjoin-ffi/src/send/mod.rs index d528d1389..53e4739aa 100644 --- a/payjoin-ffi/src/send/mod.rs +++ b/payjoin-ffi/src/send/mod.rs @@ -2,7 +2,7 @@ use std::str::FromStr; use std::sync::{Arc, RwLock}; pub use error::{ - BuildSenderError, CreateRequestError, DecapsulationError, PsbtParseError, ResponseError, + CreateRequestError, DecapsulationError, PsbtParseError, ResponseError, SenderBuilderError, SenderInputError, }; @@ -453,6 +453,8 @@ pub struct WithReplyKeyTransition( payjoin::send::v2::SessionEvent, payjoin::send::v2::Sender, payjoin::send::v2::DecapsulationError, + (), + payjoin::send::v2::Sender, >, >, >, diff --git a/payjoin/src/core/persist.rs b/payjoin/src/core/persist.rs index 5961da6cb..21458b94b 100644 --- a/payjoin/src/core/persist.rs +++ b/payjoin/src/core/persist.rs @@ -21,6 +21,16 @@ //! the subgraph of the state machine diagram which corresponds to the older //! version of the state machine. New sessions which do contain this event will //! not be interpretable by the old code. +//! +//! # Transient errors and typestate linearity +//! +//! State transitions consume the current typestate, and exactly one live +//! handle exists at any point: a successful transition returns the next +//! state, a transient rejection returns ownership of the current state +//! inside the error so the caller can retry in place, and a fatal rejection +//! closes the session. Transient rejections persist nothing, so replaying +//! the event log always reconstructs the same current state that the error +//! carries. use std::fmt; @@ -70,21 +80,26 @@ impl PersistActions { /// Handles cases where the transition either succeeds with a final result that ends the session, or hits a static condition and stays in the same state. /// State transition may also be a fatal error or transient error. #[must_use = "a transition must be persisted with .save() to advance the session"] +#[allow(clippy::type_complexity)] pub struct MaybeSuccessTransitionWithNoResults( - Result, Rejection>, + Result< + AcceptOptionalTransition, + Rejection, + >, ); impl MaybeSuccessTransitionWithNoResults where Err: std::error::Error, + CurrentState: fmt::Debug, { pub(crate) fn fatal(event: Event, error: Err) -> Self { MaybeSuccessTransitionWithNoResults(Err(Rejection::fatal(event, error))) } - pub(crate) fn transient(error: Err) -> Self { - MaybeSuccessTransitionWithNoResults(Err(Rejection::transient(error))) + pub(crate) fn transient(error: Err, current_state: CurrentState) -> Self { + MaybeSuccessTransitionWithNoResults(Err(Rejection::transient(error, current_state))) } pub(crate) fn no_results(current_state: CurrentState) -> Self { @@ -103,7 +118,10 @@ where self, ) -> ( PersistActions, - Result, ApiError>, + Result< + OptionalTransitionOutcome, + ApiError, + >, ) { match self.0 { Ok(AcceptOptionalTransition::Success(AcceptNextState(event, success_value))) => ( @@ -114,19 +132,20 @@ where (PersistActions::NoOp, Ok(OptionalTransitionOutcome::Stasis(current_state))), Err(Rejection::Fatal(RejectFatal(event, error))) => (PersistActions::SaveAndClose(event), Err(ApiError::Fatal(error))), - Err(Rejection::Transient(RejectTransient(error))) => - (PersistActions::NoOp, Err(ApiError::Transient(error))), + Err(Rejection::Transient(RejectTransient(error, current_state))) => + (PersistActions::NoOp, Err(ApiError::Transient(error, current_state))), Err(Rejection::ReplyableError(RejectReplyableError(event, _, error))) => (PersistActions::Save(event), Err(ApiError::Fatal(error))), } } + #[allow(clippy::type_complexity)] pub fn save

( self, persister: &P, ) -> Result< OptionalTransitionOutcome, - PersistedError, + PersistedError, > where P: SessionPersister, @@ -136,12 +155,13 @@ where Ok(outcome.map_err(InternalPersistedError::Api)?) } + #[allow(clippy::type_complexity)] pub async fn save_async

( self, persister: &P, ) -> Result< OptionalTransitionOutcome, - PersistedError, + PersistedError, > where P: AsyncSessionPersister, @@ -158,14 +178,19 @@ where /// A transition that can result in a state transition, fatal error, or successfully have no results. #[must_use = "a transition must be persisted with .save() to advance the session"] +#[allow(clippy::type_complexity)] pub struct MaybeFatalTransitionWithNoResults( - Result, Rejection>, + Result< + AcceptOptionalTransition, + Rejection, + >, ); impl MaybeFatalTransitionWithNoResults where Err: std::error::Error, + CurrentState: fmt::Debug, { pub(crate) fn fatal(event: Event, error: Err) -> Self { MaybeFatalTransitionWithNoResults(Err(Rejection::fatal(event, error))) @@ -175,8 +200,8 @@ where MaybeFatalTransitionWithNoResults(Ok(AcceptOptionalTransition::NoResults(current_state))) } - pub(crate) fn transient(error: Err) -> Self { - MaybeFatalTransitionWithNoResults(Err(Rejection::transient(error))) + pub(crate) fn transient(error: Err, current_state: CurrentState) -> Self { + MaybeFatalTransitionWithNoResults(Err(Rejection::transient(error, current_state))) } pub(crate) fn success(event: Event, next_state: NextState) -> Self { @@ -190,7 +215,7 @@ where self, ) -> ( PersistActions, - Result, ApiError>, + Result, ApiError>, ) { match self.0 { Ok(AcceptOptionalTransition::Success(AcceptNextState(event, next_state))) => @@ -199,19 +224,20 @@ where (PersistActions::NoOp, Ok(OptionalTransitionOutcome::Stasis(current_state))), Err(Rejection::Fatal(RejectFatal(event, error))) => (PersistActions::SaveAndClose(event), Err(ApiError::Fatal(error))), - Err(Rejection::Transient(RejectTransient(error))) => - (PersistActions::NoOp, Err(ApiError::Transient(error))), + Err(Rejection::Transient(RejectTransient(error, current_state))) => + (PersistActions::NoOp, Err(ApiError::Transient(error, current_state))), Err(Rejection::ReplyableError(RejectReplyableError(event, _, error))) => (PersistActions::Save(event), Err(ApiError::Fatal(error))), } } + #[allow(clippy::type_complexity)] pub fn save

( self, persister: &P, ) -> Result< OptionalTransitionOutcome, - PersistedError, + PersistedError, > where P: SessionPersister, @@ -221,12 +247,13 @@ where Ok(outcome.map_err(InternalPersistedError::Api)?) } + #[allow(clippy::type_complexity)] pub async fn save_async

( self, persister: &P, ) -> Result< OptionalTransitionOutcome, - PersistedError, + PersistedError, > where P: AsyncSessionPersister, @@ -241,23 +268,28 @@ where } } +pub(crate) type FatalTransitionResult = + Result, Rejection>; + /// A transition that can be either fatal, transient, or a state transition. #[must_use = "a transition must be persisted with .save() to advance the session"] -pub struct MaybeFatalTransition( - pub(crate) Result, Rejection>, +pub struct MaybeFatalTransition( + pub(crate) FatalTransitionResult, ); -impl MaybeFatalTransition +impl + MaybeFatalTransition where Err: std::error::Error, ErrorState: fmt::Debug, + CurrentState: fmt::Debug, { pub(crate) fn fatal(event: Event, error: Err) -> Self { MaybeFatalTransition(Err(Rejection::fatal(event, error))) } - pub(crate) fn transient(error: Err) -> Self { - MaybeFatalTransition(Err(Rejection::transient(error))) + pub(crate) fn transient(error: Err, current_state: CurrentState) -> Self { + MaybeFatalTransition(Err(Rejection::transient(error, current_state))) } pub(crate) fn success(event: Event, next_state: NextState) -> Self { @@ -268,15 +300,16 @@ where MaybeFatalTransition(Err(Rejection::replyable_error(event, error_state, error))) } + #[allow(clippy::type_complexity)] pub(crate) fn deconstruct( self, - ) -> (PersistActions, Result>) { + ) -> (PersistActions, Result>) { match self.0 { Ok(AcceptNextState(event, next_state)) => (PersistActions::Save(event), Ok(next_state)), Err(Rejection::Fatal(RejectFatal(event, error))) => (PersistActions::SaveAndClose(event), Err(ApiError::Fatal(error))), - Err(Rejection::Transient(RejectTransient(error))) => - (PersistActions::NoOp, Err(ApiError::Transient(error))), + Err(Rejection::Transient(RejectTransient(error, current_state))) => + (PersistActions::NoOp, Err(ApiError::Transient(error, current_state))), Err(Rejection::ReplyableError(RejectReplyableError(event, error_state, error))) => (PersistActions::Save(event), Err(ApiError::FatalWithState(error, error_state))), } @@ -285,7 +318,7 @@ where pub fn save

( self, persister: &P, - ) -> Result> + ) -> Result> where P: SessionPersister, { @@ -297,11 +330,12 @@ where pub async fn save_async

( self, persister: &P, - ) -> Result> + ) -> Result> where P: AsyncSessionPersister, Err: Send, ErrorState: Send, + CurrentState: Send, NextState: Send, Event: Send, { @@ -314,33 +348,39 @@ where /// A transition that can result in a state transition or a transient error. /// Fatal errors cannot occur in this transition. #[must_use = "a transition must be persisted with .save() to advance the session"] -pub struct MaybeTransientTransition( - Result, RejectTransient>, +pub struct MaybeTransientTransition( + Result, RejectTransient>, ); -impl MaybeTransientTransition +impl + MaybeTransientTransition where Err: std::error::Error, + CurrentState: fmt::Debug, { pub(crate) fn success(event: Event, next_state: NextState) -> Self { MaybeTransientTransition(Ok(AcceptNextState(event, next_state))) } - pub(crate) fn transient(error: Err) -> Self { - MaybeTransientTransition(Err(RejectTransient(error))) + pub(crate) fn transient(error: Err, current_state: CurrentState) -> Self { + MaybeTransientTransition(Err(RejectTransient(error, current_state))) } - pub(crate) fn deconstruct(self) -> (PersistActions, Result>) { + #[allow(clippy::type_complexity)] + pub(crate) fn deconstruct( + self, + ) -> (PersistActions, Result>) { match self.0 { Ok(AcceptNextState(event, next_state)) => (PersistActions::Save(event), Ok(next_state)), - Err(RejectTransient(error)) => (PersistActions::NoOp, Err(ApiError::Transient(error))), + Err(RejectTransient(error, current_state)) => + (PersistActions::NoOp, Err(ApiError::Transient(error, current_state))), } } pub fn save

( self, persister: &P, - ) -> Result> + ) -> Result> where P: SessionPersister, { @@ -352,10 +392,11 @@ where pub async fn save_async

( self, persister: &P, - ) -> Result> + ) -> Result> where P: AsyncSessionPersister, Err: Send, + CurrentState: Send, NextState: Send, Event: Send, { @@ -453,14 +494,16 @@ impl MaybeTerminalTransition { /// saved event keeps the session live for replay while the caller receives the /// fatal protocol error. #[must_use = "a transition must be persisted with .save() to advance the session"] -pub struct MaybeTerminalSuccessTransition( - MaybeTerminalSuccessOutcome, +pub struct MaybeTerminalSuccessTransition( + MaybeTerminalSuccessOutcome, ); -impl MaybeTerminalSuccessTransition +impl + MaybeTerminalSuccessTransition where Err: std::error::Error, NextState: fmt::Debug, + CurrentState: fmt::Debug, { pub(crate) fn advance(event: Event, next_state: NextState) -> Self { Self(MaybeTerminalSuccessOutcome::Advance(AcceptNextState(event, next_state))) @@ -478,14 +521,15 @@ where Self(MaybeTerminalSuccessOutcome::FatalTerminate(event, error)) } - pub(crate) fn transient(error: Err) -> Self { - Self(MaybeTerminalSuccessOutcome::Transient(error)) + pub(crate) fn transient(error: Err, current_state: CurrentState) -> Self { + Self(MaybeTerminalSuccessOutcome::Transient(error, current_state)) } #[allow(clippy::type_complexity)] pub(crate) fn deconstruct( self, - ) -> (PersistActions, Result, ApiError>) { + ) -> (PersistActions, Result, ApiError>) + { match self.0 { MaybeTerminalSuccessOutcome::Advance(AcceptNextState(event, next_state)) => (PersistActions::Save(event), Ok(Some(next_state))), @@ -495,15 +539,19 @@ where (PersistActions::Save(event), Err(ApiError::FatalWithState(error, next_state))), MaybeTerminalSuccessOutcome::FatalTerminate(event, error) => (PersistActions::SaveAndClose(event), Err(ApiError::Fatal(error))), - MaybeTerminalSuccessOutcome::Transient(error) => - (PersistActions::NoOp, Err(ApiError::Transient(error))), + MaybeTerminalSuccessOutcome::Transient(error, current_state) => + (PersistActions::NoOp, Err(ApiError::Transient(error, current_state))), } } + #[allow(clippy::type_complexity)] pub fn save

( self, persister: &P, - ) -> Result, PersistedError> + ) -> Result< + Option, + PersistedError, + > where P: SessionPersister, { @@ -512,14 +560,19 @@ where Ok(outcome.map_err(InternalPersistedError::Api)?) } + #[allow(clippy::type_complexity)] pub async fn save_async

( self, persister: &P, - ) -> Result, PersistedError> + ) -> Result< + Option, + PersistedError, + > where P: AsyncSessionPersister, Err: Send, NextState: Send, + CurrentState: Send, Event: Send, { let (actions, outcome) = self.deconstruct(); @@ -569,13 +622,14 @@ impl TerminalTransition { pub enum MaybeFatalOrSuccessTransition { Success(Event), NoResults(CurrentState), - Transient(RejectTransient), + Transient(RejectTransient), Fatal(RejectFatal), } impl MaybeFatalOrSuccessTransition where Err: std::error::Error, + CurrentState: fmt::Debug, { pub(crate) fn success(event: Event) -> Self { MaybeFatalOrSuccessTransition::Success(event) } @@ -584,8 +638,8 @@ where MaybeFatalOrSuccessTransition::Fatal(RejectFatal(event, error)) } - pub(crate) fn transient(error: Err) -> Self { - MaybeFatalOrSuccessTransition::Transient(RejectTransient(error)) + pub(crate) fn transient(error: Err, current_state: CurrentState) -> Self { + MaybeFatalOrSuccessTransition::Transient(RejectTransient(error, current_state)) } pub(crate) fn no_results(current_state: CurrentState) -> Self { @@ -595,26 +649,29 @@ where #[allow(clippy::type_complexity)] pub(crate) fn deconstruct( self, - ) -> (PersistActions, Result, ApiError>) - { + ) -> ( + PersistActions, + Result, ApiError>, + ) { match self { MaybeFatalOrSuccessTransition::Success(event) => (PersistActions::SaveAndClose(event), Ok(OptionalTransitionOutcome::Progress(()))), MaybeFatalOrSuccessTransition::NoResults(current_state) => (PersistActions::NoOp, Ok(OptionalTransitionOutcome::Stasis(current_state))), - MaybeFatalOrSuccessTransition::Transient(RejectTransient(error)) => - (PersistActions::NoOp, Err(ApiError::Transient(error))), + MaybeFatalOrSuccessTransition::Transient(RejectTransient(error, current_state)) => + (PersistActions::NoOp, Err(ApiError::Transient(error, current_state))), MaybeFatalOrSuccessTransition::Fatal(RejectFatal(event, error)) => (PersistActions::SaveAndClose(event), Err(ApiError::Fatal(error))), } } + #[allow(clippy::type_complexity)] pub fn save

( self, persister: &P, ) -> Result< OptionalTransitionOutcome<(), CurrentState>, - PersistedError, + PersistedError, > where P: SessionPersister, @@ -624,12 +681,13 @@ where Ok(outcome.map_err(InternalPersistedError::Api)?) } + #[allow(clippy::type_complexity)] pub async fn save_async

( self, persister: &P, ) -> Result< OptionalTransitionOutcome<(), CurrentState>, - PersistedError, + PersistedError, > where P: AsyncSessionPersister, @@ -651,12 +709,12 @@ enum MaybeTerminalOutcome { Terminate(Event), } -enum MaybeTerminalSuccessOutcome { +enum MaybeTerminalSuccessOutcome { Advance(AcceptNextState), Terminate(Event), FatalAdvance(Event, NextState, Err), FatalTerminate(Event, Err), - Transient(Err), + Transient(Err, CurrentState), } /// Wrapper that represents either a successful state transition or indicates no state change occurred @@ -668,15 +726,17 @@ pub enum AcceptOptionalTransition { } /// Wrapper representing a fatal or transient rejection of a state transition. -pub enum Rejection { +pub enum Rejection { Fatal(RejectFatal), - Transient(RejectTransient), + Transient(RejectTransient), ReplyableError(RejectReplyableError), } -impl Rejection { +impl Rejection { pub fn fatal(event: Event, error: Err) -> Self { Rejection::Fatal(RejectFatal(event, error)) } - pub fn transient(error: Err) -> Self { Rejection::Transient(RejectTransient(error)) } + pub fn transient(error: Err, current_state: CurrentState) -> Self { + Rejection::Transient(RejectTransient(error, current_state)) + } pub fn replyable_error(event: Event, error_state: ErrorState, error: Err) -> Self { Rejection::ReplyableError(RejectReplyableError(event, error_state, error)) } @@ -686,8 +746,10 @@ impl Rejection { /// When this error occurs, the session must be closed and cannot be resumed. pub struct RejectFatal(pub(crate) Event, pub(crate) Err); /// Represents a transient rejection of a state transition. -/// When this error occurs, the session should resume from its current state. -pub struct RejectTransient(pub(crate) Err); +/// When this error occurs, nothing is persisted and the session should resume +/// from the current state, which is carried alongside the error so the caller +/// can retry in place. +pub struct RejectTransient(pub(crate) Err, pub(crate) CurrentState); /// Represents a replyable error that transitions to an error state but keeps the session open. /// When this error occurs, the session transitions to the ErrorState. pub struct RejectReplyableError( @@ -700,9 +762,9 @@ pub struct RejectReplyableError( /// The wrapper contains the error and should be returned to the caller. pub struct RejectBadInitInputs(Err); -impl fmt::Display for RejectTransient { +impl fmt::Display for RejectTransient { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - let RejectTransient(err) = self; + let RejectTransient(err, _) = self; write!(f, "{err}") } } @@ -713,13 +775,16 @@ pub struct PersistedError< ApiError: std::error::Error, StorageError: std::error::Error, ErrorState: fmt::Debug = (), ->(InternalPersistedError); + CurrentState: fmt::Debug = (), +>(InternalPersistedError); -impl PersistedError +impl + PersistedError where StorageErr: std::error::Error, ApiErr: std::error::Error, ErrorState: fmt::Debug, + CurrentState: fmt::Debug, { #[allow(dead_code)] pub fn storage_error(self) -> Option { @@ -729,10 +794,15 @@ where } } + /// The protocol error that rejected the transition, regardless of whether + /// it was transient or fatal. + /// + /// On a transient error this drops the current state carried by the + /// error; use [`Self::transient_state`] to recover it instead. pub fn api_error(self) -> Option { match self.0 { InternalPersistedError::Api( - ApiError::Fatal(e) | ApiError::Transient(e) | ApiError::FatalWithState(e, _), + ApiError::Fatal(e) | ApiError::Transient(e, _) | ApiError::FatalWithState(e, _), ) => Some(e), _ => None, } @@ -748,7 +818,7 @@ where pub fn api_error_ref(&self) -> Option<&ApiErr> { match &self.0 { InternalPersistedError::Api( - ApiError::Fatal(e) | ApiError::Transient(e) | ApiError::FatalWithState(e, _), + ApiError::Fatal(e) | ApiError::Transient(e, _) | ApiError::FatalWithState(e, _), ) => Some(e), _ => None, } @@ -760,28 +830,77 @@ where _ => None, } } + + /// The typestate to retry from after a transient rejection. + /// + /// Transient rejections persist nothing, so the returned state is exactly + /// the state the failed transition was called on: retry by calling the + /// same transition on it again. Returns `None` for fatal and storage + /// errors. A storage error means the transition outcome is unknown, and + /// recovery is replaying the event log rather than retrying in memory. + pub fn transient_state(self) -> Option { + match self.0 { + InternalPersistedError::Api(ApiError::Transient(_, state)) => Some(state), + _ => None, + } + } + + /// True if the transition was rejected transiently: nothing was + /// persisted, and the session can be retried in place from the state + /// returned by [`Self::transient_state`]. + pub fn is_transient(&self) -> bool { + matches!(self.0, InternalPersistedError::Api(ApiError::Transient(..))) + } + + /// True if the transition failed fatally: an event was persisted, and + /// the session is closed or has moved to an error state (see + /// [`Self::error_state`]). + /// + /// Storage errors are neither transient nor fatal. They mean the + /// transition outcome is unknown, and recovery is replaying the event + /// log; detect them with [`Self::storage_error_ref`]. + pub fn is_fatal(&self) -> bool { + matches!( + self.0, + InternalPersistedError::Api(ApiError::Fatal(_) | ApiError::FatalWithState(..)) + ) + } } -impl - From> - for PersistedError +impl< + ApiError: std::error::Error, + StorageError: std::error::Error, + ErrorState: fmt::Debug, + CurrentState: fmt::Debug, + > From> + for PersistedError { - fn from(value: InternalPersistedError) -> Self { + fn from( + value: InternalPersistedError, + ) -> Self { PersistedError(value) } } -impl - std::error::Error for PersistedError +impl< + ApiError: std::error::Error, + StorageError: std::error::Error, + ErrorState: fmt::Debug, + CurrentState: fmt::Debug, + > std::error::Error for PersistedError { } -impl - fmt::Display for PersistedError +impl< + ApiErr: std::error::Error, + StorageError: std::error::Error, + ErrorState: fmt::Debug, + CurrentState: fmt::Debug, + > fmt::Display for PersistedError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match &self.0 { - InternalPersistedError::Api(ApiError::Transient(err)) => + InternalPersistedError::Api(ApiError::Transient(err, _)) => write!(f, "Transient error: {err}"), InternalPersistedError::Api( ApiError::Fatal(err) | ApiError::FatalWithState(err, _), @@ -792,9 +911,10 @@ impl { - /// Error indicating that the session should be retried from the same state - Transient(Err), +pub(crate) enum ApiError { + /// Error indicating that the session should be retried from the same state, + /// which is returned alongside the error + Transient(Err, CurrentState), /// Error indicating that the session is terminally closed Fatal(Err), /// Fatal error that results in a state transition to ErrorState @@ -802,26 +922,30 @@ pub(crate) enum ApiError { } #[derive(Debug)] -pub(crate) enum InternalPersistedError +pub(crate) enum InternalPersistedError where ApiErr: std::error::Error, StorageErr: std::error::Error, ErrorState: fmt::Debug, + CurrentState: fmt::Debug, { /// Error indicating that the session failed to progress to the next success state. - Api(ApiError), + Api(ApiError), /// Error indicating that application failed to save the session event. Storage(StorageErr), } -impl From> - for InternalPersistedError +impl From> + for InternalPersistedError where Err: std::error::Error, StorageErr: std::error::Error, ErrorState: fmt::Debug, + CurrentState: fmt::Debug, { - fn from(api: ApiError) -> Self { InternalPersistedError::Api(api) } + fn from(api: ApiError) -> Self { + InternalPersistedError::Api(api) + } } /// Represents a state transition that either progresses to a new state or maintains the current state @@ -1112,6 +1236,7 @@ mod tests { async fn test_maybe_transient_transition() { let event = InMemoryTestEvent("foo".to_string()); let next_state = "Next state".to_string(); + let current_state = "Current state".to_string(); let test_cases = vec![ TestCase { @@ -1128,15 +1253,24 @@ mod tests { }, }, TestCase { - make_transition: Box::new(|| { - MaybeTransientTransition::transient(InMemoryTestError {}) + make_transition: Box::new({ + let current_state = current_state.clone(); + move || { + MaybeTransientTransition::transient( + InMemoryTestError {}, + current_state.clone(), + ) + } }), expected_result: ExpectedResult { events: vec![], is_closed: false, error: Some( - InternalPersistedError::Api(ApiError::Transient(InMemoryTestError {})) - .into(), + InternalPersistedError::Api(ApiError::Transient( + InMemoryTestError {}, + current_state.clone(), + )) + .into(), ), success: None, }, @@ -1237,9 +1371,12 @@ mod tests { make_transition: Box::new({ let close_event = close_event.clone(); move || { - MaybeTerminalSuccessTransition::<_, InMemoryTestState, InMemoryTestError>::terminate( - close_event.clone(), - ) + MaybeTerminalSuccessTransition::< + _, + InMemoryTestState, + InMemoryTestError, + InMemoryTestState, + >::terminate(close_event.clone()) } }), expected_result: ExpectedResult { @@ -1274,9 +1411,13 @@ mod tests { make_transition: Box::new({ let fatal_close_event = fatal_close_event.clone(); move || { - MaybeTerminalSuccessTransition::<_, InMemoryTestState, InMemoryTestError>::fatal_terminate( - fatal_close_event.clone(), - InMemoryTestError {}, + MaybeTerminalSuccessTransition::< + _, + InMemoryTestState, + InMemoryTestError, + InMemoryTestState, + >::fatal_terminate( + fatal_close_event.clone(), InMemoryTestError {} ) } }), @@ -1290,19 +1431,28 @@ mod tests { }, }, TestCase { - make_transition: Box::new(|| { - MaybeTerminalSuccessTransition::< - InMemoryTestEvent, - InMemoryTestState, - InMemoryTestError, - >::transient(InMemoryTestError {}) + make_transition: Box::new({ + let current_state = "Current state".to_string(); + move || { + MaybeTerminalSuccessTransition::< + InMemoryTestEvent, + InMemoryTestState, + InMemoryTestError, + InMemoryTestState, + >::transient( + InMemoryTestError {}, current_state.clone() + ) + } }), expected_result: ExpectedResult { events: vec![], is_closed: false, error: Some( - InternalPersistedError::Api(ApiError::Transient(InMemoryTestError {})) - .into(), + InternalPersistedError::Api(ApiError::Transient( + InMemoryTestError {}, + "Current state".to_string(), + )) + .into(), ), success: None, }, @@ -1333,16 +1483,24 @@ mod tests { }, }, TestCase { - make_transition: Box::new(|| MaybeFatalTransition::transient(InMemoryTestError {})), + make_transition: Box::new(|| { + MaybeFatalTransition::transient( + InMemoryTestError {}, + "Current state".to_string(), + ) + }), expected_result: ExpectedResult::< _, - PersistedError, + PersistedError, > { events: vec![], is_closed: false, error: Some( - InternalPersistedError::Api(ApiError::Transient(InMemoryTestError {})) - .into(), + InternalPersistedError::Api(ApiError::Transient( + InMemoryTestError {}, + "Current state".to_string(), + )) + .into(), ), success: None, }, @@ -1399,7 +1557,12 @@ mod tests { }), expected_result: ExpectedResult::< OptionalTransitionOutcome, - PersistedError, + PersistedError< + InMemoryTestError, + std::convert::Infallible, + (), + InMemoryTestState, + >, > { events: vec![], is_closed: false, @@ -1408,15 +1571,24 @@ mod tests { }, }, TestCase { - make_transition: Box::new(|| { - MaybeSuccessTransitionWithNoResults::transient(InMemoryTestError {}) + make_transition: Box::new({ + let current_state = current_state.clone(); + move || { + MaybeSuccessTransitionWithNoResults::transient( + InMemoryTestError {}, + current_state.clone(), + ) + } }), expected_result: ExpectedResult { events: vec![], is_closed: false, error: Some( - InternalPersistedError::Api(ApiError::Transient(InMemoryTestError {})) - .into(), + InternalPersistedError::Api(ApiError::Transient( + InMemoryTestError {}, + current_state.clone(), + )) + .into(), ), success: None, }, @@ -1478,7 +1650,12 @@ mod tests { }), expected_result: ExpectedResult::< OptionalTransitionOutcome, - PersistedError, + PersistedError< + InMemoryTestError, + std::convert::Infallible, + (), + InMemoryTestState, + >, > { events: vec![], is_closed: false, @@ -1486,6 +1663,29 @@ mod tests { success: Some(OptionalTransitionOutcome::Stasis(current_state.clone())), }, }, + TestCase { + make_transition: Box::new({ + let current_state = current_state.clone(); + move || { + MaybeFatalTransitionWithNoResults::transient( + InMemoryTestError {}, + current_state.clone(), + ) + } + }), + expected_result: ExpectedResult { + events: vec![], + is_closed: false, + error: Some( + InternalPersistedError::Api(ApiError::Transient( + InMemoryTestError {}, + current_state.clone(), + )) + .into(), + ), + success: None, + }, + }, TestCase { make_transition: Box::new({ let error_event = error_event.clone(); @@ -1536,7 +1736,12 @@ mod tests { }), expected_result: ExpectedResult::< OptionalTransitionOutcome<(), InMemoryTestState>, - PersistedError, + PersistedError< + InMemoryTestError, + std::convert::Infallible, + (), + InMemoryTestState, + >, > { events: vec![], is_closed: false, @@ -1564,15 +1769,24 @@ mod tests { }, }, TestCase { - make_transition: Box::new(|| { - MaybeFatalOrSuccessTransition::transient(InMemoryTestError {}) + make_transition: Box::new({ + let current_state = current_state.clone(); + move || { + MaybeFatalOrSuccessTransition::transient( + InMemoryTestError {}, + current_state.clone(), + ) + } }), expected_result: ExpectedResult { events: vec![], is_closed: false, error: Some( - InternalPersistedError::Api(ApiError::Transient(InMemoryTestError {})) - .into(), + InternalPersistedError::Api(ApiError::Transient( + InMemoryTestError {}, + current_state.clone(), + )) + .into(), ), success: None, }, @@ -1592,6 +1806,9 @@ mod tests { ); assert!(storage_error.storage_error_ref().is_some()); assert!(storage_error.api_error_ref().is_none()); + assert!(!storage_error.is_transient()); + assert!(!storage_error.is_fatal()); + assert_eq!(storage_error.transient_state(), None); // Test Internal API error cases let fatal_error = PersistedError::( @@ -1599,11 +1816,32 @@ mod tests { ); assert!(fatal_error.storage_error_ref().is_none()); assert!(fatal_error.api_error_ref().is_some()); - - let transient_error = PersistedError::( - InternalPersistedError::Api(ApiError::Transient(api_err.clone())), + assert!(!fatal_error.is_transient()); + assert!(fatal_error.is_fatal()); + assert_eq!(fatal_error.transient_state(), None); + + let fatal_with_state_error = PersistedError::( + InternalPersistedError::Api(ApiError::FatalWithState( + api_err.clone(), + "Error state".to_string(), + )), + ); + assert!(fatal_with_state_error.storage_error_ref().is_none()); + assert!(fatal_with_state_error.api_error_ref().is_some()); + assert!(!fatal_with_state_error.is_transient()); + assert!(fatal_with_state_error.is_fatal()); + assert_eq!(fatal_with_state_error.error_state(), Some("Error state".to_string())); + + let transient_error = PersistedError::( + InternalPersistedError::Api(ApiError::Transient( + api_err.clone(), + "Current state".to_string(), + )), ); assert!(transient_error.storage_error_ref().is_none()); assert!(transient_error.api_error_ref().is_some()); + assert!(transient_error.is_transient()); + assert!(!transient_error.is_fatal()); + assert_eq!(transient_error.transient_state(), Some("Current state".to_string())); } } diff --git a/payjoin/src/core/receive/v2/mod.rs b/payjoin/src/core/receive/v2/mod.rs index 205b3d09c..4ec00fb2e 100644 --- a/payjoin/src/core/receive/v2/mod.rs +++ b/payjoin/src/core/receive/v2/mod.rs @@ -594,7 +594,7 @@ impl Receiver { e, ); } else { - return MaybeFatalTransitionWithNoResults::transient(e); + return MaybeFatalTransitionWithNoResults::transient(e, current_state); }, _ => return MaybeFatalTransitionWithNoResults::fatal( @@ -769,6 +769,7 @@ impl Receiver { Receiver, Error, Receiver, + Self, > { match self.state.original.check_broadcast_suitability(min_fee_rate, can_broadcast) { Ok(()) => MaybeFatalTransition::success( @@ -779,7 +780,7 @@ impl Receiver { }, ), Err(Error::Implementation(e)) => - MaybeFatalTransition::transient(Error::Implementation(e)), + MaybeFatalTransition::transient(Error::Implementation(e), self), Err(e) => MaybeFatalTransition::replyable_error( SessionEvent::GotReplyableError((&e).into()), Receiver { @@ -849,6 +850,7 @@ impl Receiver { Receiver, Error, Receiver, + Self, > { match self.state.original.check_inputs_not_owned(is_owned) { Ok(()) => MaybeFatalTransition::success( @@ -859,7 +861,7 @@ impl Receiver { }, ), Err(e) => match e { - Error::Implementation(_) => MaybeFatalTransition::transient(e), + Error::Implementation(_) => MaybeFatalTransition::transient(e, self), _ => MaybeFatalTransition::replyable_error( SessionEvent::GotReplyableError((&e).into()), Receiver { @@ -909,6 +911,7 @@ impl Receiver { Receiver, Error, Receiver, + Self, > { match self.state.original.check_no_inputs_seen_before(is_known) { Ok(()) => MaybeFatalTransition::success( @@ -919,7 +922,7 @@ impl Receiver { }, ), Err(e) => match e { - Error::Implementation(_) => MaybeFatalTransition::transient(e), + Error::Implementation(_) => MaybeFatalTransition::transient(e, self), _ => MaybeFatalTransition::replyable_error( SessionEvent::GotReplyableError((&e).into()), Receiver { @@ -974,15 +977,16 @@ impl Receiver { Receiver, Error, Receiver, + Self, > { let fallback_tx = Some(self.state.fallback_tx()); - match self.state.original.identify_receiver_outputs(is_receiver_output) { + match self.state.original.clone().identify_receiver_outputs(is_receiver_output) { Ok(inner) => MaybeFatalTransition::success( SessionEvent::IdentifiedReceiverOutputs(inner.owned_vouts.clone()), Receiver { state: WantsOutputs { inner }, session_context: self.session_context }, ), Err(e) => match e { - Error::Implementation(_) => MaybeFatalTransition::transient(e), + Error::Implementation(_) => MaybeFatalTransition::transient(e, self), _ => MaybeFatalTransition::replyable_error( SessionEvent::GotReplyableError((&e).into()), Receiver { @@ -1207,12 +1211,14 @@ impl Receiver { self, min_fee_rate: Option, max_effective_fee_rate: Option, - ) -> MaybeFatalTransition, ProtocolError> { + ) -> MaybeFatalTransition, ProtocolError, (), Self> + { let max_effective_fee_rate = max_effective_fee_rate.or(Some(self.session_context.max_fee_rate)); match self .state .inner + .clone() .calculate_psbt_context_with_fee_range(min_fee_rate, max_effective_fee_rate) { Ok(psbt_context) => MaybeFatalTransition::success( @@ -1222,7 +1228,8 @@ impl Receiver { session_context: self.session_context, }, ), - Err(e) => MaybeFatalTransition::transient(ProtocolError::OriginalPayload(e.into())), + Err(e) => + MaybeFatalTransition::transient(ProtocolError::OriginalPayload(e.into()), self), } } @@ -1253,16 +1260,19 @@ impl Receiver { pub fn finalize_proposal( self, wallet_process_psbt: impl Fn(&Psbt) -> Result, - ) -> MaybeTransientTransition, ImplementationError> + ) -> MaybeTransientTransition, ImplementationError, Self> { - let original_psbt = self.state.psbt_context.original_psbt.clone(); - let payjoin_psbt = match self.state.psbt_context.finalize_proposal(wallet_process_psbt) { - Ok(payjoin_psbt) => payjoin_psbt, - Err(e) => { - return MaybeTransientTransition::transient(e); - } + let payjoin_psbt = + match self.state.psbt_context.clone().finalize_proposal(wallet_process_psbt) { + Ok(payjoin_psbt) => payjoin_psbt, + Err(e) => { + return MaybeTransientTransition::transient(e, self); + } + }; + let psbt_context = PsbtContext { + payjoin_psbt: payjoin_psbt.clone(), + original_psbt: self.state.psbt_context.original_psbt, }; - let psbt_context = PsbtContext { payjoin_psbt: payjoin_psbt.clone(), original_psbt }; let payjoin_proposal = PayjoinProposal { psbt_context: psbt_context.clone() }; MaybeTransientTransition::success( SessionEvent::FinalizedProposal(payjoin_psbt), @@ -1369,6 +1379,7 @@ impl Receiver { Receiver, ProtocolError, Receiver, + Self, > { match process_post_res(res, ohttp_context.into_inner()) { Ok(_) => MaybeFatalTransition::success( @@ -1389,9 +1400,10 @@ impl Receiver { ProtocolError::V2(InternalSessionError::DirectoryResponse(e).into()), ) } else { - MaybeFatalTransition::transient(ProtocolError::V2( - InternalSessionError::DirectoryResponse(e).into(), - )) + MaybeFatalTransition::transient( + ProtocolError::V2(InternalSessionError::DirectoryResponse(e).into()), + self, + ) }, } } @@ -1461,10 +1473,10 @@ impl Receiver { /// Process an OHTTP Encapsulated HTTP POST Error response /// to ensure it has been posted properly. pub fn process_error_response( - &self, + self, res: &[u8], ohttp_context: OhttpResponse, - ) -> MaybeTerminalSuccessTransition, ProtocolError> + ) -> MaybeTerminalSuccessTransition, ProtocolError, Self> { let pending = self.pending_fallback_after_protocol_failure(); let event = match &pending { @@ -1479,7 +1491,7 @@ impl Receiver { MaybeTerminalSuccessTransition::advance(event, pending_fallback), (Ok(_), None) => MaybeTerminalSuccessTransition::terminate(event), (Err(e), _) if !e.is_fatal() => - MaybeTerminalSuccessTransition::transient(protocol_error(e)), + MaybeTerminalSuccessTransition::transient(protocol_error(e), self), (Err(e), Some(pending_fallback)) => MaybeTerminalSuccessTransition::fatal_advance( event, pending_fallback, @@ -1548,9 +1560,12 @@ impl Receiver { Ok(Some(tx)) => { let tx_id = tx.compute_txid(); if tx_id != payjoin_txid { - return MaybeFatalOrSuccessTransition::transient(Error::Implementation( - ImplementationError::from(format!("Payjoin transaction ID mismatch. Expected: {payjoin_txid}, Got: {tx_id}").as_str()), - )); + return MaybeFatalOrSuccessTransition::transient( + Error::Implementation(ImplementationError::from( + format!("Payjoin transaction ID mismatch. Expected: {payjoin_txid}, Got: {tx_id}").as_str(), + )), + self.clone(), + ); } // TODO: should we check for witness and scriptsig on the tx? let mut sender_witnesses = vec![]; @@ -1566,7 +1581,11 @@ impl Receiver { )); } Ok(None) => {} - Err(e) => return MaybeFatalOrSuccessTransition::transient(Error::Implementation(e)), + Err(e) => + return MaybeFatalOrSuccessTransition::transient( + Error::Implementation(e), + self.clone(), + ), } // If the Payjoin proposal was not found, check the fallback transaction, as it is @@ -1577,7 +1596,11 @@ impl Receiver { SessionOutcome::FallbackBroadcasted, )), Ok(None) => {} - Err(e) => return MaybeFatalOrSuccessTransition::transient(Error::Implementation(e)), + Err(e) => + return MaybeFatalOrSuccessTransition::transient( + Error::Implementation(e), + self.clone(), + ), } MaybeFatalOrSuccessTransition::no_results(self.clone()) @@ -1859,6 +1882,7 @@ pub mod test { let receiver = v2::Receiver { state: unchecked_proposal, session_context: SHARED_CONTEXT.clone() }; + let expected_state = receiver.clone(); let unchecked_proposal = receiver.check_broadcast_suitability(Some(FeeRate::MIN), |_| { Err(ImplementationError::new(Error::Implementation("mock error".into()))) }); @@ -1866,10 +1890,14 @@ pub mod test { match unchecked_proposal { MaybeFatalTransition(Err(Rejection::Transient(RejectTransient( Error::Implementation(error), - )))) => assert_eq!( - error.to_string(), - Error::Implementation("mock error".into()).to_string() - ), + current_state, + )))) => { + assert_eq!( + error.to_string(), + Error::Implementation("mock error".into()).to_string() + ); + assert_eq!(current_state, expected_state); + } _ => panic!("Expected Implementation error"), } @@ -1904,6 +1932,7 @@ pub mod test { .assume_interactive_receiver() .save(&persister) .expect("Persister shouldn't fail"); + let expected_state = maybe_inputs_owned.clone(); let maybe_inputs_seen = maybe_inputs_owned.check_inputs_not_owned(&mut |_| { Err(ImplementationError::new(Error::Implementation("mock error".into()))) }); @@ -1911,10 +1940,14 @@ pub mod test { match maybe_inputs_seen { MaybeFatalTransition(Err(Rejection::Transient(RejectTransient( Error::Implementation(error), - )))) => assert_eq!( - error.to_string(), - Error::Implementation("mock error".into()).to_string() - ), + current_state, + )))) => { + assert_eq!( + error.to_string(), + Error::Implementation("mock error".into()).to_string() + ); + assert_eq!(current_state, expected_state); + } _ => panic!("Expected Implementation error"), } @@ -1936,16 +1969,21 @@ pub mod test { .check_inputs_not_owned(&mut |_| Ok(false)) .save(&persister) .expect("Persister shouldn't fail"); + let expected_state = maybe_inputs_seen.clone(); let outputs_unknown = maybe_inputs_seen.check_no_inputs_seen_before(&mut |_| { Err(ImplementationError::new(Error::Implementation("mock error".into()))) }); match outputs_unknown { MaybeFatalTransition(Err(Rejection::Transient(RejectTransient( Error::Implementation(error), - )))) => assert_eq!( - error.to_string(), - Error::Implementation("mock error".into()).to_string() - ), + current_state, + )))) => { + assert_eq!( + error.to_string(), + Error::Implementation("mock error".into()).to_string() + ); + assert_eq!(current_state, expected_state); + } _ => panic!("Expected Implementation error"), } @@ -1971,22 +2009,79 @@ pub mod test { .check_no_inputs_seen_before(&mut |_| Ok(false)) .save(&persister) .expect("Persister should not fail"); + let expected_state = outputs_unknown.clone(); let wants_outputs = outputs_unknown.identify_receiver_outputs(&mut |_| { Err(ImplementationError::new(Error::Implementation("mock error".into()))) }); match wants_outputs { MaybeFatalTransition(Err(Rejection::Transient(RejectTransient( Error::Implementation(error), - )))) => assert_eq!( - error.to_string(), - Error::Implementation("mock error".into()).to_string() - ), + current_state, + )))) => { + assert_eq!( + error.to_string(), + Error::Implementation("mock error".into()).to_string() + ); + assert_eq!(current_state, expected_state); + } _ => panic!("Expected Implementation error"), } Ok(()) } + #[test] + fn transient_error_can_be_retried_from_returned_state() -> Result<(), BoxError> { + let persister = InMemoryPersister::default(); + let receiver = receiver(unchecked_proposal_v2_from_test_vector()); + + let err = receiver + .check_broadcast_suitability(None, |_| Err("mock transient error".into())) + .save(&persister) + .expect_err("implementation failure should be transient"); + assert!(err.is_transient()); + assert!(!err.is_fatal()); + + let receiver = + err.transient_state().expect("transient error should return the current state"); + let _maybe_inputs_owned = receiver + .check_broadcast_suitability(None, |_| Ok(true)) + .save(&persister) + .expect("retry from the returned state should succeed"); + + // The event log is identical to that of a run that never failed. + assert_events(&persister, &[SessionEvent::CheckedBroadcastSuitability()], false); + Ok(()) + } + + #[test] + fn transient_state_matches_replay() -> Result<(), BoxError> { + let persister = InMemoryPersister::default(); + persister.save_event(SessionEvent::Created(SHARED_CONTEXT.clone()))?; + persister.save_event(SessionEvent::RetrievedOriginalPayload { + original: unchecked_proposal_v2_from_test_vector().original, + reply_key: None, + })?; + + let (session, _history) = replay_event_log(&persister)?; + let live_receiver = match session { + ReceiveSession::UncheckedOriginalPayload(receiver) => receiver, + other => panic!("Expected UncheckedOriginalPayload, got {other:?}"), + }; + + let err = live_receiver + .check_broadcast_suitability(None, |_| Err("mock transient error".into())) + .save(&persister) + .expect_err("implementation failure should be transient"); + let state_from_error = + err.transient_state().expect("transient error should return the current state"); + + // Nothing was persisted, so replaying yields the same state the error carries. + let (replayed, _history) = replay_event_log(&persister)?; + assert_eq!(replayed, ReceiveSession::UncheckedOriginalPayload(state_from_error)); + Ok(()) + } + #[test] fn test_create_error_request() -> Result<(), BoxError> { let mock_err = mock_err(); diff --git a/payjoin/src/core/send/v2/mod.rs b/payjoin/src/core/send/v2/mod.rs index 0c8491a86..403bc0fbb 100644 --- a/payjoin/src/core/send/v2/mod.rs +++ b/payjoin/src/core/send/v2/mod.rs @@ -392,7 +392,8 @@ impl Sender { self, response: &[u8], post_ctx: OhttpResponse, - ) -> MaybeFatalTransition, DecapsulationError> { + ) -> MaybeFatalTransition, DecapsulationError, (), Self> + { match process_post_res(response, post_ctx.into_inner()) { Ok(()) => {} Err(e) => @@ -404,6 +405,7 @@ impl Sender { } else { return MaybeFatalTransition::transient( InternalDecapsulationError::DirectoryResponse(e).into(), + self, ); }, } @@ -538,7 +540,7 @@ impl Sender { > { let body = match process_get_res(response, ohttp_ctx.into_inner()) { Ok(Some(body)) => body, - Ok(None) => return MaybeSuccessTransitionWithNoResults::no_results(self.clone()), + Ok(None) => return MaybeSuccessTransitionWithNoResults::no_results(self), Err(e) => if e.is_fatal() { return MaybeSuccessTransitionWithNoResults::fatal( @@ -548,6 +550,7 @@ impl Sender { } else { return MaybeSuccessTransitionWithNoResults::transient( InternalDecapsulationError::DirectoryResponse(e).into(), + self, ); }, };