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 c2558884c..8a1ff785e 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; @@ -78,6 +80,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", @@ -503,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, @@ -540,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, @@ -576,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(()); } @@ -690,6 +787,13 @@ impl App { self.handle_error(error, persister).await, ReceiveSession::Monitor(proposal) => self.monitor_payjoin_proposal(proposal, persister).await, + 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")), } }; @@ -932,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(()) 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(()) + } } 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/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-ffi/src/receive/mod.rs b/payjoin-ffi/src/receive/mod.rs index 47d9833f0..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))), + } } } }; } -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); +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_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,6 +262,7 @@ pub enum ReceiveSession { PayjoinProposal { inner: Arc }, HasReplyableError { inner: Arc }, Monitor { inner: Arc }, + ReceiverPendingFallback { inner: Arc }, Closed { inner: Arc }, } @@ -228,6 +293,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::ReceiverPendingFallback { inner: Arc::new(inner.into()) }, ReceiveSession::Closed(session_outcome) => Self::Closed { inner: Arc::new(session_outcome.into()) }, } @@ -1208,6 +1275,7 @@ pub struct PayjoinProposalTransition( payjoin::receive::v2::SessionEvent, payjoin::receive::v2::Receiver, payjoin::receive::ProtocolError, + payjoin::receive::v2::Receiver, >, >, >, @@ -1287,13 +1355,14 @@ impl From, payjoin::receive::ProtocolError, >, >, @@ -1306,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()))) } } @@ -1435,6 +1505,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 ReceiverPendingFallback( + Arc>, +); + +impl From> + for ReceiverPendingFallback +{ + fn from(value: payjoin::receive::v2::Receiver) -> Self { + Self(Arc::new(value)) + } +} + +impl From + for payjoin::receive::v2::Receiver +{ + fn from(value: ReceiverPendingFallback) -> Self { value.0.as_ref().clone() } +} + +#[uniffi::export] +impl ReceiverPendingFallback { + 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-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/persist.rs b/payjoin/src/core/persist.rs index c706a1367..969513120 100644 --- a/payjoin/src/core/persist.rs +++ b/payjoin/src/core/persist.rs @@ -361,102 +361,163 @@ where } } -/// 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>, -); +/// A transition that always results in a state transition. +pub struct NextStateTransition(AcceptNextState); -impl MaybeSuccessTransition -where - Err: std::error::Error, -{ - pub(crate) fn success(event: Event, success_value: SuccessValue) -> Self { - MaybeSuccessTransition(Ok(AcceptNextState(event, success_value))) +impl NextStateTransition { + pub(crate) fn success(event: Event, next_state: NextState) -> Self { + NextStateTransition(AcceptNextState(event, next_state)) } - pub(crate) fn transient(error: Err) -> Self { - MaybeSuccessTransition(Err(Rejection::transient(error))) + pub(crate) fn deconstruct(self) -> (PersistActions, NextState) { + let AcceptNextState(event, next_state) = self.0; + (PersistActions::Save(event), next_state) } - pub(crate) fn fatal(event: Event, error: Err) -> Self { - MaybeSuccessTransition(Err(Rejection::fatal(event, error))) + pub fn save

(self, persister: &P) -> Result + where + P: SessionPersister, + { + let (actions, next_state) = self.deconstruct(); + actions.execute(persister)?; + Ok(next_state) } - pub(crate) fn deconstruct( - self, - ) -> (PersistActions, Result>) { + 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 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 MaybeTerminalTransition { + pub(crate) fn advance(event: Event, next_state: NextState) -> Self { + Self(MaybeTerminalOutcome::Advance(AcceptNextState(event, next_state))) + } + + pub(crate) fn terminate(event: Event) -> Self { Self(MaybeTerminalOutcome::Terminate(event)) } + + pub(crate) fn deconstruct(self) -> (PersistActions, Option) { 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))), + 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, { - let (actions, outcome) = self.deconstruct(); - actions.execute(persister).map_err(InternalPersistedError::Storage)?; - Ok(outcome.map_err(InternalPersistedError::Api)?) + let (actions, next_state) = self.deconstruct(); + actions.execute(persister)?; + Ok(next_state) } pub async fn save_async

( self, persister: &P, - ) -> Result> + ) -> Result, P::InternalStorageError> where P: AsyncSessionPersister, - Err: Send, - SuccessValue: 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)?) + let (actions, next_state) = self.deconstruct(); + actions.execute_async(persister).await?; + Ok(next_state) } } -/// A transition that always results in a state transition. -pub struct NextStateTransition(AcceptNextState); +/// 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, +); -impl NextStateTransition { - pub(crate) fn success(event: Event, next_state: NextState) -> Self { - NextStateTransition(AcceptNextState(event, next_state)) +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))) } - 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(MaybeTerminalSuccessOutcome::Terminate(event)) } - pub fn save

(self, persister: &P) -> Result + 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)) + } + + #[allow(clippy::type_complexity)] + 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::FatalWithState(error, next_state))), + 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, next_state) = self.deconstruct(); - actions.execute(persister)?; - Ok(next_state) + 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 + pub async fn save_async

