From e734e5eefa85f70619ab720a354343d4b58d115c Mon Sep 17 00:00:00 2001 From: spacebear Date: Wed, 13 May 2026 18:42:49 -0400 Subject: [PATCH 01/10] Add HasFallbackTx marker trait for receiver states Introduce a sealed::FallbackTx trait with a non-Option fallback_tx() method, implemented inside the sealed module for the in-protocol receiver states whose contract includes a confirmed broadcastable fallback. Expose access through HasFallbackTx, a public marker trait that has no methods of its own and is implemented for any type satisfying sealed::FallbackTx via a blanket impl. External crates can bound on HasFallbackTx but cannot implement it, and the method itself is only callable from inside the receive::v2 module where the sealed trait is in scope. - UncheckedOriginalPayload is deliberately excluded; it holds the sender's Original PSBT but has not yet run check_broadcast_suitability, so the PSBT is not yet verified as broadcastable. - HasReplyableError is also excluded; it will gain an optional fallback field in a later commit and continues to model the absent-fallback case at runtime. To avoid naming conflicts in intermediate commits, the existing `fallback_tx() -> Option` implementation is renamed to `maybe_fallback_tx`. It is removed entirely in a later commit. --- payjoin/src/core/receive/v2/mod.rs | 90 ++++++++++++++++++++++++++---- 1 file changed, 78 insertions(+), 12 deletions(-) diff --git a/payjoin/src/core/receive/v2/mod.rs b/payjoin/src/core/receive/v2/mod.rs index 2c3f1bd5b..b85a60560 100644 --- a/payjoin/src/core/receive/v2/mod.rs +++ b/payjoin/src/core/receive/v2/mod.rs @@ -233,61 +233,61 @@ impl ReceiveSession { mod sealed { pub trait State { - fn fallback_tx(&self) -> Option { None } + fn maybe_fallback_tx(&self) -> Option { None } } impl State for super::Initialized {} impl State for super::UncheckedOriginalPayload { - fn fallback_tx(&self) -> Option { + fn maybe_fallback_tx(&self) -> Option { Some(self.original.psbt.clone().extract_tx_unchecked_fee_rate()) } } impl State for super::MaybeInputsOwned { - fn fallback_tx(&self) -> Option { + fn maybe_fallback_tx(&self) -> Option { Some(self.original.psbt.clone().extract_tx_unchecked_fee_rate()) } } impl State for super::MaybeInputsSeen { - fn fallback_tx(&self) -> Option { + fn maybe_fallback_tx(&self) -> Option { Some(self.original.psbt.clone().extract_tx_unchecked_fee_rate()) } } impl State for super::OutputsUnknown { - fn fallback_tx(&self) -> Option { + fn maybe_fallback_tx(&self) -> Option { Some(self.original.psbt.clone().extract_tx_unchecked_fee_rate()) } } impl State for super::WantsOutputs { - fn fallback_tx(&self) -> Option { + fn maybe_fallback_tx(&self) -> Option { Some(self.inner.original_psbt.clone().extract_tx_unchecked_fee_rate()) } } impl State for super::WantsInputs { - fn fallback_tx(&self) -> Option { + fn maybe_fallback_tx(&self) -> Option { Some(self.inner.original_psbt.clone().extract_tx_unchecked_fee_rate()) } } impl State for super::WantsFeeRange { - fn fallback_tx(&self) -> Option { + fn maybe_fallback_tx(&self) -> Option { Some(self.inner.original_psbt.clone().extract_tx_unchecked_fee_rate()) } } impl State for super::ProvisionalProposal { - fn fallback_tx(&self) -> Option { + fn maybe_fallback_tx(&self) -> Option { Some(self.psbt_context.original_psbt.clone().extract_tx_unchecked_fee_rate()) } } impl State for super::PayjoinProposal { - fn fallback_tx(&self) -> Option { + fn maybe_fallback_tx(&self) -> Option { Some(self.psbt_context.original_psbt.clone().extract_tx_unchecked_fee_rate()) } } @@ -295,10 +295,68 @@ mod sealed { impl State for super::HasReplyableError {} impl State for super::Monitor { - fn fallback_tx(&self) -> Option { + fn maybe_fallback_tx(&self) -> Option { Some(self.psbt_context.original_psbt.clone().extract_tx_unchecked_fee_rate()) } } + + pub trait FallbackTx: State { + fn fallback_tx(&self) -> bitcoin::Transaction; + } + + impl FallbackTx for super::MaybeInputsOwned { + fn fallback_tx(&self) -> bitcoin::Transaction { + self.original.psbt.clone().extract_tx_unchecked_fee_rate() + } + } + + impl FallbackTx for super::MaybeInputsSeen { + fn fallback_tx(&self) -> bitcoin::Transaction { + self.original.psbt.clone().extract_tx_unchecked_fee_rate() + } + } + + impl FallbackTx for super::OutputsUnknown { + fn fallback_tx(&self) -> bitcoin::Transaction { + self.original.psbt.clone().extract_tx_unchecked_fee_rate() + } + } + + impl FallbackTx for super::WantsOutputs { + fn fallback_tx(&self) -> bitcoin::Transaction { + self.inner.original_psbt.clone().extract_tx_unchecked_fee_rate() + } + } + + impl FallbackTx for super::WantsInputs { + fn fallback_tx(&self) -> bitcoin::Transaction { + self.inner.original_psbt.clone().extract_tx_unchecked_fee_rate() + } + } + + impl FallbackTx for super::WantsFeeRange { + fn fallback_tx(&self) -> bitcoin::Transaction { + self.inner.original_psbt.clone().extract_tx_unchecked_fee_rate() + } + } + + impl FallbackTx for super::ProvisionalProposal { + fn fallback_tx(&self) -> bitcoin::Transaction { + self.psbt_context.original_psbt.clone().extract_tx_unchecked_fee_rate() + } + } + + impl FallbackTx for super::PayjoinProposal { + fn fallback_tx(&self) -> bitcoin::Transaction { + self.psbt_context.original_psbt.clone().extract_tx_unchecked_fee_rate() + } + } + + impl FallbackTx for super::Monitor { + fn fallback_tx(&self) -> bitcoin::Transaction { + self.psbt_context.original_psbt.clone().extract_tx_unchecked_fee_rate() + } + } } /// Sealed trait for V2 receive session states. @@ -310,6 +368,14 @@ pub trait State: sealed::State {} impl State for S {} +/// Marker trait for receiver protocol states that hold a verified broadcastable +/// fallback transaction. +/// +/// This trait is sealed to prevent external implementations. +pub trait HasFallbackTx: sealed::FallbackTx {} + +impl HasFallbackTx for T {} + /// A higher-level receiver construct which will be taken through different states through the /// protocol workflow. /// @@ -349,7 +415,7 @@ impl Receiver { /// /// This is a terminal transition — the session cannot be used after cancellation. pub fn cancel(self) -> TerminalTransition> { - let fallback = self.state.fallback_tx(); + let fallback = self.state.maybe_fallback_tx(); TerminalTransition::new(SessionEvent::Closed(SessionOutcome::Cancel), fallback) } } From 060370628ba76dfe497b23ccb89fe8b4851a1777 Mon Sep 17 00:00:00 2001 From: spacebear Date: Wed, 13 May 2026 19:18:50 -0400 Subject: [PATCH 02/10] Add PendingFallback typestate and entry events PendingFallback represents a receiver session that was cancelled or hit a fatal protocol error, and has a fallback transaction available to broadcast. While the session sits in PendingFallback the implementer holds an obligation to broadcast, discard, or otherwise handle the fallback transaction (e.g. save it to wallet DB for later broadcasting). This state is preserved across restarts and session replays until the implemeter calls `close()`, indicating that the handoff of the fallback transaction is complete and no longer a payjoin concern. --- payjoin-cli/src/app/v2/mod.rs | 3 + payjoin-ffi/src/receive/mod.rs | 78 ++++++++++++++++++++++++++ payjoin/src/core/receive/v2/mod.rs | 66 ++++++++++++++++++++++ payjoin/src/core/receive/v2/session.rs | 76 ++++++++++++++++++++++++- 4 files changed, 222 insertions(+), 1 deletion(-) diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index c2558884c..29f0440dd 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -78,6 +78,7 @@ impl StatusText for ReceiveSession { ReceiveSession::HasReplyableError(_) => "Session failure, waiting to post error response", ReceiveSession::Monitor(_) => "Monitoring payjoin proposal", + ReceiveSession::PendingFallback(_) => "Pending fallback handling", ReceiveSession::Closed(session_outcome) => match session_outcome { ReceiverSessionOutcome::Failure => "Session failure", ReceiverSessionOutcome::Success(_) => "Session success, Payjoin proposal was broadcasted", @@ -690,6 +691,8 @@ impl App { self.handle_error(error, persister).await, ReceiveSession::Monitor(proposal) => self.monitor_payjoin_proposal(proposal, persister).await, + ReceiveSession::PendingFallback(_) => + return Err(anyhow!("PendingFallback session handling not yet implemented")), ReceiveSession::Closed(_) => return Err(anyhow!("Session closed")), } }; diff --git a/payjoin-ffi/src/receive/mod.rs b/payjoin-ffi/src/receive/mod.rs index 47d9833f0..3911ad542 100644 --- a/payjoin-ffi/src/receive/mod.rs +++ b/payjoin-ffi/src/receive/mod.rs @@ -198,6 +198,7 @@ pub enum ReceiveSession { PayjoinProposal { inner: Arc }, HasReplyableError { inner: Arc }, Monitor { inner: Arc }, + PendingFallback { inner: Arc }, Closed { inner: Arc }, } @@ -228,6 +229,8 @@ impl From for ReceiveSession { ReceiveSession::HasReplyableError(inner) => Self::HasReplyableError { inner: Arc::new(inner.into()) }, ReceiveSession::Monitor(inner) => Self::Monitor { inner: Arc::new(inner.into()) }, + ReceiveSession::PendingFallback(inner) => + Self::PendingFallback { inner: Arc::new(inner.into()) }, ReceiveSession::Closed(session_outcome) => Self::Closed { inner: Arc::new(session_outcome.into()) }, } @@ -1435,6 +1438,81 @@ impl Monitor { } } +#[derive(uniffi::Object)] +#[allow(clippy::type_complexity)] +pub struct PendingFallbackTransition( + Arc< + RwLock< + Option>, + >, + >, +); + +#[uniffi::export] +impl PendingFallbackTransition { + pub fn save( + &self, + persister: Arc, + ) -> Result<(), ReceiverPersistedError> { + 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"); + + value + .save(&adapter) + .map_err(|e| ReceiverPersistedError::from(ImplementationError::new(e)))?; + Ok(()) + } + + pub async fn save_async( + &self, + persister: Arc, + ) -> Result<(), ReceiverPersistedError> { + 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") + }; + + value + .save_async(&adapter) + .await + .map_err(|e| ReceiverPersistedError::from(ImplementationError::new(e)))?; + Ok(()) + } +} + +#[derive(Clone, uniffi::Object)] +pub struct PendingFallback( + Arc>, +); + +impl From> + for PendingFallback +{ + fn from(value: payjoin::receive::v2::Receiver) -> Self { + Self(Arc::new(value)) + } +} + +impl From + for payjoin::receive::v2::Receiver +{ + fn from(value: PendingFallback) -> Self { value.0.as_ref().clone() } +} + +#[uniffi::export] +impl PendingFallback { + pub fn fallback_tx(&self) -> Vec { + payjoin::bitcoin::consensus::encode::serialize(self.0.fallback_tx()) + } + + pub fn close(&self) -> PendingFallbackTransition { + PendingFallbackTransition(Arc::new(RwLock::new(Some(self.0.as_ref().clone().close())))) + } +} + /// Session persister that should save and load events as JSON strings. #[uniffi::export(with_foreign)] pub trait JsonReceiverSessionPersister: Send + Sync { diff --git a/payjoin/src/core/receive/v2/mod.rs b/payjoin/src/core/receive/v2/mod.rs index b85a60560..367dbf757 100644 --- a/payjoin/src/core/receive/v2/mod.rs +++ b/payjoin/src/core/receive/v2/mod.rs @@ -145,6 +145,7 @@ pub enum ReceiveSession { PayjoinProposal(Receiver), HasReplyableError(Receiver), Monitor(Receiver), + PendingFallback(Receiver), Closed(SessionOutcome), } @@ -198,6 +199,24 @@ impl ReceiveSession { (ReceiveSession::PayjoinProposal(state), SessionEvent::PostedPayjoinProposal()) => Ok(state.apply_payjoin_posted()), + (session, SessionEvent::Cancelled) => + try_pending_fallback(session, SessionOutcome::Cancel).map_err(|session| { + InternalReplayError::InvalidEvent( + Box::new(SessionEvent::Cancelled), + Some(session), + ) + .into() + }), + + (session, SessionEvent::ProtocolFailed) => + try_pending_fallback(session, SessionOutcome::Failure).map_err(|session| { + InternalReplayError::InvalidEvent( + Box::new(SessionEvent::ProtocolFailed), + Some(session), + ) + .into() + }), + (_, SessionEvent::Closed(session_outcome)) => Ok(ReceiveSession::Closed(session_outcome)), @@ -217,6 +236,7 @@ impl ReceiveSession { ReceiveSession::PayjoinProposal(r) => r.session_context, ReceiveSession::HasReplyableError(r) => r.session_context, ReceiveSession::Monitor(r) => r.session_context, + ReceiveSession::PendingFallback(r) => r.session_context, ReceiveSession::Closed(session_outcome) => return Ok(ReceiveSession::Closed(session_outcome)), }, @@ -231,6 +251,36 @@ impl ReceiveSession { } } +fn pending_fallback_from( + r: Receiver, + outcome: SessionOutcome, +) -> ReceiveSession { + let fallback_tx = r.state.fallback_tx(); + ReceiveSession::PendingFallback(Receiver { + state: PendingFallback { fallback_tx, outcome }, + session_context: r.session_context, + }) +} + +fn try_pending_fallback( + session: ReceiveSession, + outcome: SessionOutcome, +) -> Result> { + match session { + ReceiveSession::MaybeInputsOwned(receiver) => Ok(pending_fallback_from(receiver, outcome)), + ReceiveSession::MaybeInputsSeen(receiver) => Ok(pending_fallback_from(receiver, outcome)), + ReceiveSession::OutputsUnknown(receiver) => Ok(pending_fallback_from(receiver, outcome)), + ReceiveSession::WantsOutputs(receiver) => Ok(pending_fallback_from(receiver, outcome)), + ReceiveSession::WantsInputs(receiver) => Ok(pending_fallback_from(receiver, outcome)), + ReceiveSession::WantsFeeRange(receiver) => Ok(pending_fallback_from(receiver, outcome)), + ReceiveSession::ProvisionalProposal(receiver) => + Ok(pending_fallback_from(receiver, outcome)), + ReceiveSession::PayjoinProposal(receiver) => Ok(pending_fallback_from(receiver, outcome)), + ReceiveSession::Monitor(receiver) => Ok(pending_fallback_from(receiver, outcome)), + session => Err(Box::new(session)), + } +} + mod sealed { pub trait State { fn maybe_fallback_tx(&self) -> Option { None } @@ -300,6 +350,8 @@ mod sealed { } } + impl State for super::PendingFallback {} + pub trait FallbackTx: State { fn fallback_tx(&self) -> bitcoin::Transaction; } @@ -420,6 +472,20 @@ impl Receiver { } } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PendingFallback { + fallback_tx: bitcoin::Transaction, + outcome: SessionOutcome, +} + +impl Receiver { + pub fn fallback_tx(&self) -> &bitcoin::Transaction { &self.state.fallback_tx } + + pub fn close(self) -> TerminalTransition { + TerminalTransition::new(SessionEvent::Closed(self.state.outcome), ()) + } +} + #[derive(Debug, Clone)] pub struct ReceiverBuilder(SessionContext); diff --git a/payjoin/src/core/receive/v2/session.rs b/payjoin/src/core/receive/v2/session.rs index c8e6b5069..749a84a4d 100644 --- a/payjoin/src/core/receive/v2/session.rs +++ b/payjoin/src/core/receive/v2/session.rs @@ -166,6 +166,8 @@ impl SessionHistory { SessionOutcome::Failure | SessionOutcome::Cancel => SessionStatus::Failed, SessionOutcome::FallbackBroadcasted => SessionStatus::FallbackBroadcasted, }, + Some(SessionEvent::Cancelled | SessionEvent::ProtocolFailed) => + SessionStatus::PendingFallback, _ => SessionStatus::Active, } } @@ -180,6 +182,7 @@ pub enum SessionStatus { Failed, Completed, FallbackBroadcasted, + PendingFallback, } /// Represents a piece of information that the receiver has obtained from the session @@ -198,6 +201,8 @@ pub enum SessionEvent { FinalizedProposal(bitcoin::Psbt), GotReplyableError(JsonReply), PostedPayjoinProposal(), + Cancelled, + ProtocolFailed, Closed(SessionOutcome), } @@ -229,7 +234,8 @@ mod tests { use crate::receive::tests::original_from_test_vector; use crate::receive::v2::test::{mock_err, SHARED_CONTEXT}; use crate::receive::v2::{ - Initialized, MaybeInputsOwned, ProvisionalProposal, Receiver, UncheckedOriginalPayload, + Initialized, MaybeInputsOwned, PendingFallback, ProvisionalProposal, Receiver, + UncheckedOriginalPayload, }; use crate::receive::{InternalPayloadError, PayloadError}; @@ -299,6 +305,8 @@ mod tests { SessionEvent::AppliedFeeRange(provisional_proposal.state.psbt_context.clone()), SessionEvent::FinalizedProposal(payjoin_proposal.psbt().clone()), SessionEvent::GotReplyableError(mock_err()), + SessionEvent::Cancelled, + SessionEvent::ProtocolFailed, ]; for event in test_cases { @@ -523,6 +531,72 @@ mod tests { run_session_history_test_async(&test).await; } + #[tokio::test] + async fn replaying_cancelled_session_enters_pending_fallback() { + let session_context = SHARED_CONTEXT.clone(); + let original = original_from_test_vector(); + let reply_key = Some(crate::HpkeKeyPair::gen_keypair().1); + let expected_fallback = original.psbt.clone().extract_tx_unchecked_fee_rate(); + + let test = SessionHistoryTest { + events: vec![ + SessionEvent::Created(session_context.clone()), + SessionEvent::RetrievedOriginalPayload { + original: original.clone(), + reply_key: reply_key.clone(), + }, + SessionEvent::CheckedBroadcastSuitability(), + SessionEvent::Cancelled, + ], + expected_session_history: SessionHistoryExpectedOutcome { + fallback_tx: Some(expected_fallback.clone()), + expected_status: SessionStatus::PendingFallback, + }, + expected_receiver_state: ReceiveSession::PendingFallback(Receiver { + state: PendingFallback { + fallback_tx: expected_fallback, + outcome: SessionOutcome::Cancel, + }, + session_context: SessionContext { reply_key, ..session_context }, + }), + }; + run_session_history_test(&test); + run_session_history_test_async(&test).await; + } + + #[tokio::test] + async fn replaying_protocol_failed_session_enters_pending_fallback() { + let session_context = SHARED_CONTEXT.clone(); + let original = original_from_test_vector(); + let reply_key = Some(crate::HpkeKeyPair::gen_keypair().1); + let expected_fallback = original.psbt.clone().extract_tx_unchecked_fee_rate(); + + let test = SessionHistoryTest { + events: vec![ + SessionEvent::Created(session_context.clone()), + SessionEvent::RetrievedOriginalPayload { + original: original.clone(), + reply_key: reply_key.clone(), + }, + SessionEvent::CheckedBroadcastSuitability(), + SessionEvent::ProtocolFailed, + ], + expected_session_history: SessionHistoryExpectedOutcome { + fallback_tx: Some(expected_fallback.clone()), + expected_status: SessionStatus::PendingFallback, + }, + expected_receiver_state: ReceiveSession::PendingFallback(Receiver { + state: PendingFallback { + fallback_tx: expected_fallback, + outcome: SessionOutcome::Failure, + }, + session_context: SessionContext { reply_key, ..session_context }, + }), + }; + run_session_history_test(&test); + run_session_history_test_async(&test).await; + } + #[tokio::test] async fn test_contributed_inputs() { let persister = InMemoryPersister::::default(); From 33a19c50449720e01faeccbd1bf5afdfd61e18e9 Mon Sep 17 00:00:00 2001 From: spacebear Date: Wed, 13 May 2026 21:01:19 -0400 Subject: [PATCH 03/10] Carry optional fallback on HasReplyableError HasReplyableError represents a receiver session that hit a replyable error before reaching PendingFallback. The struct must model the runtime fact that some sources can hand it a verified broadcastable fallback and others cannot. Encoding the field as Option keeps that distinction at the type level without weakening the HasFallback trait contract. --- payjoin/src/core/receive/v2/mod.rs | 189 ++++++++++++++++++++++++----- 1 file changed, 161 insertions(+), 28 deletions(-) diff --git a/payjoin/src/core/receive/v2/mod.rs b/payjoin/src/core/receive/v2/mod.rs index 367dbf757..8fdd6ec0c 100644 --- a/payjoin/src/core/receive/v2/mod.rs +++ b/payjoin/src/core/receive/v2/mod.rs @@ -42,6 +42,7 @@ pub use session::{ #[cfg(target_arch = "wasm32")] use web_time::Duration; +use self::sealed::FallbackTx; use super::error::{Error, InputContributionError}; use super::{ common, InternalPayloadError, JsonReply, OutputSubstitutionError, ProtocolError, SelectionError, @@ -220,27 +221,42 @@ impl ReceiveSession { (_, SessionEvent::Closed(session_outcome)) => Ok(ReceiveSession::Closed(session_outcome)), - (session, SessionEvent::GotReplyableError(error)) => + (session, SessionEvent::GotReplyableError(error)) => { + let (session_context, fallback_tx) = match session { + ReceiveSession::Initialized(r) => (r.session_context, None), + ReceiveSession::UncheckedOriginalPayload(r) => (r.session_context, None), + ReceiveSession::MaybeInputsOwned(r) => + (r.session_context, Some(r.state.fallback_tx())), + ReceiveSession::MaybeInputsSeen(r) => + (r.session_context, Some(r.state.fallback_tx())), + ReceiveSession::OutputsUnknown(r) => + (r.session_context, Some(r.state.fallback_tx())), + ReceiveSession::WantsOutputs(r) => + (r.session_context, Some(r.state.fallback_tx())), + ReceiveSession::WantsInputs(r) => + (r.session_context, Some(r.state.fallback_tx())), + ReceiveSession::WantsFeeRange(r) => + (r.session_context, Some(r.state.fallback_tx())), + ReceiveSession::ProvisionalProposal(r) => + (r.session_context, Some(r.state.fallback_tx())), + ReceiveSession::PayjoinProposal(r) => + (r.session_context, Some(r.state.fallback_tx())), + ReceiveSession::HasReplyableError(r) => + (r.session_context, r.state.fallback_tx.clone()), + ReceiveSession::Monitor(r) => (r.session_context, Some(r.state.fallback_tx())), + ReceiveSession::PendingFallback(r) => { + let fallback_tx = r.fallback_tx().clone(); + (r.session_context, Some(fallback_tx)) + } + ReceiveSession::Closed(session_outcome) => + return Ok(ReceiveSession::Closed(session_outcome)), + }; + Ok(ReceiveSession::HasReplyableError(Receiver { - state: HasReplyableError { error_reply: error.clone() }, - session_context: match session { - ReceiveSession::Initialized(r) => r.session_context, - ReceiveSession::UncheckedOriginalPayload(r) => r.session_context, - ReceiveSession::MaybeInputsOwned(r) => r.session_context, - ReceiveSession::MaybeInputsSeen(r) => r.session_context, - ReceiveSession::OutputsUnknown(r) => r.session_context, - ReceiveSession::WantsOutputs(r) => r.session_context, - ReceiveSession::WantsInputs(r) => r.session_context, - ReceiveSession::WantsFeeRange(r) => r.session_context, - ReceiveSession::ProvisionalProposal(r) => r.session_context, - ReceiveSession::PayjoinProposal(r) => r.session_context, - ReceiveSession::HasReplyableError(r) => r.session_context, - ReceiveSession::Monitor(r) => r.session_context, - ReceiveSession::PendingFallback(r) => r.session_context, - ReceiveSession::Closed(session_outcome) => - return Ok(ReceiveSession::Closed(session_outcome)), - }, - })), + state: HasReplyableError { error_reply: error, fallback_tx }, + session_context, + })) + } (current_state, event) => Err(InternalReplayError::InvalidEvent( Box::new(event), @@ -342,7 +358,9 @@ mod sealed { } } - impl State for super::HasReplyableError {} + impl State for super::HasReplyableError { + fn maybe_fallback_tx(&self) -> Option { self.fallback_tx.clone() } + } impl State for super::Monitor { fn maybe_fallback_tx(&self) -> Option { @@ -785,7 +803,7 @@ impl Receiver { Err(e) => MaybeFatalTransition::replyable_error( SessionEvent::GotReplyableError((&e).into()), Receiver { - state: HasReplyableError { error_reply: (&e).into() }, + state: HasReplyableError { error_reply: (&e).into(), fallback_tx: None }, session_context: self.session_context, }, e, @@ -865,7 +883,10 @@ impl Receiver { _ => MaybeFatalTransition::replyable_error( SessionEvent::GotReplyableError((&e).into()), Receiver { - state: HasReplyableError { error_reply: (&e).into() }, + state: HasReplyableError { + error_reply: (&e).into(), + fallback_tx: Some(self.state.fallback_tx()), + }, session_context: self.session_context, }, e, @@ -922,7 +943,10 @@ impl Receiver { _ => MaybeFatalTransition::replyable_error( SessionEvent::GotReplyableError((&e).into()), Receiver { - state: HasReplyableError { error_reply: (&e).into() }, + state: HasReplyableError { + error_reply: (&e).into(), + fallback_tx: Some(self.state.fallback_tx()), + }, session_context: self.session_context, }, e, @@ -971,6 +995,7 @@ impl Receiver { Error, Receiver, > { + let fallback_tx = Some(self.state.fallback_tx()); match self.state.original.identify_receiver_outputs(is_receiver_output) { Ok(inner) => MaybeFatalTransition::success( SessionEvent::IdentifiedReceiverOutputs(inner.owned_vouts.clone()), @@ -981,7 +1006,7 @@ impl Receiver { _ => MaybeFatalTransition::replyable_error( SessionEvent::GotReplyableError((&e).into()), Receiver { - state: HasReplyableError { error_reply: (&e).into() }, + state: HasReplyableError { error_reply: (&e).into(), fallback_tx }, session_context: self.session_context, }, e, @@ -1349,6 +1374,7 @@ impl Receiver { #[derive(Debug, Clone, PartialEq)] pub struct HasReplyableError { error_reply: JsonReply, + fallback_tx: Option, } impl Receiver { @@ -1603,6 +1629,10 @@ pub mod test { JsonReply::from(&res) } + pub(crate) fn mock_fallback_tx() -> bitcoin::Transaction { + PARSED_ORIGINAL_PSBT.clone().extract_tx_unchecked_fee_rate() + } + #[test] fn test_monitor_typestate() -> Result<(), BoxError> { let psbt_ctx = PsbtContext { @@ -1879,7 +1909,10 @@ pub mod test { assert_eq!(mock_err.to_json(), expected_json); let receiver = Receiver { - state: HasReplyableError { error_reply: mock_err.clone() }, + state: HasReplyableError { + error_reply: mock_err.clone(), + fallback_tx: Some(mock_fallback_tx()), + }, session_context: SHARED_CONTEXT.clone(), }; @@ -1893,7 +1926,10 @@ pub mod test { let now = crate::time::Time::now(); let context = SessionContext { expiration: now, ..SHARED_CONTEXT.clone() }; let receiver = Receiver { - state: HasReplyableError { error_reply: mock_err() }, + state: HasReplyableError { + error_reply: mock_err(), + fallback_tx: Some(mock_fallback_tx()), + }, session_context: context.clone(), }; @@ -2036,9 +2072,17 @@ pub mod test { // States without a fallback transaction do_cancel_test!(Initialized {}, None); - do_cancel_test!(HasReplyableError { error_reply: mock_err() }, None); + do_cancel_test!(HasReplyableError { error_reply: mock_err(), fallback_tx: None }, None); // States with a fallback transaction + let error_fallback = mock_fallback_tx(); + do_cancel_test!( + HasReplyableError { + error_reply: mock_err(), + fallback_tx: Some(error_fallback.clone()) + }, + Some(error_fallback) + ); do_cancel_test!( UncheckedOriginalPayload { original: original.clone() }, Some(expected_tx.clone()) @@ -2059,4 +2103,93 @@ pub mod test { ); do_cancel_test!(Monitor { psbt_context: psbt_ctx }, Some(expected_tx)); } + + #[test] + fn replaying_replyable_error_from_unchecked_captures_no_fallback() { + let state = unchecked_proposal_v2_from_test_vector(); + let error = mock_err(); + let session = ReceiveSession::UncheckedOriginalPayload(Receiver { + state, + session_context: SHARED_CONTEXT.clone(), + }); + + let replayed = session + .process_event(SessionEvent::GotReplyableError(error.clone())) + .expect("replyable error should replay"); + + match replayed { + ReceiveSession::HasReplyableError(receiver) => { + assert_eq!(receiver.state.error_reply, error); + assert_eq!(receiver.state.fallback_tx, None); + } + other => panic!("Expected HasReplyableError, got {other:?}"), + } + } + + #[test] + fn replaying_replyable_error_from_initialized_captures_no_fallback() { + let error = mock_err(); + let session = ReceiveSession::Initialized(Receiver { + state: Initialized {}, + session_context: SHARED_CONTEXT.clone(), + }); + + let replayed = session + .process_event(SessionEvent::GotReplyableError(error.clone())) + .expect("replyable error should replay"); + + match replayed { + ReceiveSession::HasReplyableError(receiver) => { + assert_eq!(receiver.state.error_reply, error); + assert_eq!(receiver.state.fallback_tx, None); + } + other => panic!("Expected HasReplyableError, got {other:?}"), + } + } + + #[test] + fn replaying_replyable_error_from_replyable_error_carries_some_fallback() { + let expected_fallback = mock_fallback_tx(); + let error = mock_err(); + let session = ReceiveSession::HasReplyableError(Receiver { + state: HasReplyableError { + error_reply: mock_err(), + fallback_tx: Some(expected_fallback.clone()), + }, + session_context: SHARED_CONTEXT.clone(), + }); + + let replayed = session + .process_event(SessionEvent::GotReplyableError(error.clone())) + .expect("replyable error should replay"); + + match replayed { + ReceiveSession::HasReplyableError(receiver) => { + assert_eq!(receiver.state.error_reply, error); + assert_eq!(receiver.state.fallback_tx, Some(expected_fallback)); + } + other => panic!("Expected HasReplyableError, got {other:?}"), + } + } + + #[test] + fn replaying_replyable_error_from_replyable_error_carries_no_fallback() { + let error = mock_err(); + let session = ReceiveSession::HasReplyableError(Receiver { + state: HasReplyableError { error_reply: mock_err(), fallback_tx: None }, + session_context: SHARED_CONTEXT.clone(), + }); + + let replayed = session + .process_event(SessionEvent::GotReplyableError(error.clone())) + .expect("replyable error should replay"); + + match replayed { + ReceiveSession::HasReplyableError(receiver) => { + assert_eq!(receiver.state.error_reply, error); + assert_eq!(receiver.state.fallback_tx, None); + } + other => panic!("Expected HasReplyableError, got {other:?}"), + } + } } From b01c7ad1b6e48e98375cf19790cbdf678126451f Mon Sep 17 00:00:00 2001 From: spacebear Date: Wed, 13 May 2026 23:01:36 -0400 Subject: [PATCH 04/10] Add MaybeTerminal transition types for fork-on-state APIs Introduce MaybeTerminalTransition for the no-error fork (used by cancel) and MaybeTerminalSuccessTransition for the error-bearing fork (used by process_error_response). Both expose advance and terminate constructors that map to Save and SaveAndClose actions respectively. The success variant returns Option; the error variants preserve the caller's distinction between transient, fatal-advance, and fatal-terminate. --- payjoin/src/core/persist.rs | 313 ++++++++++++++++++++++++++++++++++-- 1 file changed, 303 insertions(+), 10 deletions(-) diff --git a/payjoin/src/core/persist.rs b/payjoin/src/core/persist.rs index c706a1367..81f89eb0b 100644 --- a/payjoin/src/core/persist.rs +++ b/payjoin/src/core/persist.rs @@ -265,6 +265,11 @@ where MaybeFatalTransition(Err(Rejection::replyable_error(event, error_state, error))) } + #[allow(dead_code)] + pub(crate) fn fatal_with_state(event: Event, error_state: ErrorState, error: Err) -> Self { + MaybeFatalTransition(Err(Rejection::replyable_error(event, error_state, error))) + } + pub(crate) fn deconstruct( self, ) -> (PersistActions, Result>) { @@ -361,6 +366,40 @@ where } } +/// A transition that always results in a state transition. +pub struct NextStateTransition(AcceptNextState); + +impl NextStateTransition { + pub(crate) fn success(event: Event, next_state: NextState) -> Self { + NextStateTransition(AcceptNextState(event, next_state)) + } + + pub(crate) fn deconstruct(self) -> (PersistActions, NextState) { + let AcceptNextState(event, next_state) = self.0; + (PersistActions::Save(event), next_state) + } + + pub fn save

