diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index 99d53e030..31129249c 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use anyhow::{anyhow, Context, Result}; use payjoin::bitcoin::consensus::encode::serialize_hex; use payjoin::bitcoin::{Amount, FeeRate}; -use payjoin::persist::{OptionalTransitionOutcome, SessionPersister}; +use payjoin::persist::{OptionalTransitionOutcome, ProcessedErrorOutcome, SessionPersister}; use payjoin::receive::v2::{ replay_event_log as replay_receiver_event_log, HasReplyableError, Initialized, MaybeInputsOwned, MaybeInputsSeen, Monitor, OutputsUnknown, PayjoinProposal, @@ -1014,37 +1014,56 @@ 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)), - }; + let session_id = persister.session_id(); + for retries in 0..3 { + 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(ProcessedErrorOutcome::Delivered(_)) => { + println!("Error reply delivered. Run `payjoin-cli cancel {session_id}` to broadcast the fallback transaction."); + return Ok(()); + } + Ok(ProcessedErrorOutcome::Terminated) => { + println!("Session {session_id} aborted. No fallback transaction available."); + return Ok(()); + } + Ok(ProcessedErrorOutcome::Transient(current_session, err)) if retries < 2 => { + tracing::warn!("Transient error posting error reply: {err}; retrying"); + session = current_session; + continue; + } + 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_ref() { + Some(_) => { + println!("Session {session_id} failed. Run `payjoin-cli cancel {session_id}` to cancel and broadcast the fallback transaction."); + return Ok(()); + } + None => return Err(e.into()), + } + } + _ => { + println!("Session {session_id} failed. Run `payjoin-cli cancel {session_id}` to cancel and broadcast the fallback transaction."); + return Err(anyhow!("Giving up after 3 transient errors posting error reply to session {session_id}")); } - None => return Err(anyhow!("Failed to process error response")), } } - Ok(()) } diff --git a/payjoin-ffi/src/receive/mod.rs b/payjoin-ffi/src/receive/mod.rs index 8c746ef54..ce9a4fd4d 100644 --- a/payjoin-ffi/src/receive/mod.rs +++ b/payjoin-ffi/src/receive/mod.rs @@ -1358,6 +1358,52 @@ impl From }, + /// Session terminated — closed successfully. + Terminated, + /// Transient error — nothing persisted. Retry from the returned state. + Transient { current_state: Arc, error: ReceiverPersistedError }, + /// Fatal error — session may have been persisted with error state to allow retry. + Fatal(ReceiverPersistedError), +} + +impl HasReplyableErrorTransitionOutcome { + #[allow(clippy::type_complexity)] + fn from_core( + value: Result< + payjoin::persist::ProcessedErrorOutcome< + payjoin::receive::v2::Receiver, + payjoin::receive::v2::Receiver, + payjoin::receive::ProtocolError, + >, + payjoin::persist::PersistedError< + payjoin::receive::ProtocolError, + crate::error::ForeignError, + payjoin::receive::v2::Receiver, + >, + >, + ) -> Self { + match value { + Ok(payjoin::persist::ProcessedErrorOutcome::Delivered(pending)) => + HasReplyableErrorTransitionOutcome::Delivered { + pending_fallback: Arc::new(pending.into()), + }, + Ok(payjoin::persist::ProcessedErrorOutcome::Terminated) => + HasReplyableErrorTransitionOutcome::Terminated, + Ok(payjoin::persist::ProcessedErrorOutcome::Transient(state, err)) => + HasReplyableErrorTransitionOutcome::Transient { + current_state: Arc::new(state.into()), + error: ReceiverPersistedError::from(ImplementationError::new(err)), + }, + Err(e) => HasReplyableErrorTransitionOutcome::Fatal(e.into()), + } + } +} + #[derive(uniffi::Object)] #[allow(clippy::type_complexity)] pub struct HasReplyableErrorTransition( @@ -1368,6 +1414,7 @@ pub struct HasReplyableErrorTransition( payjoin::receive::v2::SessionEvent, payjoin::receive::v2::Receiver, payjoin::receive::ProtocolError, + payjoin::receive::v2::Receiver, >, >, >, @@ -1379,29 +1426,28 @@ impl HasReplyableErrorTransition { pub fn save( &self, persister: Arc, - ) -> Result>, ReceiverPersistedError> { + ) -> HasReplyableErrorTransitionOutcome { let adapter = CallbackPersisterAdapter::new(persister); let mut inner = self.0.write().expect("Lock should not be poisoned"); let value = inner.take().expect("Already saved or moved"); - let pending_fallback = value.save(&adapter).map_err(ReceiverPersistedError::from)?; - Ok(pending_fallback.map(|pending_fallback| Arc::new(pending_fallback.into()))) + let outcome = value.save(&adapter); + HasReplyableErrorTransitionOutcome::from_core(outcome) } pub async fn save_async( &self, persister: Arc, - ) -> Result>, ReceiverPersistedError> { + ) -> HasReplyableErrorTransitionOutcome { let adapter = AsyncCallbackPersisterAdapter::new(persister); let value = { let mut inner = self.0.write().expect("Lock should not be poisoned"); inner.take().expect("Already saved or moved") }; - let pending_fallback = - value.save_async(&adapter).await.map_err(ReceiverPersistedError::from)?; - Ok(pending_fallback.map(|pending_fallback| Arc::new(pending_fallback.into()))) + let outcome = value.save_async(&adapter).await; + HasReplyableErrorTransitionOutcome::from_core(outcome) } } diff --git a/payjoin/src/core/persist.rs b/payjoin/src/core/persist.rs index 5961da6cb..03deb6bb0 100644 --- a/payjoin/src/core/persist.rs +++ b/payjoin/src/core/persist.rs @@ -452,15 +452,21 @@ impl MaybeTerminalTransition { /// Fatal outcomes still persist an event. When the fatal outcome advances, the /// saved event keeps the session live for replay while the caller receives the /// fatal protocol error. +/// +/// Transient outcomes do not return an error — they return the current state +/// via [`ProcessedErrorOutcome::Transient`] so the caller can retry without +/// re-fetching the session. #[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,32 +484,41 @@ where Self(MaybeTerminalSuccessOutcome::FatalTerminate(event, error)) } - pub(crate) fn transient(error: Err) -> Self { - Self(MaybeTerminalSuccessOutcome::Transient(error)) + /// Build a transient outcome. Nothing is persisted. The current state is + /// returned to the caller so it can retry from the same state. + pub(crate) fn transient(current_state: CurrentState, error: Err) -> Self { + Self(MaybeTerminalSuccessOutcome::Transient(current_state, error)) } #[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))), + (PersistActions::Save(event), Ok(ProcessedErrorOutcome::Delivered(next_state))), MaybeTerminalSuccessOutcome::Terminate(event) => - (PersistActions::SaveAndClose(event), Ok(None)), + (PersistActions::SaveAndClose(event), Ok(ProcessedErrorOutcome::Terminated)), + MaybeTerminalSuccessOutcome::Transient(current_state, error) => + (PersistActions::NoOp, Ok(ProcessedErrorOutcome::Transient(current_state, error))), MaybeTerminalSuccessOutcome::FatalAdvance(event, next_state, error) => (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))), } } + #[allow(clippy::type_complexity)] pub fn save

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