( + self, + persister: &P, + ) -> Result, PersistedError> where P: AsyncSessionPersister, + Err: Send, NextState: Send, Event: Send, { - let (actions, next_state) = self.deconstruct(); - actions.execute_async(persister).await?; - Ok(next_state) + let (actions, outcome) = self.deconstruct(); + actions.execute_async(persister).await.map_err(InternalPersistedError::Storage)?; + Ok(outcome.map_err(InternalPersistedError::Api)?) } } @@ -576,6 +637,19 @@ where /// Wrapper that marks the progression of a state machine pub struct AcceptNextState(Event, NextState); +enum MaybeTerminalOutcome { + Advance(AcceptNextState), + Terminate(Event), +} + +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 @@ -1086,44 +1160,119 @@ mod tests { } #[tokio::test] - async fn test_maybe_success_transition() { + async fn test_maybe_terminal_transition() { let event = InMemoryTestEvent("foo".to_string()); - let error_event = InMemoryTestEvent("error event".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(); - move || MaybeSuccessTransition::success(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(()), + 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(|| { - MaybeSuccessTransition::transient(InMemoryTestError {}) + 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![], + 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::Transient(InMemoryTestError {})) - .into(), + InternalPersistedError::Api(ApiError::Fatal(InMemoryTestError {})).into(), ), success: None, }, }, TestCase { make_transition: Box::new({ - let error_event = error_event.clone(); - move || MaybeSuccessTransition::fatal(error_event.clone(), InMemoryTestError {}) + let fatal_close_event = fatal_close_event.clone(); + move || { + MaybeTerminalSuccessTransition::<_, InMemoryTestState, InMemoryTestError>::fatal_terminate( + fatal_close_event.clone(), + InMemoryTestError {}, + ) + } }), expected_result: ExpectedResult { - events: vec![error_event.clone()], + events: vec![fatal_close_event.clone()], is_closed: true, error: Some( InternalPersistedError::Api(ApiError::Fatal(InMemoryTestError {})).into(), @@ -1131,6 +1280,24 @@ mod tests { 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); diff --git a/payjoin/src/core/receive/v2/mod.rs b/payjoin/src/core/receive/v2/mod.rs index 2c3f1bd5b..8eb9ff395 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, @@ -55,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; @@ -145,6 +147,7 @@ pub enum ReceiveSession { PayjoinProposal(Receiver), HasReplyableError(Receiver), Monitor(Receiver), + PendingFallback(Receiver), Closed(SessionOutcome), } @@ -198,29 +201,63 @@ 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)), - (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::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), @@ -231,72 +268,126 @@ impl ReceiveSession { } } -mod sealed { - pub trait State { - fn fallback_tx(&self) -> Option { None } +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 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, +) -> 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::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 {} impl State for super::Initialized {} + 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 {} - impl State for super::UncheckedOriginalPayload { - fn fallback_tx(&self) -> Option { - Some(self.original.psbt.clone().extract_tx_unchecked_fee_rate()) - } + pub trait FallbackTx: State { + fn fallback_tx(&self) -> bitcoin::Transaction; } - impl State for super::MaybeInputsOwned { - fn fallback_tx(&self) -> Option { - Some(self.original.psbt.clone().extract_tx_unchecked_fee_rate()) + impl FallbackTx for super::MaybeInputsOwned { + fn fallback_tx(&self) -> bitcoin::Transaction { + self.original.psbt.clone().extract_tx_unchecked_fee_rate() } } - impl State for super::MaybeInputsSeen { - fn fallback_tx(&self) -> Option { - Some(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 State for super::OutputsUnknown { - fn fallback_tx(&self) -> Option { - Some(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 State for super::WantsOutputs { - fn fallback_tx(&self) -> Option { - Some(self.inner.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 State for super::WantsInputs { - fn fallback_tx(&self) -> Option { - Some(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 State for super::WantsFeeRange { - fn fallback_tx(&self) -> Option { - Some(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 State for super::ProvisionalProposal { - fn fallback_tx(&self) -> Option { - Some(self.psbt_context.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 State for super::PayjoinProposal { - fn fallback_tx(&self) -> Option { - Some(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 State for super::HasReplyableError {} - - impl State for super::Monitor { - fn fallback_tx(&self) -> Option { - Some(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() } } } @@ -310,6 +401,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. /// @@ -340,17 +439,38 @@ 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.fallback_tx(); - TerminalTransition::new(SessionEvent::Closed(SessionOutcome::Cancel), fallback) +#[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), ()) + } +} + +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), ()) } } @@ -601,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 @@ -653,7 +780,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, @@ -733,7 +860,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, @@ -790,7 +920,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, @@ -839,6 +972,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()), @@ -849,7 +983,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, @@ -1183,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(), @@ -1194,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 { @@ -1217,9 +1363,26 @@ impl Receiver { #[derive(Debug, Clone, PartialEq)] pub struct HasReplyableError { error_reply: JsonReply, + fallback_tx: Option, } 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( @@ -1255,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)] @@ -1419,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; @@ -1471,6 +1649,46 @@ pub mod test { JsonReply::from(&res) } + pub(crate) fn mock_fallback_tx() -> bitcoin::Transaction { + 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 { @@ -1747,7 +1965,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(), }; @@ -1761,7 +1982,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(), }; @@ -1777,6 +2001,126 @@ 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 expected_tx = mock_fallback_tx(); + 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"); + let pending_fallback = err.error_state().expect("pending fallback should be carried"); + + assert_eq!(pending_fallback.fallback_tx(), &expected_tx); + // 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(); @@ -1879,52 +2223,304 @@ 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() }, None); - - // States with a fallback transaction - 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.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::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 { + 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] + 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:?}"), + } } } 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(); 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);