(self, persister: &P) -> Result + where + P: SessionPersister, + { + let (actions, next_state) = self.deconstruct(); + actions.execute(persister)?; + Ok(next_state) + } + + pub async fn save_async

(self, persister: &P) -> Result + where + P: AsyncSessionPersister, + NextState: Send, + Event: Send, + { + let (actions, next_state) = self.deconstruct(); + actions.execute_async(persister).await?; + Ok(next_state) + } +} + /// A transition that can result in the completion of a state machine or a transient error /// Fatal errors cannot occur in this transition. pub struct MaybeSuccessTransition( @@ -371,14 +410,17 @@ impl MaybeSuccessTransition where Err: std::error::Error, { + #[allow(dead_code)] pub(crate) fn success(event: Event, success_value: SuccessValue) -> Self { MaybeSuccessTransition(Ok(AcceptNextState(event, success_value))) } + #[allow(dead_code)] pub(crate) fn transient(error: Err) -> Self { MaybeSuccessTransition(Err(Rejection::transient(error))) } + #[allow(dead_code)] pub(crate) fn fatal(event: Event, error: Err) -> Self { MaybeSuccessTransition(Err(Rejection::fatal(event, error))) } @@ -426,20 +468,30 @@ where } } -/// A transition that always results in a state transition. -pub struct NextStateTransition(AcceptNextState); +/// A transition that either advances to a live state or terminates the session. +/// +/// No error path exists. Both outcomes are successful from the protocol's point +/// of view. The choice is determined by the source typestate's internal data, +/// not by the caller. +pub struct MaybeTerminalTransition(MaybeTerminalOutcome); -impl NextStateTransition { - pub(crate) fn success(event: Event, next_state: NextState) -> Self { - NextStateTransition(AcceptNextState(event, next_state)) +#[allow(dead_code)] +impl MaybeTerminalTransition { + pub(crate) fn advance(event: Event, next_state: NextState) -> Self { + Self(MaybeTerminalOutcome::Advance(AcceptNextState(event, next_state))) } - pub(crate) fn deconstruct(self) -> (PersistActions, NextState) { - let AcceptNextState(event, next_state) = self.0; - (PersistActions::Save(event), next_state) + pub(crate) fn terminate(event: Event) -> Self { Self(MaybeTerminalOutcome::Terminate(event)) } + + pub(crate) fn deconstruct(self) -> (PersistActions, Option) { + match self.0 { + MaybeTerminalOutcome::Advance(AcceptNextState(event, next_state)) => + (PersistActions::Save(event), Some(next_state)), + MaybeTerminalOutcome::Terminate(event) => (PersistActions::SaveAndClose(event), None), + } } - pub fn save

