From 1c003c99cc1f74085fb7def0eb005152d0e11fca Mon Sep 17 00:00:00 2001 From: Benalleng Date: Wed, 8 Jul 2026 09:29:00 -0400 Subject: [PATCH] Make persister no longer treat transient errors as an error Previously a transient error was treated as an error in the persister alongside fatal errors but unlike fatal errors transient errors were not giving the state at which it was reached. This commit returns the current state alongside the error while still retaining a NoOp in the persister with the expectation that the caller will want to try again. --- payjoin-cli/src/app/v2/mod.rs | 71 +++++++++++++-------- payjoin-ffi/src/receive/mod.rs | 60 +++++++++++++++--- payjoin/src/core/persist.rs | 99 ++++++++++++++++++++++-------- payjoin/src/core/receive/v2/mod.rs | 36 +++++++---- 4 files changed, 196 insertions(+), 70 deletions(-) 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(()) }