( self, persister: &P, - ) -> Result, PersistedError> + ) -> Result< + ProcessedErrorOutcome, + PersistedError, + > where P: AsyncSessionPersister, Err: Send, NextState: Send, + CurrentState: Send, Event: Send, { let (actions, outcome) = self.deconstruct(); @@ -651,12 +671,39 @@ enum MaybeTerminalOutcome { Terminate(Event), } -enum MaybeTerminalSuccessOutcome { +enum MaybeTerminalSuccessOutcome { Advance(AcceptNextState), Terminate(Event), FatalAdvance(Event, NextState, Err), FatalTerminate(Event, Err), - Transient(Err), + Transient(CurrentState, Err), +} + +/// The successful outcome of a [`MaybeTerminalSuccessTransition`]. +/// +/// The session either advanced to a new state, terminated (closed), or hit a +/// transient error and should be retried from the returned current state. +#[derive(Debug)] +pub enum ProcessedErrorOutcome { + /// Error reply was delivered. Session advanced to pending fallback. + Delivered(NextState), + /// Session terminated — closed successfully, no next state. + Terminated, + /// Transient error — nothing was persisted. Retry from the current state. + Transient(CurrentState, Err), +} + +impl PartialEq + for ProcessedErrorOutcome +{ + fn eq(&self, other: &Self) -> bool { + match (self, other) { + (Self::Delivered(a), Self::Delivered(b)) => a == b, + (Self::Terminated, Self::Terminated) => true, + (Self::Transient(a, e1), Self::Transient(b, e2)) => a == b && e1 == e2, + _ => false, + } + } } /// Wrapper that represents either a successful state transition or indicates no state change occurred @@ -738,6 +785,13 @@ where } } + pub fn error_state(self) -> Option { + match self.0 { + InternalPersistedError::Api(ApiError::FatalWithState(_, state)) => Some(state), + _ => None, + } + } + pub fn storage_error_ref(&self) -> Option<&StorageErr> { match &self.0 { InternalPersistedError::Storage(e) => Some(e), @@ -754,8 +808,8 @@ where } } - pub fn error_state(self) -> Option { - match self.0 { + pub fn error_state_ref(&self) -> Option<&ErrorState> { + match &self.0 { InternalPersistedError::Api(ApiError::FatalWithState(_, state)) => Some(state), _ => None, } @@ -1230,7 +1284,7 @@ mod tests { events: vec![event.clone()], is_closed: false, error: None, - success: Some(Some(next_state.clone())), + success: Some(ProcessedErrorOutcome::Delivered(next_state.clone())), }, }, TestCase { @@ -1246,7 +1300,7 @@ mod tests { events: vec![close_event.clone()], is_closed: true, error: None, - success: Some(None), + success: Some(ProcessedErrorOutcome::Terminated), }, }, TestCase { @@ -1295,16 +1349,13 @@ mod tests { InMemoryTestEvent, InMemoryTestState, InMemoryTestError, - >::transient(InMemoryTestError {}) + >::transient((), InMemoryTestError {}) }), expected_result: ExpectedResult { events: vec![], is_closed: false, - error: Some( - InternalPersistedError::Api(ApiError::Transient(InMemoryTestError {})) - .into(), - ), - success: None, + error: None, + success: Some(ProcessedErrorOutcome::Transient((), InMemoryTestError {})), }, }, ]; diff --git a/payjoin/src/core/receive/v2/mod.rs b/payjoin/src/core/receive/v2/mod.rs index 205b3d09c..c26195fb5 100644 --- a/payjoin/src/core/receive/v2/mod.rs +++ b/payjoin/src/core/receive/v2/mod.rs @@ -1464,8 +1464,12 @@ impl Receiver { &self, res: &[u8], ohttp_context: OhttpResponse, - ) -> MaybeTerminalSuccessTransition, ProtocolError> - { + ) -> MaybeTerminalSuccessTransition< + SessionEvent, + Receiver, + ProtocolError, + Receiver, + > { let pending = self.pending_fallback_after_protocol_failure(); let event = match &pending { Some(_) => SessionEvent::ProtocolFailed, @@ -1479,7 +1483,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(self.clone(), protocol_error(e)), (Err(e), Some(pending_fallback)) => MaybeTerminalSuccessTransition::fatal_advance( event, pending_fallback, @@ -1630,7 +1634,8 @@ pub mod test { use super::*; use crate::output_substitution::OutputSubstitution; use crate::persist::{ - InMemoryPersister, OptionalTransitionOutcome, RejectTransient, Rejection, SessionPersister, + InMemoryPersister, OptionalTransitionOutcome, ProcessedErrorOutcome, RejectTransient, + Rejection, SessionPersister, }; use crate::receive::optional_parameters::Params; use crate::receive::v2; @@ -2046,10 +2051,11 @@ pub mod test { let response = ohttp_response_for(&req.body, http::StatusCode::OK); let persister = InMemoryPersister::::default(); - let pending_fallback = receiver - .process_error_response(&response, ctx) - .save(&persister)? - .expect("pending fallback should be returned"); + let pending_fallback = + match receiver.process_error_response(&response, ctx).save(&persister)? { + ProcessedErrorOutcome::Delivered(pending) => pending, + other => panic!("expected Advance, got {other:?}"), + }; assert_eq!(pending_fallback.fallback_tx(), &expected_tx); assert_events(&persister, &[SessionEvent::ProtocolFailed], false); @@ -2063,9 +2069,9 @@ pub mod test { let response = ohttp_response_for(&req.body, http::StatusCode::OK); let persister = InMemoryPersister::::default(); - let pending_fallback = receiver.process_error_response(&response, ctx).save(&persister)?; + let outcome = receiver.process_error_response(&response, ctx).save(&persister)?; - assert!(pending_fallback.is_none()); + assert!(matches!(outcome, ProcessedErrorOutcome::Terminated)); assert_events(&persister, &[SessionEvent::Closed(SessionOutcome::Aborted)], true); Ok(()) } @@ -2121,12 +2127,16 @@ pub mod test { let response = ohttp_response_for(&req.body, http::StatusCode::INTERNAL_SERVER_ERROR); let persister = InMemoryPersister::::default(); - let err = receiver + let outcome = receiver .process_error_response(&response, ctx) .save(&persister) - .expect_err("transient response should error"); + .expect("transient response should succeed with current state"); - assert!(err.api_error_ref().is_some()); + match outcome { + ProcessedErrorOutcome::Transient(returned_session, _err) => + assert_eq!(returned_session, receiver), + other => panic!("expected Transient, got {other:?}"), + } assert_events(&persister, &[], false); Ok(()) }