(self, persister: &P) -> Result + pub fn save

(self, persister: &P) -> Result, P::InternalStorageError> where P: SessionPersister, { @@ -448,7 +500,10 @@ impl NextStateTransition { Ok(next_state) } - pub async fn save_async

(self, persister: &P) -> Result + pub async fn save_async

( + self, + persister: &P, + ) -> Result, P::InternalStorageError> where P: AsyncSessionPersister, NextState: Send, @@ -460,6 +515,85 @@ impl NextStateTransition { } } +/// A transition that can either advance, terminate, or fail transiently. +/// +/// 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. +pub struct MaybeTerminalSuccessTransition( + MaybeTerminalSuccessOutcome, +); + +#[allow(dead_code)] +impl MaybeTerminalSuccessTransition +where + Err: std::error::Error, +{ + pub(crate) fn advance(event: Event, next_state: NextState) -> Self { + Self(MaybeTerminalSuccessOutcome::Advance(AcceptNextState(event, next_state))) + } + + pub(crate) fn terminate(event: Event) -> Self { + Self(MaybeTerminalSuccessOutcome::Terminate(event)) + } + + pub(crate) fn fatal_advance(event: Event, next_state: NextState, error: Err) -> Self { + Self(MaybeTerminalSuccessOutcome::FatalAdvance(event, next_state, error)) + } + + pub(crate) fn fatal_terminate(event: Event, error: Err) -> Self { + Self(MaybeTerminalSuccessOutcome::FatalTerminate(event, error)) + } + + pub(crate) fn transient(error: Err) -> Self { + Self(MaybeTerminalSuccessOutcome::Transient(error)) + } + + pub(crate) fn deconstruct( + self, + ) -> (PersistActions, Result, ApiError>) { + match self.0 { + MaybeTerminalSuccessOutcome::Advance(AcceptNextState(event, next_state)) => + (PersistActions::Save(event), Ok(Some(next_state))), + MaybeTerminalSuccessOutcome::Terminate(event) => + (PersistActions::SaveAndClose(event), Ok(None)), + MaybeTerminalSuccessOutcome::FatalAdvance(event, _next_state, error) => + (PersistActions::Save(event), Err(ApiError::Fatal(error))), + MaybeTerminalSuccessOutcome::FatalTerminate(event, error) => + (PersistActions::SaveAndClose(event), Err(ApiError::Fatal(error))), + MaybeTerminalSuccessOutcome::Transient(error) => + (PersistActions::NoOp, Err(ApiError::Transient(error))), + } + } + + pub fn save

( + self, + persister: &P, + ) -> Result, PersistedError> + where + P: SessionPersister, + { + let (actions, outcome) = self.deconstruct(); + actions.execute(persister).map_err(InternalPersistedError::Storage)?; + Ok(outcome.map_err(InternalPersistedError::Api)?) + } + + pub async fn save_async

