From 7f41f46aa342228e21d8478564b193f42a4ed120 Mon Sep 17 00:00:00 2001 From: spacebear Date: Thu, 9 Jul 2026 18:30:11 -0400 Subject: [PATCH 1/4] Return current state in transient errors Transient state transitions consumed the typestate and returned only an error, so callers could not retry without cloning the state ahead of every call or replaying the whole session event log. Carry ownership of the current typestate inside the transient rejection, the way mpsc::SendError hands back the unsent value. This preserves typestate linearity: a transition yields the next state on success, hands the same state back on a transient error, or closes the session on a fatal error. Transient rejections persist nothing, so the returned state always matches what replay would reconstruct. RejectTransient, Rejection, ApiError, and PersistedError gain a CurrentState parameter defaulting to (). MaybeFatalTransition, MaybeTransientTransition, and MaybeTerminalSuccessTransition grow a trailing CurrentState parameter; the WithNoResults and FatalOrSuccess transitions reuse the CurrentState they already carry for the stasis arm. The new PersistedError::transient_state accessor recovers the state. process_error_response now consumes self like every other transition. It was the one borrowing transition, which left a live handle able to re-post and append a duplicate event after a successful save; consuming self extends the exactly-one-live-handle invariant to the error-reply path, with the receiver returned through the transient error like everywhere else. Sites that moved inner data into by-value calls before the transient return (identify_receiver_outputs, apply_fee_range, finalize_proposal) now clone that data up front, adding one PSBT-sized clone to their happy paths. --- payjoin-ffi/src/receive/error.rs | 5 +- payjoin-ffi/src/receive/mod.rs | 9 + payjoin-ffi/src/send/error.rs | 17 +- payjoin-ffi/src/send/mod.rs | 2 + payjoin/src/core/persist.rs | 447 +++++++++++++++++++++-------- payjoin/src/core/receive/v2/mod.rs | 129 ++++++--- payjoin/src/core/send/v2/mod.rs | 7 +- 7 files changed, 443 insertions(+), 173 deletions(-) diff --git a/payjoin-ffi/src/receive/error.rs b/payjoin-ffi/src/receive/error.rs index c3fbf8419..bed8e92a2 100644 --- a/payjoin-ffi/src/receive/error.rs +++ b/payjoin-ffi/src/receive/error.rs @@ -76,13 +76,14 @@ 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)); 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..f3991eb9a 100644 --- a/payjoin-ffi/src/send/error.rs +++ b/payjoin-ffi/src/send/error.rs @@ -202,12 +202,13 @@ impl From for SenderPersistedError { fn from(value: ImplementationError) -> Self { SenderPersistedError::Storage(Arc::new(value)) } } -impl From> +impl From> for SenderPersistedError where S: std::error::Error + Send + Sync + 'static, + C: std::fmt::Debug, { - fn from(err: payjoin::persist::PersistedError) -> Self { + 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)); @@ -221,11 +222,13 @@ where } } -impl From> for SenderPersistedError +impl From> + for SenderPersistedError where S: std::error::Error + Send + Sync + 'static, + C: std::fmt::Debug, { - fn from(err: payjoin::persist::PersistedError) -> Self { + 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)); @@ -239,11 +242,13 @@ where } } -impl From> for SenderPersistedError +impl From> + for SenderPersistedError where S: std::error::Error + Send + Sync + 'static, + C: std::fmt::Debug, { - fn from(err: payjoin::persist::PersistedError) -> Self { + 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)); diff --git a/payjoin-ffi/src/send/mod.rs b/payjoin-ffi/src/send/mod.rs index d528d1389..78b927b20 100644 --- a/payjoin-ffi/src/send/mod.rs +++ b/payjoin-ffi/src/send/mod.rs @@ -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..b1cfd8927 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,56 @@ 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, + } + } } -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 +890,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 +901,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 +1215,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 +1232,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 +1350,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 +1390,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 +1410,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 +1462,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 +1536,12 @@ mod tests { }), expected_result: ExpectedResult::< OptionalTransitionOutcome, - PersistedError, + PersistedError< + InMemoryTestError, + std::convert::Infallible, + (), + InMemoryTestState, + >, > { events: vec![], is_closed: false, @@ -1408,15 +1550,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 +1629,12 @@ mod tests { }), expected_result: ExpectedResult::< OptionalTransitionOutcome, - PersistedError, + PersistedError< + InMemoryTestError, + std::convert::Infallible, + (), + InMemoryTestState, + >, > { events: vec![], is_closed: false, @@ -1486,6 +1642,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 +1715,12 @@ mod tests { }), expected_result: ExpectedResult::< OptionalTransitionOutcome<(), InMemoryTestState>, - PersistedError, + PersistedError< + InMemoryTestError, + std::convert::Infallible, + (), + InMemoryTestState, + >, > { events: vec![], is_closed: false, @@ -1564,15 +1748,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, }, @@ -1600,10 +1793,24 @@ 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())), + 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_eq!(transient_error.transient_state(), Some("Current state".to_string())); + + let fatal_error = PersistedError::( + InternalPersistedError::Api(ApiError::Fatal(api_err.clone())), + ); + assert_eq!(fatal_error.transient_state(), None); + + let storage_error = PersistedError::( + InternalPersistedError::Storage(InMemoryTestError {}), + ); + assert_eq!(storage_error.transient_state(), None); } } diff --git a/payjoin/src/core/receive/v2/mod.rs b/payjoin/src/core/receive/v2/mod.rs index 205b3d09c..5d6ecadad 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,16 +2009,21 @@ 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"), } 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, ); }, }; From c7dc8d676d58006f9c063b5f8e100e1a4f19700e Mon Sep 17 00:00:00 2001 From: spacebear Date: Thu, 9 Jul 2026 18:35:55 -0400 Subject: [PATCH 2/4] Expose transient vs fatal on PersistedError Poll loops and retry logic need to branch on whether a persisted error is retryable, but PersistedError collapsed the distinction that the internal ApiError variants already encode (issue #1422). Add is_transient() and is_fatal() predicates. A transient error carries the state to resume from via transient_state(); a fatal error closed the session or moved it to an error state. Storage errors are neither: the transition outcome is unknown and recovery is replaying the event log. Cover the retry loop end to end: retrying a failed transition from the state returned by transient_state() produces an event log identical to a run that never failed, and the carried state equals what replaying the event log reconstructs. --- payjoin/src/core/persist.rs | 51 +++++++++++++++++++++++------ payjoin/src/core/receive/v2/mod.rs | 52 ++++++++++++++++++++++++++++++ 2 files changed, 93 insertions(+), 10 deletions(-) diff --git a/payjoin/src/core/persist.rs b/payjoin/src/core/persist.rs index b1cfd8927..21458b94b 100644 --- a/payjoin/src/core/persist.rs +++ b/payjoin/src/core/persist.rs @@ -844,6 +844,27 @@ where _ => 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< @@ -1785,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::( @@ -1792,6 +1816,21 @@ mod tests { ); assert!(fatal_error.storage_error_ref().is_none()); assert!(fatal_error.api_error_ref().is_some()); + 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( @@ -1801,16 +1840,8 @@ mod tests { ); 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())); - - let fatal_error = PersistedError::( - InternalPersistedError::Api(ApiError::Fatal(api_err.clone())), - ); - assert_eq!(fatal_error.transient_state(), None); - - let storage_error = PersistedError::( - InternalPersistedError::Storage(InMemoryTestError {}), - ); - assert_eq!(storage_error.transient_state(), None); } } diff --git a/payjoin/src/core/receive/v2/mod.rs b/payjoin/src/core/receive/v2/mod.rs index 5d6ecadad..4ec00fb2e 100644 --- a/payjoin/src/core/receive/v2/mod.rs +++ b/payjoin/src/core/receive/v2/mod.rs @@ -2030,6 +2030,58 @@ pub mod test { 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(); From 7f7003de9145198a0f0e0519eb212e38df224b90 Mon Sep 17 00:00:00 2001 From: spacebear Date: Thu, 9 Jul 2026 18:44:48 -0400 Subject: [PATCH 3/4] Classify FFI persisted errors as transient/fatal Bindings had no way to tell whether a failed transition is retryable: ReceiverPersistedError and SenderPersistedError flattened the distinction the core error model encodes (issue #1422). uniffi error enums cannot carry methods, so encode the classification in the variant structure, which surfaces as distinct exception types in each binding language. Both enums become Transient/Fatal/Storage, with the protocol error carried inside the first two. The sender's per-kind variants move into a new SenderError enum nested inside the classification, mirroring the receiver's ReceiverError, and its three From impls collapse into a macro mirroring the receiver's. The From impls pick the variant with PersistedError::is_transient. Rename the FFI builder error from BuildSenderError to SenderBuilderError, mirroring ReceiverBuilderError. Besides the symmetry, this avoids a name collision in the generated Dart: uniffi-dart flattens each error-enum variant to a top-level class named , so SenderError::Build produces BuildSenderException, the same class the BuildSenderError object generates once its Error suffix is rewritten to Exception. --- payjoin-ffi/src/receive/error.rs | 66 +++++++++++++-- payjoin-ffi/src/send/error.rs | 135 +++++++++++++++---------------- payjoin-ffi/src/send/mod.rs | 2 +- 3 files changed, 129 insertions(+), 74 deletions(-) diff --git a/payjoin-ffi/src/receive/error.rs b/payjoin-ffi/src/receive/error.rs index bed8e92a2..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), @@ -88,12 +92,18 @@ macro_rules! impl_persisted_error_from { 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) } } }; @@ -321,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/send/error.rs b/payjoin-ffi/src/send/error.rs index f3991eb9a..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,90 +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, - C: std::fmt::Debug, -{ - 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, - C: std::fmt::Debug, -{ - 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, - C: std::fmt::Debug, -{ - 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 78b927b20..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, }; From a7c000ea458d8caa03e93913165d587d10da90f7 Mon Sep 17 00:00:00 2001 From: DanGould Date: Fri, 10 Jul 2026 21:19:40 +0800 Subject: [PATCH 4/4] Handle transient errors in CLI send/receive loops Add transient arms to the two existing poll loops and convert the three one-shot POST flows into retry loops that recover the state with transient_state(). - get_proposed_payjoin_psbt: transient arm retries the poll in place. Drops the defensive sender.clone() that existed because the state was previously unrecoverable from errors. - long_poll_fallback: transient arm retries the poll in place. - send_payjoin_proposal: becomes a retry loop; re-POSTing the finalized proposal is idempotent from both Bitcoin and directory perspectives. - post_original_proposal: becomes a retry loop; re-POSTing the Original PSBT is idempotent from the directory's perspective. - handle_error: branches on is_transient() and retries the error reply, resolving the ambiguity between Transient and FatalTerminate outcomes (issue #1709). Transient retries are paced by a fixed 5 second delay so a misbehaving directory or relay is not hammered in a tight loop. No delay is needed on the happy path because the directory holds long polls open, and no retry cap is needed because the request constructors fail once the session expires, exiting each loop through the existing fatal arms. The receiver's proposal-checking chain intentionally keeps propagating transient errors instead of retrying in place: those come from wallet and database callbacks whose recovery path is resuming the session, and apply_fee_range's transient error is deterministic for a given proposal, so an in-place retry loop would spin until session expiry. Co-authored-by: spacebear --- payjoin-cli/src/app/v2/mod.rs | 154 +++++++++++++++++++++------------- 1 file changed, 97 insertions(+), 57 deletions(-) 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 {