( + self, + persister: &P, + ) -> Result, PersistedError> + where + P: AsyncSessionPersister, + Err: Send, + NextState: Send, + Event: Send, + { + let (actions, outcome) = self.deconstruct(); + actions.execute_async(persister).await.map_err(InternalPersistedError::Storage)?; + Ok(outcome.map_err(InternalPersistedError::Api)?) + } +} + /// A transition that unconditionally terminates the session. /// /// Unlike other transition types, this always succeeds at the protocol level @@ -576,6 +710,21 @@ where /// Wrapper that marks the progression of a state machine pub struct AcceptNextState(Event, NextState); +#[allow(dead_code)] +enum MaybeTerminalOutcome { + Advance(AcceptNextState), + Terminate(Event), +} + +#[allow(dead_code)] +enum MaybeTerminalSuccessOutcome { + Advance(AcceptNextState), + Terminate(Event), + FatalAdvance(Event, NextState, Err), + FatalTerminate(Event, Err), + Transient(Err), +} + /// Wrapper that represents either a successful state transition or indicates no state change occurred pub enum AcceptOptionalTransition { /// A state transition that was successful and returned session event to be persisted @@ -1085,6 +1234,150 @@ mod tests { run_test_cases!(test_cases); } + #[tokio::test] + async fn test_maybe_terminal_transition() { + let event = InMemoryTestEvent("foo".to_string()); + let close_event = InMemoryTestEvent("close".to_string()); + let next_state = "Next state".to_string(); + + let test_cases = vec![ + TestCase { + make_transition: Box::new({ + let event = event.clone(); + let next_state = next_state.clone(); + move || MaybeTerminalTransition::advance(event.clone(), next_state.clone()) + }), + expected_result: ExpectedResult { + events: vec![event.clone()], + is_closed: false, + error: None, + success: Some(Some(next_state.clone())), + }, + }, + TestCase { + make_transition: Box::new({ + let close_event = close_event.clone(); + move || { + MaybeTerminalTransition::<_, InMemoryTestState>::terminate( + close_event.clone(), + ) + } + }), + expected_result: ExpectedResult { + events: vec![close_event.clone()], + is_closed: true, + error: None, + success: Some(None), + }, + }, + ]; + + run_test_cases!(test_cases); + } + + #[tokio::test] + async fn test_maybe_terminal_success_transition() { + let event = InMemoryTestEvent("foo".to_string()); + let close_event = InMemoryTestEvent("close".to_string()); + let fatal_event = InMemoryTestEvent("fatal".to_string()); + let fatal_close_event = InMemoryTestEvent("fatal close".to_string()); + let next_state = "Next state".to_string(); + + let test_cases = vec![ + TestCase { + make_transition: Box::new({ + let event = event.clone(); + let next_state = next_state.clone(); + move || { + MaybeTerminalSuccessTransition::advance(event.clone(), next_state.clone()) + } + }), + expected_result: ExpectedResult { + events: vec![event.clone()], + is_closed: false, + error: None, + success: Some(Some(next_state.clone())), + }, + }, + TestCase { + make_transition: Box::new({ + let close_event = close_event.clone(); + move || { + MaybeTerminalSuccessTransition::<_, InMemoryTestState, InMemoryTestError>::terminate( + close_event.clone(), + ) + } + }), + expected_result: ExpectedResult { + events: vec![close_event.clone()], + is_closed: true, + error: None, + success: Some(None), + }, + }, + TestCase { + make_transition: Box::new({ + let fatal_event = fatal_event.clone(); + let next_state = next_state.clone(); + move || { + MaybeTerminalSuccessTransition::fatal_advance( + fatal_event.clone(), + next_state.clone(), + InMemoryTestError {}, + ) + } + }), + expected_result: ExpectedResult { + events: vec![fatal_event.clone()], + is_closed: false, + error: Some( + InternalPersistedError::Api(ApiError::Fatal(InMemoryTestError {})).into(), + ), + success: None, + }, + }, + TestCase { + make_transition: Box::new({ + let fatal_close_event = fatal_close_event.clone(); + move || { + MaybeTerminalSuccessTransition::<_, InMemoryTestState, InMemoryTestError>::fatal_terminate( + fatal_close_event.clone(), + InMemoryTestError {}, + ) + } + }), + expected_result: ExpectedResult { + events: vec![fatal_close_event.clone()], + is_closed: true, + error: Some( + InternalPersistedError::Api(ApiError::Fatal(InMemoryTestError {})).into(), + ), + success: None, + }, + }, + TestCase { + make_transition: Box::new(|| { + MaybeTerminalSuccessTransition::< + InMemoryTestEvent, + InMemoryTestState, + InMemoryTestError, + >::transient(InMemoryTestError {}) + }), + expected_result: ExpectedResult { + events: vec![], + is_closed: false, + error: Some( + InternalPersistedError::Api(ApiError::Transient(InMemoryTestError {})) + .into(), + ), + success: None, + }, + }, + ]; + + run_test_cases!(test_cases); + } + #[tokio::test] async fn test_maybe_success_transition() { let event = InMemoryTestEvent("foo".to_string()); From 306e958dc1f9ffec1e14f1e0000519e0dbf492c8 Mon Sep 17 00:00:00 2001 From: spacebear Date: Wed, 13 May 2026 23:13:28 -0400 Subject: [PATCH 05/10] Route receiver cancel and fatal errors through PendingFallback The receiver side of v2 had a single blanket cancel implementation that always terminated the session and handed the wallet an Option. Fatal protocol errors emitted Closed(Failure) directly. Both shapes lost the wallet's obligation to broadcast the original transaction across a restart whenever a fallback existed. Replace the blanket cancel with typestate-aware impls: - impl Receiver::cancel advances to PendingFallback - Receiver::cancel and Receiver ::cancel terminate with Closed(Cancel); neither holds a verified fallback - Receiver::cancel forks on the optional fallback: Some advances to PendingFallback, None terminates with Closed(Cancel) --- payjoin-ffi/csharp/UnitTests.cs | 4 +- .../dart/test/test_payjoin_unit_test.dart | 8 +- payjoin-ffi/javascript/test/unit.test.ts | 8 +- .../python/test/test_payjoin_unit_test.py | 4 +- payjoin-ffi/src/receive/mod.rs | 187 ++++-- payjoin-ffi/src/send/mod.rs | 16 +- payjoin/src/core/receive/v2/mod.rs | 575 +++++++++++++----- payjoin/tests/integration.rs | 13 +- 8 files changed, 579 insertions(+), 236 deletions(-) diff --git a/payjoin-ffi/csharp/UnitTests.cs b/payjoin-ffi/csharp/UnitTests.cs index c4a456513..655286fa5 100644 --- a/payjoin-ffi/csharp/UnitTests.cs +++ b/payjoin-ffi/csharp/UnitTests.cs @@ -216,7 +216,7 @@ public void SenderCancelFromWithReplyKey() Assert.NotEmpty(pendingFallback.FallbackTx()); var cancelledResult = PayjoinMethods.ReplaySenderEventLog(senderPersister); - Assert.IsType(cancelledResult.State()); + Assert.IsType(cancelledResult.State()); pendingFallback.Close().Save(senderPersister); var closedResult = PayjoinMethods.ReplaySenderEventLog(senderPersister); @@ -246,7 +246,7 @@ public async Task SenderCancelFromWithReplyKeyAsync() Assert.NotEmpty(pendingFallback.FallbackTx()); var cancelledResult = await PayjoinMethods.ReplaySenderEventLogAsync(senderPersister); - Assert.IsType(cancelledResult.State()); + Assert.IsType(cancelledResult.State()); await pendingFallback.Close().SaveAsync(senderPersister); var closedResult = await PayjoinMethods.ReplaySenderEventLogAsync(senderPersister); diff --git a/payjoin-ffi/dart/test/test_payjoin_unit_test.dart b/payjoin-ffi/dart/test/test_payjoin_unit_test.dart index 95c4cc0ca..2cbb806c7 100644 --- a/payjoin-ffi/dart/test/test_payjoin_unit_test.dart +++ b/payjoin-ffi/dart/test/test_payjoin_unit_test.dart @@ -191,8 +191,8 @@ void main() { ); expect( cancelledResult.state(), - isA(), - reason: "sender should be in PendingFallback state after cancel", + isA(), + reason: "sender should be in SenderPendingFallback state after cancel", ); pendingFallback.close().save(persister: sender_persister); final closedResult = payjoin.replaySenderEventLog( @@ -236,8 +236,8 @@ void main() { ); expect( cancelledResult.state(), - isA(), - reason: "sender should be in PendingFallback state after cancel", + isA(), + reason: "sender should be in SenderPendingFallback state after cancel", ); await pendingFallback.close().saveAsync(persister: sender_persister); final closedResult = await payjoin.replaySenderEventLogAsync( diff --git a/payjoin-ffi/javascript/test/unit.test.ts b/payjoin-ffi/javascript/test/unit.test.ts index a88b5882c..89386421e 100644 --- a/payjoin-ffi/javascript/test/unit.test.ts +++ b/payjoin-ffi/javascript/test/unit.test.ts @@ -221,8 +221,8 @@ function runUnitTests(name: string, payjoin: typeof nodejsPayjoin) { payjoin.replaySenderEventLog(senderPersister); assert.strictEqual( cancelledResult.state().tag, - "PendingFallback", - "State should be PendingFallback after cancel", + "SenderPendingFallback", + "State should be SenderPendingFallback after cancel", ); pendingFallback.close().save(senderPersister); @@ -269,8 +269,8 @@ function runUnitTests(name: string, payjoin: typeof nodejsPayjoin) { await payjoin.replaySenderEventLogAsync(senderPersister); assert.strictEqual( cancelledResult.state().tag, - "PendingFallback", - "State should be PendingFallback after cancel", + "SenderPendingFallback", + "State should be SenderPendingFallback after cancel", ); await pendingFallback.close().saveAsync(senderPersister); diff --git a/payjoin-ffi/python/test/test_payjoin_unit_test.py b/payjoin-ffi/python/test/test_payjoin_unit_test.py index 44c0f32a0..61b5c0780 100644 --- a/payjoin-ffi/python/test/test_payjoin_unit_test.py +++ b/payjoin-ffi/python/test/test_payjoin_unit_test.py @@ -219,7 +219,7 @@ def test_sender_cancel(self): self.assertIsNotNone(pending_fallback) self.assertTrue(len(pending_fallback.fallback_tx()) > 0) result = payjoin.replay_sender_event_log(persister) - self.assertTrue(result.state().is_PENDING_FALLBACK()) + self.assertTrue(result.state().is_SENDER_PENDING_FALLBACK()) pending_fallback.close().save(persister) result = payjoin.replay_sender_event_log(persister) self.assertTrue(result.state().is_CLOSED()) @@ -259,7 +259,7 @@ async def run_test(): self.assertIsNotNone(pending_fallback) self.assertTrue(len(pending_fallback.fallback_tx()) > 0) result = await payjoin.replay_sender_event_log_async(persister) - self.assertTrue(result.state().is_PENDING_FALLBACK()) + self.assertTrue(result.state().is_SENDER_PENDING_FALLBACK()) await pending_fallback.close().save_async(persister) result = await payjoin.replay_sender_event_log_async(persister) self.assertTrue(result.state().is_CLOSED()) diff --git a/payjoin-ffi/src/receive/mod.rs b/payjoin-ffi/src/receive/mod.rs index 3911ad542..3224644c5 100644 --- a/payjoin-ffi/src/receive/mod.rs +++ b/payjoin-ffi/src/receive/mod.rs @@ -65,87 +65,151 @@ macro_rules! impl_save_for_transition { }; } -/// A terminal transition produced by cancelling a receiver session. +enum ReceiverCancelTransition { + Terminal(payjoin::persist::TerminalTransition), + PendingFallback( + payjoin::persist::NextStateTransition< + payjoin::receive::v2::SessionEvent, + payjoin::receive::v2::Receiver, + >, + ), + MaybePendingFallback( + payjoin::persist::MaybeTerminalTransition< + payjoin::receive::v2::SessionEvent, + payjoin::receive::v2::Receiver, + >, + ), +} + +/// A transition produced by cancelling a receiver session. #[derive(uniffi::Object)] pub struct CancelTransition { - transition: RwLock< - Option< - payjoin::persist::TerminalTransition< - payjoin::receive::v2::SessionEvent, - Option, - >, - >, - >, + transition: RwLock>, } #[uniffi::export] impl CancelTransition { - /// Persist the cancellation and return the fallback transaction if available. - /// - /// The fallback transaction is the consensus-encoded raw transaction bytes, - /// or `None` if the session was cancelled before the sender's original - /// proposal was received. + /// Persist the cancellation and return pending fallback handling if needed. pub fn save( &self, persister: Arc, - ) -> Result>, ReceiverPersistedError> { + ) -> Result>, ReceiverPersistedError> { let adapter = CallbackPersisterAdapter::new(persister); let mut inner = self.transition.write().expect("Lock should not be poisoned"); let value = inner.take().expect("Already saved or moved"); - let fallback = value - .save(&adapter) - .map_err(|e| ReceiverPersistedError::from(ImplementationError::new(e)))?; - Ok(fallback.map(|tx| payjoin::bitcoin::consensus::serialize(&tx))) + match value { + ReceiverCancelTransition::Terminal(transition) => { + transition + .save(&adapter) + .map_err(|e| ReceiverPersistedError::from(ImplementationError::new(e)))?; + Ok(None) + } + ReceiverCancelTransition::PendingFallback(transition) => { + let pending_fallback = transition + .save(&adapter) + .map_err(|e| ReceiverPersistedError::from(ImplementationError::new(e)))?; + Ok(Some(Arc::new(pending_fallback.into()))) + } + ReceiverCancelTransition::MaybePendingFallback(transition) => { + let pending_fallback = transition + .save(&adapter) + .map_err(|e| ReceiverPersistedError::from(ImplementationError::new(e)))?; + Ok(pending_fallback.map(|pending_fallback| Arc::new(pending_fallback.into()))) + } + } } pub async fn save_async( &self, persister: Arc, - ) -> Result>, ReceiverPersistedError> { + ) -> Result>, ReceiverPersistedError> { let adapter = AsyncCallbackPersisterAdapter::new(persister); let value = { let mut inner = self.transition.write().expect("Lock should not be poisoned"); inner.take().expect("Already saved or moved") }; - let fallback = value - .save_async(&adapter) - .await - .map_err(|e| ReceiverPersistedError::from(ImplementationError::new(e)))?; - Ok(fallback.map(|tx| payjoin::bitcoin::consensus::serialize(&tx))) + match value { + ReceiverCancelTransition::Terminal(transition) => { + transition + .save_async(&adapter) + .await + .map_err(|e| ReceiverPersistedError::from(ImplementationError::new(e)))?; + Ok(None) + } + ReceiverCancelTransition::PendingFallback(transition) => { + let pending_fallback = transition + .save_async(&adapter) + .await + .map_err(|e| ReceiverPersistedError::from(ImplementationError::new(e)))?; + Ok(Some(Arc::new(pending_fallback.into()))) + } + ReceiverCancelTransition::MaybePendingFallback(transition) => { + let pending_fallback = transition + .save_async(&adapter) + .await + .map_err(|e| ReceiverPersistedError::from(ImplementationError::new(e)))?; + Ok(pending_fallback.map(|pending_fallback| Arc::new(pending_fallback.into()))) + } + } } } -macro_rules! impl_cancel_for_receiver { +macro_rules! impl_terminal_cancel_for_receiver { ($ty:ident) => { #[uniffi::export] impl $ty { /// Cancel the Payjoin session immediately. - /// - /// Returns a [`CancelTransition`] that, once persisted, yields the fallback - /// transaction when applicable. The fallback transaction is the sender's original - /// transaction that should be broadcast to complete the payment without Payjoin. - /// - /// This is a terminal transition — the session cannot be used after cancellation. pub fn cancel(&self) -> CancelTransition { let transition = self.0.clone().cancel(); - CancelTransition { transition: RwLock::new(Some(transition)) } + CancelTransition { + transition: RwLock::new(Some(ReceiverCancelTransition::Terminal(transition))), + } + } + } + }; +} + +macro_rules! impl_pending_fallback_cancel_for_receiver { + ($ty:ident) => { + #[uniffi::export] + impl $ty { + /// Cancel the Payjoin session and return pending fallback handling. + pub fn cancel(&self) -> CancelTransition { + let transition = self.0.clone().cancel(); + CancelTransition { + transition: RwLock::new(Some(ReceiverCancelTransition::PendingFallback( + transition, + ))), + } } } }; } -impl_cancel_for_receiver!(Initialized); -impl_cancel_for_receiver!(UncheckedOriginalPayload); -impl_cancel_for_receiver!(MaybeInputsOwned); -impl_cancel_for_receiver!(MaybeInputsSeen); -impl_cancel_for_receiver!(OutputsUnknown); -impl_cancel_for_receiver!(WantsOutputs); -impl_cancel_for_receiver!(WantsInputs); -impl_cancel_for_receiver!(WantsFeeRange); -impl_cancel_for_receiver!(ProvisionalProposal); -impl_cancel_for_receiver!(PayjoinProposal); -impl_cancel_for_receiver!(HasReplyableError); -impl_cancel_for_receiver!(Monitor); +impl_terminal_cancel_for_receiver!(Initialized); +impl_terminal_cancel_for_receiver!(UncheckedOriginalPayload); +impl_pending_fallback_cancel_for_receiver!(MaybeInputsOwned); +impl_pending_fallback_cancel_for_receiver!(MaybeInputsSeen); +impl_pending_fallback_cancel_for_receiver!(OutputsUnknown); +impl_pending_fallback_cancel_for_receiver!(WantsOutputs); +impl_pending_fallback_cancel_for_receiver!(WantsInputs); +impl_pending_fallback_cancel_for_receiver!(WantsFeeRange); +impl_pending_fallback_cancel_for_receiver!(ProvisionalProposal); +impl_pending_fallback_cancel_for_receiver!(PayjoinProposal); +impl_pending_fallback_cancel_for_receiver!(Monitor); + +#[uniffi::export] +impl HasReplyableError { + /// Cancel the Payjoin session without posting the receiver error response. + pub fn cancel(&self) -> CancelTransition { + let transition = self.0.clone().cancel(); + CancelTransition { + transition: RwLock::new(Some(ReceiverCancelTransition::MaybePendingFallback( + transition, + ))), + } + } +} #[derive(Debug, Clone, uniffi::Object)] pub struct ReceiverSessionEvent(payjoin::receive::v2::SessionEvent); @@ -198,7 +262,7 @@ pub enum ReceiveSession { PayjoinProposal { inner: Arc }, HasReplyableError { inner: Arc }, Monitor { inner: Arc }, - PendingFallback { inner: Arc }, + ReceiverPendingFallback { inner: Arc }, Closed { inner: Arc }, } @@ -230,7 +294,7 @@ impl From for ReceiveSession { Self::HasReplyableError { inner: Arc::new(inner.into()) }, ReceiveSession::Monitor(inner) => Self::Monitor { inner: Arc::new(inner.into()) }, ReceiveSession::PendingFallback(inner) => - Self::PendingFallback { inner: Arc::new(inner.into()) }, + Self::ReceiverPendingFallback { inner: Arc::new(inner.into()) }, ReceiveSession::Closed(session_outcome) => Self::Closed { inner: Arc::new(session_outcome.into()) }, } @@ -1211,6 +1275,7 @@ pub struct PayjoinProposalTransition( payjoin::receive::v2::SessionEvent, payjoin::receive::v2::Receiver, payjoin::receive::ProtocolError, + payjoin::receive::v2::Receiver, >, >, >, @@ -1290,13 +1355,14 @@ impl From, payjoin::receive::ProtocolError, >, >, @@ -1309,28 +1375,29 @@ impl HasReplyableErrorTransition { pub fn save( &self, persister: Arc, - ) -> Result<(), ReceiverPersistedError> { + ) -> Result>, ReceiverPersistedError> { 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"); - value.save(&adapter).map_err(ReceiverPersistedError::from)?; - Ok(()) + let pending_fallback = value.save(&adapter).map_err(ReceiverPersistedError::from)?; + Ok(pending_fallback.map(|pending_fallback| Arc::new(pending_fallback.into()))) } pub async fn save_async( &self, persister: Arc, - ) -> Result<(), ReceiverPersistedError> { + ) -> Result>, ReceiverPersistedError> { 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") }; - value.save_async(&adapter).await.map_err(ReceiverPersistedError::from)?; - Ok(()) + let pending_fallback = + value.save_async(&adapter).await.map_err(ReceiverPersistedError::from)?; + Ok(pending_fallback.map(|pending_fallback| Arc::new(pending_fallback.into()))) } } @@ -1484,26 +1551,26 @@ impl PendingFallbackTransition { } #[derive(Clone, uniffi::Object)] -pub struct PendingFallback( +pub struct ReceiverPendingFallback( Arc>, ); impl From> - for PendingFallback + for ReceiverPendingFallback { fn from(value: payjoin::receive::v2::Receiver) -> Self { Self(Arc::new(value)) } } -impl From +impl From for payjoin::receive::v2::Receiver { - fn from(value: PendingFallback) -> Self { value.0.as_ref().clone() } + fn from(value: ReceiverPendingFallback) -> Self { value.0.as_ref().clone() } } #[uniffi::export] -impl PendingFallback { +impl ReceiverPendingFallback { pub fn fallback_tx(&self) -> Vec { payjoin::bitcoin::consensus::encode::serialize(self.0.fallback_tx()) } diff --git a/payjoin-ffi/src/send/mod.rs b/payjoin-ffi/src/send/mod.rs index 2346ee40f..8cf8559c6 100644 --- a/payjoin-ffi/src/send/mod.rs +++ b/payjoin-ffi/src/send/mod.rs @@ -71,7 +71,7 @@ impl SenderCancelTransition { pub fn save( &self, persister: Arc, - ) -> Result { + ) -> Result { 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"); @@ -84,7 +84,7 @@ impl SenderCancelTransition { pub async fn save_async( &self, persister: Arc, - ) -> Result { + ) -> Result { let adapter = AsyncCallbackPersisterAdapter::new(persister); let value = { let mut inner = self.0.write().expect("Lock should not be poisoned"); @@ -105,7 +105,7 @@ macro_rules! impl_cancel_for_sender { /// Cancel the Payjoin session immediately. /// /// Returns a [`SenderCancelTransition`] that, once persisted, yields a - /// [`PendingFallback`] state. Call [`PendingFallback::fallback_tx`] to get + /// [`SenderPendingFallback`] state. Call [`SenderPendingFallback::fallback_tx`] to get /// the original transaction pub fn cancel(&self) -> SenderCancelTransition { let transition = self.0.clone().cancel(); @@ -179,7 +179,7 @@ impl SenderSessionOutcome { pub enum SendSession { WithReplyKey { inner: Arc }, PollingForProposal { inner: Arc }, - PendingFallback { inner: Arc }, + SenderPendingFallback { inner: Arc }, Closed { inner: Arc }, } @@ -192,7 +192,7 @@ impl From for SendSession { SendSession::PollingForProposal(inner) => Self::PollingForProposal { inner: Arc::new(inner.into()) }, SendSession::PendingFallback(inner) => - Self::PendingFallback { inner: Arc::new(inner.into()) }, + Self::SenderPendingFallback { inner: Arc::new(inner.into()) }, SendSession::Closed(session_outcome) => Self::Closed { inner: Arc::new(session_outcome.into()) }, } @@ -644,9 +644,9 @@ impl PollingForProposal { } #[derive(Clone, uniffi::Object)] -pub struct PendingFallback(payjoin::send::v2::Sender); +pub struct SenderPendingFallback(payjoin::send::v2::Sender); -impl From> for PendingFallback { +impl From> for SenderPendingFallback { fn from(value: payjoin::send::v2::Sender) -> Self { Self(value) } @@ -687,7 +687,7 @@ impl BroadcastedTransition { } #[uniffi::export] -impl PendingFallback { +impl SenderPendingFallback { /// Returns the fallback transaction as consensus-encoded raw bytes. /// /// This is the sender's original transaction that should be broadcast to diff --git a/payjoin/src/core/receive/v2/mod.rs b/payjoin/src/core/receive/v2/mod.rs index 8fdd6ec0c..d3d2500d3 100644 --- a/payjoin/src/core/receive/v2/mod.rs +++ b/payjoin/src/core/receive/v2/mod.rs @@ -56,7 +56,8 @@ use crate::ohttp::{ use crate::output_substitution::OutputSubstitution; use crate::persist::{ MaybeFatalOrSuccessTransition, MaybeFatalTransition, MaybeFatalTransitionWithNoResults, - MaybeSuccessTransition, MaybeTransientTransition, NextStateTransition, TerminalTransition, + MaybeTerminalSuccessTransition, MaybeTerminalTransition, MaybeTransientTransition, + NextStateTransition, TerminalTransition, }; use crate::receive::{parse_payload, InputPair, OriginalPayload, PsbtContext}; use crate::time::Time; @@ -278,6 +279,23 @@ fn pending_fallback_from( }) } +fn pending_fallback_from_replyable_error( + r: Receiver, + outcome: SessionOutcome, +) -> Result> { + let Receiver { state: HasReplyableError { error_reply, fallback_tx }, session_context } = r; + match fallback_tx { + Some(fallback_tx) => Ok(ReceiveSession::PendingFallback(Receiver { + state: PendingFallback { fallback_tx, outcome }, + session_context, + })), + None => Err(Box::new(ReceiveSession::HasReplyableError(Receiver { + state: HasReplyableError { error_reply, fallback_tx: None }, + session_context, + }))), + } +} + fn try_pending_fallback( session: ReceiveSession, outcome: SessionOutcome, @@ -292,82 +310,27 @@ fn try_pending_fallback( ReceiveSession::ProvisionalProposal(receiver) => Ok(pending_fallback_from(receiver, outcome)), ReceiveSession::PayjoinProposal(receiver) => Ok(pending_fallback_from(receiver, outcome)), + ReceiveSession::HasReplyableError(receiver) => + pending_fallback_from_replyable_error(receiver, outcome), ReceiveSession::Monitor(receiver) => Ok(pending_fallback_from(receiver, outcome)), session => Err(Box::new(session)), } } mod sealed { - pub trait State { - fn maybe_fallback_tx(&self) -> Option { None } - } - + pub trait State {} impl State for super::Initialized {} - - impl State for super::UncheckedOriginalPayload { - fn maybe_fallback_tx(&self) -> Option { - Some(self.original.psbt.clone().extract_tx_unchecked_fee_rate()) - } - } - - impl State for super::MaybeInputsOwned { - fn maybe_fallback_tx(&self) -> Option { - Some(self.original.psbt.clone().extract_tx_unchecked_fee_rate()) - } - } - - impl State for super::MaybeInputsSeen { - fn maybe_fallback_tx(&self) -> Option { - Some(self.original.psbt.clone().extract_tx_unchecked_fee_rate()) - } - } - - impl State for super::OutputsUnknown { - fn maybe_fallback_tx(&self) -> Option { - Some(self.original.psbt.clone().extract_tx_unchecked_fee_rate()) - } - } - - impl State for super::WantsOutputs { - fn maybe_fallback_tx(&self) -> Option { - Some(self.inner.original_psbt.clone().extract_tx_unchecked_fee_rate()) - } - } - - impl State for super::WantsInputs { - fn maybe_fallback_tx(&self) -> Option { - Some(self.inner.original_psbt.clone().extract_tx_unchecked_fee_rate()) - } - } - - impl State for super::WantsFeeRange { - fn maybe_fallback_tx(&self) -> Option { - Some(self.inner.original_psbt.clone().extract_tx_unchecked_fee_rate()) - } - } - - impl State for super::ProvisionalProposal { - fn maybe_fallback_tx(&self) -> Option { - Some(self.psbt_context.original_psbt.clone().extract_tx_unchecked_fee_rate()) - } - } - - impl State for super::PayjoinProposal { - fn maybe_fallback_tx(&self) -> Option { - Some(self.psbt_context.original_psbt.clone().extract_tx_unchecked_fee_rate()) - } - } - - impl State for super::HasReplyableError { - fn maybe_fallback_tx(&self) -> Option { self.fallback_tx.clone() } - } - - impl State for super::Monitor { - fn maybe_fallback_tx(&self) -> Option { - Some(self.psbt_context.original_psbt.clone().extract_tx_unchecked_fee_rate()) - } - } - + impl State for super::UncheckedOriginalPayload {} + impl State for super::MaybeInputsOwned {} + impl State for super::MaybeInputsSeen {} + impl State for super::OutputsUnknown {} + impl State for super::WantsOutputs {} + impl State for super::WantsInputs {} + impl State for super::WantsFeeRange {} + impl State for super::ProvisionalProposal {} + impl State for super::PayjoinProposal {} + impl State for super::HasReplyableError {} + impl State for super::Monitor {} impl State for super::PendingFallback {} pub trait FallbackTx: State { @@ -476,20 +439,6 @@ impl core::ops::DerefMut for Receiver { fn deref_mut(&mut self) -> &mut Self::Target { &mut self.state } } -impl Receiver { - /// Cancel the Payjoin session immediately. - /// - /// Returns a [`TerminalTransition`] that, once persisted, yields the fallback - /// transaction when applicable. The fallback transaction is the sender's original - /// transaction that should be broadcast to complete the payment without Payjoin. - /// - /// This is a terminal transition — the session cannot be used after cancellation. - pub fn cancel(self) -> TerminalTransition> { - let fallback = self.state.maybe_fallback_tx(); - TerminalTransition::new(SessionEvent::Closed(SessionOutcome::Cancel), fallback) - } -} - #[derive(Debug, Clone, PartialEq, Eq)] pub struct PendingFallback { fallback_tx: bitcoin::Transaction, @@ -504,6 +453,27 @@ impl Receiver { } } +impl Receiver { + /// Cancel the Payjoin session and surface the fallback transaction. + pub fn cancel(self) -> NextStateTransition> { + let fallback_tx = self.state.fallback_tx(); + NextStateTransition::success( + SessionEvent::Cancelled, + Receiver { + state: PendingFallback { fallback_tx, outcome: SessionOutcome::Cancel }, + session_context: self.session_context, + }, + ) + } +} + +impl Receiver { + /// Cancel before any fallback transaction exists. + pub fn cancel(self) -> TerminalTransition { + TerminalTransition::new(SessionEvent::Closed(SessionOutcome::Cancel), ()) + } +} + #[derive(Debug, Clone)] pub struct ReceiverBuilder(SessionContext); @@ -751,6 +721,13 @@ pub struct UncheckedOriginalPayload { pub(crate) original: OriginalPayload, } +impl Receiver { + /// Cancel before broadcast suitability has been checked. + pub fn cancel(self) -> TerminalTransition { + TerminalTransition::new(SessionEvent::Closed(SessionOutcome::Cancel), ()) + } +} + /// The original PSBT and the optional parameters received from the sender. /// /// This is the first typestate after the retrieval of the sender's original proposal in @@ -1340,7 +1317,12 @@ impl Receiver { self, res: &[u8], ohttp_context: ohttp::ClientResponse, - ) -> MaybeFatalTransition, ProtocolError> { + ) -> MaybeFatalTransition< + SessionEvent, + Receiver, + ProtocolError, + Receiver, + > { match process_post_res(res, ohttp_context) { Ok(_) => MaybeFatalTransition::success( SessionEvent::PostedPayjoinProposal(), @@ -1351,8 +1333,15 @@ impl Receiver { ), Err(e) => if e.is_fatal() { - MaybeFatalTransition::fatal( - SessionEvent::Closed(SessionOutcome::Failure), + MaybeFatalTransition::replyable_error( + SessionEvent::ProtocolFailed, + Receiver { + state: PendingFallback { + fallback_tx: self.state.fallback_tx(), + outcome: SessionOutcome::Failure, + }, + session_context: self.session_context.clone(), + }, ProtocolError::V2(InternalSessionError::DirectoryResponse(e).into()), ) } else { @@ -1378,6 +1367,22 @@ pub struct HasReplyableError { } impl Receiver { + /// Cancel without sending the error response. + pub fn cancel(self) -> MaybeTerminalTransition> { + let Receiver { state: HasReplyableError { fallback_tx, .. }, session_context } = self; + match fallback_tx { + Some(fallback_tx) => MaybeTerminalTransition::advance( + SessionEvent::Cancelled, + Receiver { + state: PendingFallback { fallback_tx, outcome: SessionOutcome::Cancel }, + session_context, + }, + ), + None => + MaybeTerminalTransition::terminate(SessionEvent::Closed(SessionOutcome::Cancel)), + } + } + /// Construct an OHTTP Encapsulated HTTP POST request to return /// a Receiver Error Response pub fn create_error_request( @@ -1413,28 +1418,43 @@ impl Receiver { } /// Process an OHTTP Encapsulated HTTP POST Error response - /// to ensure it has been posted properly + /// to ensure it has been posted properly. pub fn process_error_response( &self, res: &[u8], ohttp_context: ohttp::ClientResponse, - ) -> MaybeSuccessTransition { - match process_post_res(res, ohttp_context) { - Ok(_) => - MaybeSuccessTransition::success(SessionEvent::Closed(SessionOutcome::Failure), ()), - Err(e) => - if e.is_fatal() { - MaybeSuccessTransition::fatal( - SessionEvent::Closed(SessionOutcome::Failure), - ProtocolError::V2(InternalSessionError::DirectoryResponse(e).into()), - ) - } else { - MaybeSuccessTransition::transient(ProtocolError::V2( - InternalSessionError::DirectoryResponse(e).into(), - )) - }, + ) -> MaybeTerminalSuccessTransition, ProtocolError> + { + let pending = self.pending_fallback_after_protocol_failure(); + let event = match &pending { + Some(_) => SessionEvent::ProtocolFailed, + None => SessionEvent::Closed(SessionOutcome::Failure), + }; + let protocol_error = + |e| ProtocolError::V2(InternalSessionError::DirectoryResponse(e).into()); + + match (process_post_res(res, ohttp_context), pending) { + (Ok(_), Some(pending_fallback)) => + MaybeTerminalSuccessTransition::advance(event, pending_fallback), + (Ok(_), None) => MaybeTerminalSuccessTransition::terminate(event), + (Err(e), _) if !e.is_fatal() => + MaybeTerminalSuccessTransition::transient(protocol_error(e)), + (Err(e), Some(pending_fallback)) => MaybeTerminalSuccessTransition::fatal_advance( + event, + pending_fallback, + protocol_error(e), + ), + (Err(e), None) => + MaybeTerminalSuccessTransition::fatal_terminate(event, protocol_error(e)), } } + + fn pending_fallback_after_protocol_failure(&self) -> Option> { + self.state.fallback_tx.clone().map(|fallback_tx| Receiver { + state: PendingFallback { fallback_tx, outcome: SessionOutcome::Failure }, + session_context: self.session_context.clone(), + }) + } } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -1577,7 +1597,7 @@ pub mod test { use super::*; use crate::output_substitution::OutputSubstitution; use crate::persist::{ - InMemoryPersister, OptionalTransitionOutcome, RejectTransient, Rejection, + InMemoryPersister, OptionalTransitionOutcome, RejectTransient, Rejection, SessionPersister, }; use crate::receive::optional_parameters::Params; use crate::receive::v2; @@ -1633,6 +1653,42 @@ pub mod test { PARSED_ORIGINAL_PSBT.clone().extract_tx_unchecked_fee_rate() } + fn receiver(state: S) -> Receiver { + Receiver { state, session_context: SHARED_CONTEXT.clone() } + } + + fn assert_events( + persister: &InMemoryPersister, + expected_events: &[SessionEvent], + expected_closed: bool, + ) { + let inner = persister.inner.lock().expect("Shouldn't be poisoned"); + assert_eq!(&*inner.events, expected_events); + assert_eq!(inner.is_closed, expected_closed); + } + + fn ohttp_response_for(req_body: &[u8], status: http::StatusCode) -> Vec { + let server = ohttp::Server::new(SHARED_CONTEXT.ohttp_keys.0.clone()) + .expect("test OHTTP server should be valid"); + let (_, probe_response) = server.decapsulate(req_body).expect("request should decapsulate"); + let response_overhead = + probe_response.encapsulate(&[]).expect("probe should encrypt").len(); + + let (_, server_response) = + server.decapsulate(req_body).expect("request should decapsulate again"); + let mut bhttp_response = + vec![0u8; crate::directory::ENCAPSULATED_MESSAGE_BYTES - response_overhead]; + bhttp::Message::response( + bhttp::StatusCode::try_from(status.as_u16()).expect("status should be valid"), + ) + .write_bhttp(bhttp::Mode::KnownLength, &mut bhttp_response.as_mut_slice()) + .expect("BHTTP response should encode"); + let encrypted = + server_response.encapsulate(&bhttp_response).expect("response should encrypt"); + assert_eq!(encrypted.len(), crate::directory::ENCAPSULATED_MESSAGE_BYTES); + encrypted + } + #[test] fn test_monitor_typestate() -> Result<(), BoxError> { let psbt_ctx = PsbtContext { @@ -1945,6 +2001,123 @@ pub mod test { Ok(()) } + #[test] + fn process_error_response_success_with_fallback_enters_pending_fallback() -> Result<(), BoxError> + { + let expected_tx = mock_fallback_tx(); + let receiver = receiver(HasReplyableError { + error_reply: mock_err(), + fallback_tx: Some(expected_tx.clone()), + }); + let (req, ctx) = receiver.create_error_request(EXAMPLE_URL)?; + 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"); + + assert_eq!(pending_fallback.fallback_tx(), &expected_tx); + assert_events(&persister, &[SessionEvent::ProtocolFailed], false); + Ok(()) + } + + #[test] + fn process_error_response_success_without_fallback_closes_session() -> Result<(), BoxError> { + let receiver = receiver(HasReplyableError { error_reply: mock_err(), fallback_tx: None }); + let (req, ctx) = receiver.create_error_request(EXAMPLE_URL)?; + 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)?; + + assert!(pending_fallback.is_none()); + assert_events(&persister, &[SessionEvent::Closed(SessionOutcome::Failure)], true); + Ok(()) + } + + #[test] + fn process_error_response_fatal_with_fallback_enters_pending_fallback() -> Result<(), BoxError> + { + let receiver = receiver(HasReplyableError { + error_reply: mock_err(), + fallback_tx: Some(mock_fallback_tx()), + }); + let (req, ctx) = receiver.create_error_request(EXAMPLE_URL)?; + let response = ohttp_response_for(&req.body, http::StatusCode::BAD_REQUEST); + let persister = InMemoryPersister::::default(); + + let err = receiver + .process_error_response(&response, ctx) + .save(&persister) + .expect_err("fatal response should error"); + + assert!(err.api_error_ref().is_some()); + assert_events(&persister, &[SessionEvent::ProtocolFailed], false); + Ok(()) + } + + #[test] + fn process_error_response_fatal_without_fallback_closes_session() -> Result<(), BoxError> { + let receiver = receiver(HasReplyableError { error_reply: mock_err(), fallback_tx: None }); + let (req, ctx) = receiver.create_error_request(EXAMPLE_URL)?; + let response = ohttp_response_for(&req.body, http::StatusCode::BAD_REQUEST); + let persister = InMemoryPersister::::default(); + + let err = receiver + .process_error_response(&response, ctx) + .save(&persister) + .expect_err("fatal response should error"); + + assert!(err.api_error_ref().is_some()); + assert_events(&persister, &[SessionEvent::Closed(SessionOutcome::Failure)], true); + Ok(()) + } + + #[test] + fn process_error_response_transient_leaves_session_open() -> Result<(), BoxError> { + let receiver = receiver(HasReplyableError { + error_reply: mock_err(), + fallback_tx: Some(mock_fallback_tx()), + }); + let (req, ctx) = receiver.create_error_request(EXAMPLE_URL)?; + let response = ohttp_response_for(&req.body, http::StatusCode::INTERNAL_SERVER_ERROR); + let persister = InMemoryPersister::::default(); + + let err = receiver + .process_error_response(&response, ctx) + .save(&persister) + .expect_err("transient response should error"); + + assert!(err.api_error_ref().is_some()); + assert_events(&persister, &[], false); + Ok(()) + } + + #[test] + fn payjoin_proposal_fatal_response_enters_pending_fallback() -> Result<(), BoxError> { + let expected_tx = mock_fallback_tx(); + let psbt_context = PsbtContext { + original_psbt: PARSED_ORIGINAL_PSBT.clone(), + payjoin_psbt: PARSED_PAYJOIN_PROPOSAL.clone(), + }; + let proposal = receiver(PayjoinProposal { psbt_context }); + let (req, ctx) = proposal.create_post_request(EXAMPLE_URL)?; + let response = ohttp_response_for(&req.body, http::StatusCode::BAD_REQUEST); + let persister = InMemoryPersister::::default(); + + let err = proposal + .process_response(&response, ctx) + .save(&persister) + .expect_err("fatal response should error"); + let pending_fallback = err.error_state().expect("pending fallback should be carried"); + + assert_eq!(pending_fallback.fallback_tx(), &expected_tx); + assert_events(&persister, &[SessionEvent::ProtocolFailed], false); + Ok(()) + } + #[test] fn default_max_fee_rate() { let persister = InMemoryPersister::default(); @@ -2047,61 +2220,153 @@ pub mod test { } #[test] - fn cancel_returns_expected_fallback() { - macro_rules! do_cancel_test { - ($state:expr, $expected:expr) => {{ - let persister = InMemoryPersister::::default(); - let fallback = Receiver { state: $state, session_context: SHARED_CONTEXT.clone() } - .cancel() - .save(&persister) - .expect("save should succeed"); - assert_eq!(fallback, $expected, "cancel from {}", stringify!($state)); - }}; - } + fn cancel_initialized_closes_session() { + let persister = InMemoryPersister::::default(); + receiver(Initialized {}).cancel().save(&persister).expect("save should succeed"); + + assert_events(&persister, &[SessionEvent::Closed(SessionOutcome::Cancel)], true); + } + #[test] + fn cancel_unchecked_original_payload_closes_session() { + let original = + OriginalPayload { psbt: PARSED_ORIGINAL_PSBT.clone(), params: Params::default() }; + let persister = InMemoryPersister::::default(); + receiver(UncheckedOriginalPayload { original }) + .cancel() + .save(&persister) + .expect("save should succeed"); + + assert_events(&persister, &[SessionEvent::Closed(SessionOutcome::Cancel)], true); + } + + #[test] + fn cancel_has_fallback_enters_pending_fallback() { let original = OriginalPayload { psbt: PARSED_ORIGINAL_PSBT.clone(), params: Params::default() }; let expected_tx = PARSED_ORIGINAL_PSBT.clone().extract_tx_unchecked_fee_rate(); - let psbt_ctx = PsbtContext { - original_psbt: PARSED_ORIGINAL_PSBT.clone(), - payjoin_psbt: PARSED_PAYJOIN_PROPOSAL.clone(), - }; - let wants_outputs = common::WantsOutputs::new(original.clone(), vec![0]); - let wants_inputs = wants_outputs.clone().commit_outputs(); - let wants_fee_range = wants_inputs.clone().commit_inputs(); - - // States without a fallback transaction - do_cancel_test!(Initialized {}, None); - do_cancel_test!(HasReplyableError { error_reply: mock_err(), fallback_tx: None }, None); - - // States with a fallback transaction - let error_fallback = mock_fallback_tx(); - do_cancel_test!( - HasReplyableError { - error_reply: mock_err(), - fallback_tx: Some(error_fallback.clone()) - }, - Some(error_fallback) - ); - do_cancel_test!( - UncheckedOriginalPayload { original: original.clone() }, - Some(expected_tx.clone()) - ); - do_cancel_test!(MaybeInputsOwned { original: original.clone() }, Some(expected_tx.clone())); - do_cancel_test!(MaybeInputsSeen { original: original.clone() }, Some(expected_tx.clone())); - do_cancel_test!(OutputsUnknown { original }, Some(expected_tx.clone())); - do_cancel_test!(WantsOutputs { inner: wants_outputs }, Some(expected_tx.clone())); - do_cancel_test!(WantsInputs { inner: wants_inputs }, Some(expected_tx.clone())); - do_cancel_test!(WantsFeeRange { inner: wants_fee_range }, Some(expected_tx.clone())); - do_cancel_test!( - ProvisionalProposal { psbt_context: psbt_ctx.clone() }, - Some(expected_tx.clone()) - ); - do_cancel_test!( - PayjoinProposal { psbt_context: psbt_ctx.clone() }, - Some(expected_tx.clone()) - ); - do_cancel_test!(Monitor { psbt_context: psbt_ctx }, Some(expected_tx)); + let persister = InMemoryPersister::::default(); + let pending_fallback = receiver(MaybeInputsOwned { original }) + .cancel() + .save(&persister) + .expect("save should succeed"); + + assert_eq!(pending_fallback.fallback_tx(), &expected_tx); + assert_events(&persister, &[SessionEvent::Cancelled], false); + } + + #[test] + fn cancel_replyable_error_with_fallback_enters_pending_fallback() { + let expected_tx = mock_fallback_tx(); + let persister = InMemoryPersister::::default(); + let pending_fallback = receiver(HasReplyableError { + error_reply: mock_err(), + fallback_tx: Some(expected_tx.clone()), + }) + .cancel() + .save(&persister) + .expect("save should succeed") + .expect("pending fallback should be returned"); + + assert_eq!(pending_fallback.fallback_tx(), &expected_tx); + assert_events(&persister, &[SessionEvent::Cancelled], false); + } + + #[test] + fn cancel_replyable_error_without_fallback_closes_session() { + let persister = InMemoryPersister::::default(); + let pending_fallback = + receiver(HasReplyableError { error_reply: mock_err(), fallback_tx: None }) + .cancel() + .save(&persister) + .expect("save should succeed"); + + assert!(pending_fallback.is_none()); + assert_events(&persister, &[SessionEvent::Closed(SessionOutcome::Cancel)], true); + } + + #[test] + fn replaying_cancel_event_sequences_reaches_expected_states() { + let original = + OriginalPayload { psbt: PARSED_ORIGINAL_PSBT.clone(), params: Params::default() }; + let expected_tx = PARSED_ORIGINAL_PSBT.clone().extract_tx_unchecked_fee_rate(); + let replyable_error = mock_err(); + + let test_cases = vec![ + ( + vec![ + SessionEvent::Created(SHARED_CONTEXT.clone()), + SessionEvent::Closed(SessionOutcome::Cancel), + ], + ReceiveSession::Closed(SessionOutcome::Cancel), + ), + ( + vec![ + SessionEvent::Created(SHARED_CONTEXT.clone()), + SessionEvent::RetrievedOriginalPayload { + original: original.clone(), + reply_key: None, + }, + SessionEvent::Closed(SessionOutcome::Cancel), + ], + ReceiveSession::Closed(SessionOutcome::Cancel), + ), + ( + vec![ + SessionEvent::Created(SHARED_CONTEXT.clone()), + SessionEvent::RetrievedOriginalPayload { + original: original.clone(), + reply_key: None, + }, + SessionEvent::CheckedBroadcastSuitability(), + SessionEvent::Cancelled, + ], + ReceiveSession::PendingFallback(Receiver { + state: PendingFallback { + fallback_tx: expected_tx.clone(), + outcome: SessionOutcome::Cancel, + }, + session_context: SHARED_CONTEXT.clone(), + }), + ), + ( + vec![ + SessionEvent::Created(SHARED_CONTEXT.clone()), + SessionEvent::RetrievedOriginalPayload { + original: original.clone(), + reply_key: None, + }, + SessionEvent::CheckedBroadcastSuitability(), + SessionEvent::GotReplyableError(replyable_error.clone()), + SessionEvent::Cancelled, + ], + ReceiveSession::PendingFallback(Receiver { + state: PendingFallback { + fallback_tx: expected_tx, + outcome: SessionOutcome::Cancel, + }, + session_context: SHARED_CONTEXT.clone(), + }), + ), + ( + vec![ + SessionEvent::Created(SHARED_CONTEXT.clone()), + SessionEvent::RetrievedOriginalPayload { original, reply_key: None }, + SessionEvent::GotReplyableError(replyable_error), + SessionEvent::Closed(SessionOutcome::Cancel), + ], + ReceiveSession::Closed(SessionOutcome::Cancel), + ), + ]; + + for (events, expected_state) in test_cases { + let persister = InMemoryPersister::::default(); + for event in events { + persister.save_event(event).expect("save should succeed"); + } + let (state, _) = replay_event_log(&persister).expect("replay should succeed"); + assert_eq!(state, expected_state); + } } #[test] diff --git a/payjoin/tests/integration.rs b/payjoin/tests/integration.rs index 88ae1fc86..c697afdeb 100644 --- a/payjoin/tests/integration.rs +++ b/payjoin/tests/integration.rs @@ -464,7 +464,18 @@ mod integration { let err_bytes = err_response.bytes().await?; has_error.process_error_response(&err_bytes, err_ctx).save(&persister)?; - // Ensure the session is closed properly + // The error response was sent successfully and the source state + // carried a fallback transaction, so the session is now waiting + // on the wallet to acknowledge the fallback obligation. + let (session, session_history) = replay_receiver_event_log(&persister)?; + assert_eq!(session_history.status(), SessionStatus::PendingFallback); + let pending = match session { + ReceiveSession::PendingFallback(r) => r, + _ => panic!("Expected PendingFallback"), + }; + pending.close().save(&persister)?; + + // After the wallet closes, the session terminates with a Failed status. let (_, session_history) = replay_receiver_event_log(&persister)?; assert_eq!(session_history.status(), SessionStatus::Failed); From 506c41fd43409e23842b0b58d82f8907d3564b37 Mon Sep 17 00:00:00 2001 From: spacebear Date: Wed, 13 May 2026 23:14:53 -0400 Subject: [PATCH 06/10] Remove MaybeSuccessTransition and fatal_with_state alias Drop dead code. --- payjoin/src/core/persist.rs | 128 ------------------------------------ 1 file changed, 128 deletions(-) diff --git a/payjoin/src/core/persist.rs b/payjoin/src/core/persist.rs index 81f89eb0b..1bec19191 100644 --- a/payjoin/src/core/persist.rs +++ b/payjoin/src/core/persist.rs @@ -265,11 +265,6 @@ where MaybeFatalTransition(Err(Rejection::replyable_error(event, error_state, error))) } - #[allow(dead_code)] - pub(crate) fn fatal_with_state(event: Event, error_state: ErrorState, error: Err) -> Self { - MaybeFatalTransition(Err(Rejection::replyable_error(event, error_state, error))) - } - pub(crate) fn deconstruct( self, ) -> (PersistActions, Result>) { @@ -400,74 +395,6 @@ impl NextStateTransition { } } -/// A transition that can result in the completion of a state machine or a transient error -/// Fatal errors cannot occur in this transition. -pub struct MaybeSuccessTransition( - Result, Rejection>, -); - -impl MaybeSuccessTransition -where - Err: std::error::Error, -{ - #[allow(dead_code)] - pub(crate) fn success(event: Event, success_value: SuccessValue) -> Self { - MaybeSuccessTransition(Ok(AcceptNextState(event, success_value))) - } - - #[allow(dead_code)] - pub(crate) fn transient(error: Err) -> Self { - MaybeSuccessTransition(Err(Rejection::transient(error))) - } - - #[allow(dead_code)] - pub(crate) fn fatal(event: Event, error: Err) -> Self { - MaybeSuccessTransition(Err(Rejection::fatal(event, error))) - } - - pub(crate) fn deconstruct( - self, - ) -> (PersistActions, Result>) { - match self.0 { - Ok(AcceptNextState(event, success_value)) => - (PersistActions::SaveAndClose(event), Ok(success_value)), - Err(Rejection::Transient(RejectTransient(error))) => - (PersistActions::NoOp, Err(ApiError::Transient(error))), - Err(Rejection::Fatal(RejectFatal(event, error))) => - (PersistActions::SaveAndClose(event), Err(ApiError::Fatal(error))), - Err(Rejection::ReplyableError(RejectReplyableError(event, _, error))) => - (PersistActions::Save(event), Err(ApiError::Fatal(error))), - } - } - - pub fn save

( - self, - persister: &P, - ) -> Result> - where - P: SessionPersister, - { - let (actions, outcome) = self.deconstruct(); - actions.execute(persister).map_err(InternalPersistedError::Storage)?; - Ok(outcome.map_err(InternalPersistedError::Api)?) - } - - pub async fn save_async

( - self, - persister: &P, - ) -> Result> - where - P: AsyncSessionPersister, - Err: Send, - SuccessValue: Send, - Event: Send, - { - let (actions, outcome) = self.deconstruct(); - actions.execute_async(persister).await.map_err(InternalPersistedError::Storage)?; - Ok(outcome.map_err(InternalPersistedError::Api)?) - } -} - /// A transition that either advances to a live state or terminates the session. /// /// No error path exists. Both outcomes are successful from the protocol's point @@ -475,7 +402,6 @@ where /// not by the caller. pub struct MaybeTerminalTransition(MaybeTerminalOutcome); -#[allow(dead_code)] impl MaybeTerminalTransition { pub(crate) fn advance(event: Event, next_state: NextState) -> Self { Self(MaybeTerminalOutcome::Advance(AcceptNextState(event, next_state))) @@ -524,7 +450,6 @@ pub struct MaybeTerminalSuccessTransition( MaybeTerminalSuccessOutcome, ); -#[allow(dead_code)] impl MaybeTerminalSuccessTransition where Err: std::error::Error, @@ -710,13 +635,11 @@ where /// Wrapper that marks the progression of a state machine pub struct AcceptNextState(Event, NextState); -#[allow(dead_code)] enum MaybeTerminalOutcome { Advance(AcceptNextState), Terminate(Event), } -#[allow(dead_code)] enum MaybeTerminalSuccessOutcome { Advance(AcceptNextState), Terminate(Event), @@ -1378,57 +1301,6 @@ mod tests { run_test_cases!(test_cases); } - #[tokio::test] - async fn test_maybe_success_transition() { - let event = InMemoryTestEvent("foo".to_string()); - let error_event = InMemoryTestEvent("error event".to_string()); - - let test_cases = vec![ - TestCase { - make_transition: Box::new({ - let event = event.clone(); - move || MaybeSuccessTransition::success(event.clone(), ()) - }), - expected_result: ExpectedResult { - events: vec![event.clone()], - is_closed: true, - error: None, - success: Some(()), - }, - }, - TestCase { - make_transition: Box::new(|| { - MaybeSuccessTransition::transient(InMemoryTestError {}) - }), - expected_result: ExpectedResult { - events: vec![], - is_closed: false, - error: Some( - InternalPersistedError::Api(ApiError::Transient(InMemoryTestError {})) - .into(), - ), - success: None, - }, - }, - TestCase { - make_transition: Box::new({ - let error_event = error_event.clone(); - move || MaybeSuccessTransition::fatal(error_event.clone(), InMemoryTestError {}) - }), - expected_result: ExpectedResult { - events: vec![error_event.clone()], - is_closed: true, - error: Some( - InternalPersistedError::Api(ApiError::Fatal(InMemoryTestError {})).into(), - ), - success: None, - }, - }, - ]; - - run_test_cases!(test_cases); - } - #[tokio::test] async fn test_maybe_fatal_transition() { let event = InMemoryTestEvent("foo".to_string()); From 00e72be53f21bec387e3f23f42adf9994fd984c7 Mon Sep 17 00:00:00 2001 From: spacebear Date: Wed, 20 May 2026 19:32:22 -0400 Subject: [PATCH 07/10] cli: Add receiver support to cancel command --- payjoin-cli/src/app/mod.rs | 9 ++- payjoin-cli/src/app/v1.rs | 3 +- payjoin-cli/src/app/v2/mod.rs | 125 ++++++++++++++++++++++++++++++---- payjoin-cli/src/cli/mod.rs | 14 +++- payjoin-cli/src/db/v2.rs | 2 + payjoin-cli/src/main.rs | 4 +- payjoin-cli/tests/e2e.rs | 111 ++++++++++++++++++++++++++++++ 7 files changed, 250 insertions(+), 18 deletions(-) diff --git a/payjoin-cli/src/app/mod.rs b/payjoin-cli/src/app/mod.rs index 71c21039c..700999660 100644 --- a/payjoin-cli/src/app/mod.rs +++ b/payjoin-cli/src/app/mod.rs @@ -11,6 +11,8 @@ pub mod wallet; use crate::app::config::Config; use crate::app::wallet::BitcoindWallet; #[cfg(feature = "v2")] +use crate::cli::Role; +#[cfg(feature = "v2")] use crate::db::v2::SessionId; #[cfg(feature = "v1")] @@ -31,7 +33,12 @@ pub trait App: Send + Sync { #[cfg(feature = "v2")] async fn history(&self) -> Result<()>; #[cfg(feature = "v2")] - async fn cancel_sender(&self, session_id: SessionId, no_broadcast: bool) -> Result<()>; + async fn cancel( + &self, + session_id: SessionId, + no_broadcast: bool, + role: Option, + ) -> Result<()>; fn create_original_psbt( &self, diff --git a/payjoin-cli/src/app/v1.rs b/payjoin-cli/src/app/v1.rs index f557facd3..017abbf2a 100644 --- a/payjoin-cli/src/app/v1.rs +++ b/payjoin-cli/src/app/v1.rs @@ -129,10 +129,11 @@ impl AppTrait for App { } #[cfg(feature = "v2")] - async fn cancel_sender( + async fn cancel( &self, _session_id: crate::db::v2::SessionId, _no_broadcast: bool, + _role: Option, ) -> Result<()> { anyhow::bail!("cancel is only supported for v2 (BIP77) sessions") } diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index 29f0440dd..2a9a74918 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -8,13 +8,14 @@ use payjoin::persist::{OptionalTransitionOutcome, SessionPersister}; use payjoin::receive::v2::{ replay_event_log as replay_receiver_event_log, HasReplyableError, Initialized, MaybeInputsOwned, MaybeInputsSeen, Monitor, OutputsUnknown, PayjoinProposal, - ProvisionalProposal, ReceiveSession, Receiver, ReceiverBuilder, - SessionOutcome as ReceiverSessionOutcome, UncheckedOriginalPayload, WantsFeeRange, WantsInputs, - WantsOutputs, + PendingFallback as ReceiverPendingFallback, ProvisionalProposal, ReceiveSession, Receiver, + ReceiverBuilder, SessionOutcome as ReceiverSessionOutcome, UncheckedOriginalPayload, + WantsFeeRange, WantsInputs, WantsOutputs, }; use payjoin::send::v2::{ - replay_event_log as replay_sender_event_log, PendingFallback, PollingForProposal, SendSession, - Sender, SenderBuilder, SessionOutcome as SenderSessionOutcome, WithReplyKey, + replay_event_log as replay_sender_event_log, PendingFallback as SenderPendingFallback, + PollingForProposal, SendSession, Sender, SenderBuilder, SessionOutcome as SenderSessionOutcome, + WithReplyKey, }; use payjoin::{ImplementationError, PjParam, Uri}; use tokio::sync::watch; @@ -24,6 +25,7 @@ use super::wallet::BitcoindWallet; use super::App as AppTrait; use crate::app::v2::ohttp::{unwrap_ohttp_keys_or_else_fetch, RelayManager}; use crate::app::{handle_interrupt, http_agent}; +use crate::cli::Role as CliRole; use crate::db::v2::{ReceiverPersister, SenderPersister, SessionId}; use crate::db::Database; @@ -504,11 +506,42 @@ impl AppTrait for App { Ok(()) } - async fn cancel_sender(&self, session_id: SessionId, no_broadcast: bool) -> Result<()> { + async fn cancel( + &self, + session_id: SessionId, + no_broadcast: bool, + role: Option, + ) -> Result<()> { + if let Some(role) = role { + return match role { + CliRole::Sender => self.cancel_sender_session(session_id, no_broadcast), + CliRole::Receiver => self.cancel_receiver_session(session_id, no_broadcast), + }; + } + + let send_ids = self.db.get_send_session_ids()?; + let recv_ids = self.db.get_recv_session_ids()?; + let is_sender = send_ids.iter().any(|id| id.0 == session_id.0); + let is_receiver = recv_ids.iter().any(|id| id.0 == session_id.0); + + match (is_sender, is_receiver) { + (true, false) => self.cancel_sender_session(session_id, no_broadcast), + (false, true) => self.cancel_receiver_session(session_id, no_broadcast), + (true, true) => anyhow::bail!( + "Session {session_id} exists as both a sender and receiver session. \ + Pass `--role sender` or `--role receiver`." + ), + (false, false) => anyhow::bail!("Session {session_id} not found"), + } + } +} + +impl App { + fn cancel_sender_session(&self, session_id: SessionId, no_broadcast: bool) -> Result<()> { let persister = SenderPersister::from_id(self.db.clone(), session_id.clone()); let (session, _history) = replay_sender_event_log(&persister)?; - let pending: Sender = match session { + let pending: Sender = match session { SendSession::WithReplyKey(sender) => sender.cancel().save(&persister)?, SendSession::PollingForProposal(sender) => sender.cancel().save(&persister)?, SendSession::PendingFallback(sender) => sender, @@ -541,9 +574,72 @@ impl AppTrait for App { pending.close().save(&persister)?; Ok(()) } -} -impl App { + fn cancel_receiver_session(&self, session_id: SessionId, no_broadcast: bool) -> Result<()> { + let persister = ReceiverPersister::from_id(self.db.clone(), session_id.clone()); + let (session, _history) = replay_receiver_event_log(&persister)?; + + let pending: Receiver = match session { + ReceiveSession::Initialized(receiver) => { + receiver.cancel().save(&persister)?; + println!("Session {session_id} cancelled. No fallback transaction to broadcast."); + return Ok(()); + } + ReceiveSession::UncheckedOriginalPayload(receiver) => { + receiver.cancel().save(&persister)?; + println!("Session {session_id} cancelled. No fallback transaction to broadcast."); + return Ok(()); + } + ReceiveSession::MaybeInputsOwned(receiver) => receiver.cancel().save(&persister)?, + ReceiveSession::MaybeInputsSeen(receiver) => receiver.cancel().save(&persister)?, + ReceiveSession::OutputsUnknown(receiver) => receiver.cancel().save(&persister)?, + ReceiveSession::WantsOutputs(receiver) => receiver.cancel().save(&persister)?, + ReceiveSession::WantsInputs(receiver) => receiver.cancel().save(&persister)?, + ReceiveSession::WantsFeeRange(receiver) => receiver.cancel().save(&persister)?, + ReceiveSession::ProvisionalProposal(receiver) => receiver.cancel().save(&persister)?, + ReceiveSession::PayjoinProposal(receiver) => receiver.cancel().save(&persister)?, + ReceiveSession::Monitor(receiver) => receiver.cancel().save(&persister)?, + ReceiveSession::HasReplyableError(receiver) => match receiver + .cancel() + .save(&persister)? + { + Some(pending) => pending, + None => { + println!("Session {session_id} cancelled. No fallback transaction available."); + return Ok(()); + } + }, + ReceiveSession::PendingFallback(receiver) => receiver, + ReceiveSession::Closed( + ReceiverSessionOutcome::Success(_) + | ReceiverSessionOutcome::FallbackBroadcasted + | ReceiverSessionOutcome::PayjoinProposalSent, + ) => { + println!("Session {session_id} already completed successfully. Cannot cancel."); + return Ok(()); + } + ReceiveSession::Closed(_) => { + println!("Session {session_id} is already closed. Nothing left to do."); + return Ok(()); + } + }; + + if no_broadcast { + println!( + "Session {session_id} cancelled. Broadcast the fallback transaction manually:\n{}", + serialize_hex(pending.fallback_tx()) + ); + } else { + self.wallet().broadcast_tx(pending.fallback_tx())?; + println!( + "Broadcasted fallback transaction txid: {}", + pending.fallback_tx().compute_txid() + ); + } + pending.close().save(&persister)?; + Ok(()) + } + fn close_failed_session

(persister: &P, session_id: &SessionId, role: &str) where P: SessionPersister, @@ -577,7 +673,7 @@ impl App { SendSession::PendingFallback(_) => { let id = persister.session_id(); println!( - "Session {id} was cancelled. Run `payjoin-cli cancel {id}` to cancel and broadcast the original transaction." + "Session {id} was cancelled. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction." ); return Ok(()); } @@ -691,8 +787,13 @@ impl App { self.handle_error(error, persister).await, ReceiveSession::Monitor(proposal) => self.monitor_payjoin_proposal(proposal, persister).await, - ReceiveSession::PendingFallback(_) => - return Err(anyhow!("PendingFallback session handling not yet implemented")), + ReceiveSession::PendingFallback(_) => { + let id = persister.session_id(); + println!( + "Session {id} was cancelled. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction." + ); + return Ok(()); + } ReceiveSession::Closed(_) => return Err(anyhow!("Session closed")), } }; diff --git a/payjoin-cli/src/cli/mod.rs b/payjoin-cli/src/cli/mod.rs index 07b087edb..35c3aa78e 100644 --- a/payjoin-cli/src/cli/mod.rs +++ b/payjoin-cli/src/cli/mod.rs @@ -1,6 +1,6 @@ use std::path::PathBuf; -use clap::{value_parser, Parser, Subcommand}; +use clap::{value_parser, Parser, Subcommand, ValueEnum}; use payjoin::bitcoin::amount::ParseAmountError; use payjoin::bitcoin::{Amount, FeeRate}; use payjoin::Url; @@ -133,7 +133,7 @@ pub enum Commands { /// Show payjoin session history History, #[cfg(feature = "v2")] - /// Cancel a sender session, broadcasting the fallback transaction by default (BIP77/v2 only) + /// Cancel a sender or receiver session, broadcasting the fallback transaction by default (BIP77/v2 only) Cancel { /// The session ID to cancel #[arg(required = true)] @@ -142,9 +142,19 @@ pub enum Commands { /// Cancel without broadcasting the fallback transaction #[arg(long = "no-broadcast")] no_broadcast: bool, + + /// The session role to cancel + #[arg(long = "role", value_enum)] + role: Option, }, } +#[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)] +pub enum Role { + Sender, + Receiver, +} + pub fn parse_amount_in_sat(s: &str) -> Result { Amount::from_str_in(s, payjoin::bitcoin::Denomination::Satoshi) } diff --git a/payjoin-cli/src/db/v2.rs b/payjoin-cli/src/db/v2.rs index bd80db5cc..fd802179d 100644 --- a/payjoin-cli/src/db/v2.rs +++ b/payjoin-cli/src/db/v2.rs @@ -141,6 +141,8 @@ impl ReceiverPersister { } pub fn from_id(db: Arc, id: SessionId) -> Self { Self { db, session_id: id } } + + pub fn session_id(&self) -> SessionId { self.session_id.clone() } } impl SessionPersister for ReceiverPersister { diff --git a/payjoin-cli/src/main.rs b/payjoin-cli/src/main.rs index b988a46ea..b42d85d1b 100644 --- a/payjoin-cli/src/main.rs +++ b/payjoin-cli/src/main.rs @@ -79,8 +79,8 @@ async fn main() -> Result<()> { app.history().await?; } #[cfg(feature = "v2")] - Commands::Cancel { session_id, no_broadcast } => { - app.cancel_sender(SessionId(*session_id), *no_broadcast).await?; + Commands::Cancel { session_id, no_broadcast, role } => { + app.cancel(SessionId(*session_id), *no_broadcast, *role).await?; } }; diff --git a/payjoin-cli/tests/e2e.rs b/payjoin-cli/tests/e2e.rs index 2d6f45406..d63e299f1 100644 --- a/payjoin-cli/tests/e2e.rs +++ b/payjoin-cli/tests/e2e.rs @@ -788,4 +788,115 @@ mod e2e { Ok(()) } + + #[cfg(feature = "v2")] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn receiver_cancel_v2() -> Result<(), Box> { + use payjoin_test_utils::{init_tracing, TestServices}; + use tempfile::TempDir; + + type Result = std::result::Result; + + init_tracing(); + let mut services = TestServices::initialize().await?; + let temp_dir = tempdir()?; + + let result = tokio::select! { + res = services.take_ohttp_relay_handle() => Err(format!("Ohttp relay is long running: {res:?}").into()), + res = services.take_directory_handle() => Err(format!("Directory server is long running: {res:?}").into()), + res = cancel_cli_async(&services, &temp_dir) => res, + }; + + assert!(result.is_ok(), "receiver_cancel_v2 failed: {:#?}", result.unwrap_err()); + + async fn cancel_cli_async(services: &TestServices, temp_dir: &TempDir) -> Result<()> { + let receiver_db_path = temp_dir.path().join("receiver_db"); + let (bitcoind, _sender, _receiver) = init_bitcoind_sender_receiver(None, None)?; + let cert_path = &temp_dir.path().join("localhost.der"); + tokio::fs::write(cert_path, services.cert()).await?; + services.wait_for_services_ready().await?; + let ohttp_keys = services.fetch_ohttp_keys().await?; + let ohttp_keys_path = temp_dir.path().join("ohttp_keys"); + tokio::fs::write(&ohttp_keys_path, ohttp_keys.encode()?).await?; + + let receiver_rpchost = format!("http://{}/wallet/receiver", bitcoind.params.rpc_socket); + let cookie_file = &bitcoind.params.cookie_file; + let payjoin_cli = env!("CARGO_BIN_EXE_payjoin-cli"); + let directory = &services.directory_url(); + let ohttp_relay = &services.ohttp_relay_url(); + + // Start a receiver and capture its BIP21 so a session is persisted, + // then leave it parked at Initialized waiting for a proposal. + let cli_receiver = Command::new(payjoin_cli) + .arg("--root-certificate") + .arg(cert_path) + .arg("--rpchost") + .arg(&receiver_rpchost) + .arg("--cookie-file") + .arg(cookie_file) + .arg("--db-path") + .arg(&receiver_db_path) + .arg("--ohttp-relays") + .arg(ohttp_relay) + .arg("receive") + .arg(RECEIVE_SATS) + .arg("--pj-directory") + .arg(directory) + .arg("--ohttp-keys") + .arg(&ohttp_keys_path) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .expect("Failed to execute payjoin-cli receiver"); + let _bip21 = get_bip21_from_receiver(cli_receiver).await; + + // There is only one receiver session in progress. + let session_id = 1i64; + + // Run `payjoin-cli cancel --role receiver`: the session is at + // Initialized so there is no fallback transaction to broadcast. + let mut cli_cancel = Command::new(payjoin_cli) + .arg("--root-certificate") + .arg(cert_path) + .arg("--rpchost") + .arg(&receiver_rpchost) + .arg("--cookie-file") + .arg(cookie_file) + .arg("--db-path") + .arg(&receiver_db_path) + .arg("--ohttp-relays") + .arg(ohttp_relay) + .arg("cancel") + .arg(session_id.to_string()) + .arg("--role") + .arg("receiver") + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .expect("Failed to execute payjoin-cli cancel"); + + let mut cancel_stdout = + cli_cancel.stdout.take().expect("failed to take stdout of cancel"); + let timeout = tokio::time::Duration::from_secs(10); + let cancel_line = tokio::time::timeout( + timeout, + wait_for_stdout_match(&mut cancel_stdout, |l| { + l.contains("No fallback transaction to broadcast") + }), + ) + .await?; + terminate(cli_cancel).await.expect("Failed to kill payjoin-cli cancel"); + let subcommand_output = + cancel_line.expect("cancel should report no fallback transaction"); + + assert!( + subcommand_output.contains(&format!("Session {session_id} cancelled")), + "cancel should reference the cancelled session id" + ); + + Ok(()) + } + + Ok(()) + } } From 73418ceb37945e9bff9a96f5bf824948f7c3de7f Mon Sep 17 00:00:00 2001 From: Benalleng Date: Tue, 2 Jun 2026 16:32:24 -0400 Subject: [PATCH 08/10] Add ReceiveSession test vectors This adds test vectors for - SessionEvent::ProtocolFailed results in ReceiveSession::PendingFallback - ReceiveSession::PendingFallback results in RecieveSession::Closed --- payjoin/src/core/receive/v2/mod.rs | 69 ++++++++++++++++++++++++++++-- 1 file changed, 66 insertions(+), 3 deletions(-) diff --git a/payjoin/src/core/receive/v2/mod.rs b/payjoin/src/core/receive/v2/mod.rs index d3d2500d3..fee21a68e 100644 --- a/payjoin/src/core/receive/v2/mod.rs +++ b/payjoin/src/core/receive/v2/mod.rs @@ -2342,7 +2342,7 @@ pub mod test { ], ReceiveSession::PendingFallback(Receiver { state: PendingFallback { - fallback_tx: expected_tx, + fallback_tx: expected_tx.clone(), outcome: SessionOutcome::Cancel, }, session_context: SHARED_CONTEXT.clone(), @@ -2351,12 +2351,75 @@ pub mod test { ( vec![ SessionEvent::Created(SHARED_CONTEXT.clone()), - SessionEvent::RetrievedOriginalPayload { original, reply_key: None }, - SessionEvent::GotReplyableError(replyable_error), + SessionEvent::RetrievedOriginalPayload { + original: original.clone(), + reply_key: None, + }, + SessionEvent::CheckedBroadcastSuitability(), + SessionEvent::ProtocolFailed, + ], + ReceiveSession::PendingFallback(Receiver { + state: PendingFallback { + fallback_tx: expected_tx.clone(), + outcome: SessionOutcome::Failure, + }, + session_context: SHARED_CONTEXT.clone(), + }), + ), + ( + vec![ + SessionEvent::Created(SHARED_CONTEXT.clone()), + SessionEvent::RetrievedOriginalPayload { + original: original.clone(), + reply_key: None, + }, + SessionEvent::CheckedBroadcastSuitability(), + SessionEvent::GotReplyableError(replyable_error.clone()), + SessionEvent::ProtocolFailed, + ], + ReceiveSession::PendingFallback(Receiver { + state: PendingFallback { + fallback_tx: expected_tx, + outcome: SessionOutcome::Failure, + }, + session_context: SHARED_CONTEXT.clone(), + }), + ), + ( + vec![ + SessionEvent::Created(SHARED_CONTEXT.clone()), + SessionEvent::RetrievedOriginalPayload { + original: original.clone(), + reply_key: None, + }, + SessionEvent::GotReplyableError(replyable_error.clone()), + SessionEvent::Closed(SessionOutcome::Cancel), + ], + ReceiveSession::Closed(SessionOutcome::Cancel), + ), + ( + vec![ + SessionEvent::Created(SHARED_CONTEXT.clone()), + SessionEvent::RetrievedOriginalPayload { + original: original.clone(), + reply_key: None, + }, + SessionEvent::CheckedBroadcastSuitability(), + SessionEvent::Cancelled, SessionEvent::Closed(SessionOutcome::Cancel), ], ReceiveSession::Closed(SessionOutcome::Cancel), ), + ( + vec![ + SessionEvent::Created(SHARED_CONTEXT.clone()), + SessionEvent::RetrievedOriginalPayload { original, reply_key: None }, + SessionEvent::CheckedBroadcastSuitability(), + SessionEvent::ProtocolFailed, + SessionEvent::Closed(SessionOutcome::Failure), + ], + ReceiveSession::Closed(SessionOutcome::Failure), + ), ]; for (events, expected_state) in test_cases { From 461b714193b12ba8e33f59f05c4873178f95ff98 Mon Sep 17 00:00:00 2001 From: Benalleng Date: Wed, 3 Jun 2026 11:56:09 -0400 Subject: [PATCH 09/10] Use FatalWithState for FatalAdvance --- payjoin-ffi/src/receive/error.rs | 6 ++++-- payjoin/src/core/persist.rs | 12 +++++++----- payjoin/src/core/receive/v2/mod.rs | 5 ++++- 3 files changed, 15 insertions(+), 8 deletions(-) diff --git a/payjoin-ffi/src/receive/error.rs b/payjoin-ffi/src/receive/error.rs index b96f9e9fb..60f25fd5f 100644 --- a/payjoin-ffi/src/receive/error.rs +++ b/payjoin-ffi/src/receive/error.rs @@ -59,11 +59,13 @@ macro_rules! impl_persisted_error_from { $api_error_ty:ty, $receiver_arm:expr ) => { - impl From> for ReceiverPersistedError + impl From> + for ReceiverPersistedError where S: std::error::Error + Send + Sync + 'static, + E: std::fmt::Debug, { - fn from(err: payjoin::persist::PersistedError<$api_error_ty, S>) -> Self { + fn from(err: payjoin::persist::PersistedError<$api_error_ty, S, E>) -> 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/src/core/persist.rs b/payjoin/src/core/persist.rs index 1bec19191..969513120 100644 --- a/payjoin/src/core/persist.rs +++ b/payjoin/src/core/persist.rs @@ -453,6 +453,7 @@ pub struct MaybeTerminalSuccessTransition( impl MaybeTerminalSuccessTransition where Err: std::error::Error, + NextState: fmt::Debug, { pub(crate) fn advance(event: Event, next_state: NextState) -> Self { Self(MaybeTerminalSuccessOutcome::Advance(AcceptNextState(event, next_state))) @@ -474,16 +475,17 @@ where Self(MaybeTerminalSuccessOutcome::Transient(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))), MaybeTerminalSuccessOutcome::Terminate(event) => (PersistActions::SaveAndClose(event), Ok(None)), - MaybeTerminalSuccessOutcome::FatalAdvance(event, _next_state, error) => - (PersistActions::Save(event), Err(ApiError::Fatal(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) => @@ -494,7 +496,7 @@ where pub fn save

( self, persister: &P, - ) -> Result, PersistedError> + ) -> Result, PersistedError> where P: SessionPersister, { @@ -506,7 +508,7 @@ where pub async fn save_async

( self, persister: &P, - ) -> Result, PersistedError> + ) -> Result, PersistedError> where P: AsyncSessionPersister, Err: Send, diff --git a/payjoin/src/core/receive/v2/mod.rs b/payjoin/src/core/receive/v2/mod.rs index fee21a68e..8eb9ff395 100644 --- a/payjoin/src/core/receive/v2/mod.rs +++ b/payjoin/src/core/receive/v2/mod.rs @@ -2040,6 +2040,7 @@ pub mod test { #[test] fn process_error_response_fatal_with_fallback_enters_pending_fallback() -> Result<(), BoxError> { + let expected_tx = mock_fallback_tx(); let receiver = receiver(HasReplyableError { error_reply: mock_err(), fallback_tx: Some(mock_fallback_tx()), @@ -2052,8 +2053,10 @@ pub mod test { .process_error_response(&response, ctx) .save(&persister) .expect_err("fatal response should error"); + let pending_fallback = err.error_state().expect("pending fallback should be carried"); - assert!(err.api_error_ref().is_some()); + assert_eq!(pending_fallback.fallback_tx(), &expected_tx); + // assert!(err.api_error_ref().is_some()); assert_events(&persister, &[SessionEvent::ProtocolFailed], false); Ok(()) } From 2eedcd3d10915655571260cc1af2239b49ba014c Mon Sep 17 00:00:00 2001 From: Benalleng Date: Thu, 4 Jun 2026 09:31:14 -0400 Subject: [PATCH 10/10] Notify sender that receiver needs to cancel and broadcast fallback manually --- payjoin-cli/src/app/v2/mod.rs | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index 2a9a74918..8a1ff785e 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -1036,7 +1036,18 @@ impl App { }; if let Err(e) = session.process_error_response(&err_bytes, err_ctx).save(persister) { - return Err(anyhow!("Failed to process error response: {}", 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." + ); + } + None => return Err(anyhow!("Failed to process error response")), + } } Ok(())