From aae2302d185e78dba0c95578ee086ded9744d199 Mon Sep 17 00:00:00 2001 From: xstoicunicornx Date: Tue, 14 Jul 2026 09:16:12 -0700 Subject: [PATCH 1/3] Encapsulate expiry handling in cancel functions Expiry was handled in six places with three different message shapes: bespoke finalize_expired_sender/receiver helpers, inline branches in resume_payjoins, cancel_*_session, handle_error, and long_poll_fallback. Delegate every expiry site to cancel_sender_session/cancel_receiver_session, which already know how to unwind every session typestate and both replay outcomes. Replay re-detects expiry from the event log, so the in-memory typestate can simply be dropped at the RelayPost::Expired sites. The cancel functions' expired-replay branches now share one finalize_expired_session helper that reports the fallback transaction when the event log has one and closes the session either way. Previously an expired session with no fallback (e.g. a receiver that never got an original payload) returned an error without closing, which would have made resume retry it forever. Also fix two copy-paste bugs in cancel_receiver_session: it closed with role "sender" and reported replay failures as sender sessions. close_failed_session now takes the Role enum so the mismatch cannot recur. --- payjoin-cli/src/app/v2/mod.rs | 187 ++++++++++++++-------------------- 1 file changed, 75 insertions(+), 112 deletions(-) diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index 43c74f0c1..913e7db57 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use anyhow::{anyhow, Context, Result}; use payjoin::bitcoin::consensus::encode::serialize_hex; -use payjoin::bitcoin::{Amount, FeeRate}; +use payjoin::bitcoin::{Amount, FeeRate, Transaction}; use payjoin::persist::{OptionalTransitionOutcome, SessionPersister}; use payjoin::receive::v2::{ replay_event_log as replay_receiver_event_log, HasReplyableError, Initialized, @@ -367,23 +367,17 @@ impl AppTrait for App { }), )); } - Err(e) => { - if e.is_expired() { - println!("Receiver session expired: {session_id}"); - match e.expiry_fallback_tx() { - Some(tx) => println!( - "Broadcast the original transaction manually:\n{}", - serialize_hex(tx) - ), - None => println!( - "No fallback transaction available for expired receiver session: {session_id}" - ), - } - } else { - tracing::error!("An error {:?} occurred while replaying session", e); - println!("Session failed to replay: {session_id} - {e}"); + Err(e) if e.is_expired() => { + if let Err(err) = self.cancel_receiver_session(session_id.clone(), true) { + tracing::error!( + "Failed to cancel expired receiver session {session_id}: {err:?}" + ); } - Self::close_failed_session(&recv_persister, &session_id, "receiver"); + } + Err(e) => { + tracing::error!("An error {:?} occurred while replaying session", e); + println!("Session failed to replay: {session_id} - {e}"); + Self::close_failed_session(&recv_persister, &session_id, Role::Receiver); } } } @@ -401,23 +395,17 @@ impl AppTrait for App { }), )); } - Err(e) => { - if e.is_expired() { - println!("Sender session expired: {session_id}"); - match e.expiry_fallback_tx() { - Some(tx) => println!( - "Broadcast the original transaction manually:\n{}", - serialize_hex(tx) - ), - None => println!( - "No fallback transaction available for expired sender session: {session_id}" - ), - } - } else { - tracing::error!("An error {:?} occurred while replaying session", e); - println!("Session failed to replay: {session_id} - {e}"); + Err(e) if e.is_expired() => { + if let Err(err) = self.cancel_sender_session(session_id.clone(), true) { + tracing::error!( + "Failed to cancel expired sender session {session_id}: {err:?}" + ); } - Self::close_failed_session(&sender_persister, &session_id, "sender"); + } + Err(e) => { + tracing::error!("An error {:?} occurred while replaying session", e); + println!("Session failed to replay: {session_id} - {e}"); + Self::close_failed_session(&sender_persister, &session_id, Role::Sender); } } } @@ -620,14 +608,12 @@ impl App { let (session, history) = match replay_sender_event_log(&persister) { Ok((session, history)) => (session, history), Err(e) if e.is_expired() => { - let tx = e.expiry_fallback_tx().cloned().ok_or_else(|| { - anyhow!("Expired sender session {session_id} has no fallback transaction") - })?; - println!( - "Session {session_id} expired. Broadcast the original transaction manually:\n{}", - serialize_hex(&tx) + Self::finalize_expired_session( + &persister, + &session_id, + Role::Sender, + e.expiry_fallback_tx(), ); - Self::close_failed_session(&persister, &session_id, "sender"); return Ok(()); } Err(e) => return Err(anyhow!("Failed to replay sender session {session_id}: {:?}", e)), @@ -675,17 +661,16 @@ impl App { let (session, history) = match replay_receiver_event_log(&persister) { Ok((session, history)) => (session, history), Err(e) if e.is_expired() => { - let tx = e.expiry_fallback_tx().cloned().ok_or_else(|| { - anyhow!("Expired receiver session {session_id} has no fallback transaction") - })?; - println!( - "Session {session_id} expired. Broadcast the original transaction manually:\n{}", - serialize_hex(&tx) + Self::finalize_expired_session( + &persister, + &session_id, + Role::Receiver, + e.expiry_fallback_tx(), ); - Self::close_failed_session(&persister, &session_id, "sender"); return Ok(()); } - Err(e) => return Err(anyhow!("Failed to replay sender session {session_id}: {:?}", e)), + Err(e) => + return Err(anyhow!("Failed to replay receiver session {session_id}: {:?}", e)), }; let pending: Receiver = match session { @@ -757,47 +742,41 @@ impl App { Ok(()) } - fn close_failed_session

(persister: &P, session_id: &SessionId, role: &str) + fn close_failed_session

(persister: &P, session_id: &SessionId, role: Role) where P: SessionPersister, { if let Err(close_err) = SessionPersister::close(persister) { - tracing::error!("Failed to close {} session {}: {:?}", role, session_id, close_err); + tracing::error!( + "Failed to close {} session {}: {:?}", + role.as_str(), + session_id, + close_err + ); } else { - tracing::debug!("Closed failed {} session: {}", role, session_id); + tracing::debug!("Closed failed {} session: {}", role.as_str(), session_id); } } - fn finalize_expired_sender( - &self, - pending: Sender, - persister: &SenderPersister, - ) -> Result<()> { - let tx = pending.fallback_tx(); - let session_id = persister.session_id(); - println!( - "Session {} expired. Broadcast the original transaction manually:\n{}", - session_id, - serialize_hex(tx) - ); - Self::close_failed_session(persister, &session_id, "sender"); - Ok(()) - } - - fn finalize_expired_receiver( - &self, - pending: Receiver, - persister: &ReceiverPersister, - ) -> Result<()> { - let tx = pending.fallback_tx(); - let session_id = persister.session_id(); - println!( - "Session {} expired. Broadcast the original transaction manually:\n{}", - session_id, - serialize_hex(tx) - ); - Self::close_failed_session(persister, &session_id, "receiver"); - Ok(()) + /// Report an expired session to the user, pointing at the fallback + /// transaction from the event log if one exists, and close the session + /// so it is not resumed again. + fn finalize_expired_session

( + persister: &P, + session_id: &SessionId, + role: Role, + fallback_tx: Option<&Transaction>, + ) where + P: SessionPersister, + { + match fallback_tx { + Some(tx) => println!( + "Session {session_id} expired. Broadcast the original transaction manually:\n{}", + serialize_hex(tx) + ), + None => println!("Session {session_id} expired. No fallback transaction available."), + } + Self::close_failed_session(persister, session_id, role); } async fn process_sender_session( @@ -839,8 +818,7 @@ impl App { match self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await? { RelayPost::Posted(resp, ctx) => (resp, ctx), RelayPost::Expired => - return self - .finalize_expired_sender(sender.cancel().save(persister)?, persister), + return self.cancel_sender_session(persister.session_id(), true), }; match sender.process_response(&response.bytes().await?, ctx).save(persister) { Ok(sender) => { @@ -868,8 +846,7 @@ impl App { match self.post_via_relay(|relay| sender.create_poll_request(relay)).await? { RelayPost::Posted(resp, ctx) => (resp, ctx), RelayPost::Expired => - return self - .finalize_expired_sender(sender.cancel().save(persister)?, persister), + return self.cancel_sender_session(persister.session_id(), true), }; let res = sender.clone().process_response(&response.bytes().await?, ctx).save(persister); @@ -909,7 +886,7 @@ impl App { RelayPost::Posted(resp, ctx) => (resp, ctx), RelayPost::Expired => { let session_id = persister.session_id(); - Self::close_failed_session(persister, &session_id, "receiver"); + self.cancel_receiver_session(session_id.clone(), true)?; return Err(anyhow!("Receiver session expired: {session_id}")); } }; @@ -1119,15 +1096,12 @@ impl App { persister: &ReceiverPersister, ) -> Result<()> { loop { - let (res, ohttp_ctx) = match self - .post_via_relay(|relay| proposal.create_post_request(relay)) - .await? - { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => - return self - .finalize_expired_receiver(proposal.cancel().save(persister)?, persister), - }; + let (res, ohttp_ctx) = + match self.post_via_relay(|relay| proposal.create_post_request(relay)).await? { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => + return self.cancel_receiver_session(persister.session_id(), true), + }; let payjoin_psbt = proposal.psbt().clone(); match proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister) { Ok(session) => { @@ -1208,23 +1182,12 @@ impl App { persister: &ReceiverPersister, ) -> Result<()> { loop { - let (err_response, err_ctx) = match self - .post_via_relay(|relay| session.create_error_request(relay)) - .await? - { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => match session.cancel().save(persister)? { - Some(pending) => return self.finalize_expired_receiver(pending, persister), - None => { - let session_id = persister.session_id(); - println!( - "Session {session_id} expired before the error reply could be sent. No fallback transaction available." - ); - Self::close_failed_session(persister, &session_id, "receiver"); - return Ok(()); - } - }, - }; + let (err_response, err_ctx) = + match self.post_via_relay(|relay| session.create_error_request(relay)).await? { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => + return self.cancel_receiver_session(persister.session_id(), true), + }; let err_bytes = match err_response.bytes().await { Ok(bytes) => bytes, Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)), From 3e02ea70437721f262daa8336114ba16f4fc942f Mon Sep 17 00:00:00 2001 From: xstoicunicornx Date: Tue, 14 Jul 2026 10:37:34 -0700 Subject: [PATCH 2/3] Prefix session output with role and session id Session-scoped messages were inconsistent: some embedded the session id mid-sentence, some named the role, and many ("Posted Original PSBT...", "No response yet.") had neither, making interleaved output from concurrently resumed sessions impossible to attribute. Route all session-scoped printing through a print_session helper that prefixes every line with "[ ]", padding the role to the widest form so ids and messages align in columns. A SessionPrint extension trait implemented for the sender and receiver persisters lets call sites write persister.print(msg): the persister already carries the session id, and its type determines the role, so a message can never claim the wrong one. The resume completion loop and finalize_expired_session print via the free helper since they do not hold a typed persister at that point. Raw payloads (BIP21 URI, fallback transaction hex) stay on their own unprefixed lines for clean copy-paste. The "Payjoin sent" print moves from the shared process_pj_response to its call sites so v2 can label it while v1 output is unchanged. Receiver interrupt messaging now gives the same resume/cancel instructions as the sender's. Update e2e assertions that parsed the old formats: session ids are read from the bracket prefix, and txid extraction uses the last token rather than a fixed index. --- payjoin-cli/src/app/mod.rs | 2 - payjoin-cli/src/app/v1.rs | 3 +- payjoin-cli/src/app/v2/mod.rs | 216 ++++++++++++++++++++-------------- payjoin-cli/tests/e2e.rs | 24 ++-- 4 files changed, 140 insertions(+), 105 deletions(-) diff --git a/payjoin-cli/src/app/mod.rs b/payjoin-cli/src/app/mod.rs index b10cf6adb..a0174e187 100644 --- a/payjoin-cli/src/app/mod.rs +++ b/payjoin-cli/src/app/mod.rs @@ -67,8 +67,6 @@ pub trait App: Send + Sync { let tx = signed.extract_tx()?; let txid = self.wallet().broadcast_tx(&tx)?; - - println!("Payjoin sent. TXID: {txid}"); Ok(txid) } } diff --git a/payjoin-cli/src/app/v1.rs b/payjoin-cli/src/app/v1.rs index 5fdbaf0d8..a7082005d 100644 --- a/payjoin-cli/src/app/v1.rs +++ b/payjoin-cli/src/app/v1.rs @@ -102,7 +102,8 @@ impl AppTrait for App { } }; - self.process_pj_response(psbt)?; + let txid = self.process_pj_response(psbt)?; + println!("Payjoin sent. TXID: {txid}"); Ok(()) } diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index 913e7db57..cb2464864 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -122,6 +122,7 @@ fn print_header() { println!("{: ]`, +/// with the role padded so sender and receiver messages align in columns. +/// The prefix lets output interleaved from concurrently resumed sessions be +/// attributed to its session. +fn print_session(role: Role, id: &SessionId, msg: impl fmt::Display) { + // Pad the role to the widest form, `Receiver`. + const W: usize = "Receiver".len(); + println!("[{: { session_id: SessionId, role: Role, @@ -236,7 +265,8 @@ impl AppTrait for App { } }; - self.process_pj_response(psbt)?; + let txid = self.process_pj_response(psbt)?; + println!("Payjoin sent. TXID: {txid}"); Ok(()) } PjParam::V2(pj_param) => { @@ -259,7 +289,7 @@ impl AppTrait for App { let psbt = self.create_original_psbt(&address, amount, fee_rate)?; let persister = SenderPersister::new(self.db.clone(), bip21, receiver_pubkey)?; - println!("Send session established: {}", persister.session_id()); + persister.print("Session established"); let sender = SenderBuilder::from_parts(psbt, pj_param, &address, Some(amount)) .build_recommended(fee_rate)? @@ -275,16 +305,16 @@ impl AppTrait for App { Ok(()) => return Ok(()), Err(err) => { let id = persister.session_id(); - println!("Session {id} failed. Run `payjoin-cli cancel {id}` to cancel and broadcast the original transaction."); + persister.print(format_args!("Session failed. Run `payjoin-cli cancel {id}` to cancel and broadcast the original transaction.")); return Err(err); } } }, _ = interrupt.changed() => { let session_id = persister.session_id(); - println!( - "Session {session_id} interrupted. Call `payjoin-cli resume --session-id {session_id}` again to resume, `payjoin-cli resume` to resume all sessions, or `payjoin-cli cancel {session_id}` to cancel and broadcast the original transaction." - ); + persister.print(format_args!( + "Session interrupted. Call `payjoin-cli resume --session-id {session_id}` again to resume, `payjoin-cli resume` to resume all sessions, or `payjoin-cli cancel {session_id}` to cancel and broadcast the original transaction." + )); return Err(anyhow!("Interrupted")) } } @@ -322,10 +352,10 @@ impl AppTrait for App { receiver_builder = receiver_builder.with_expiration(expiration); } let session = receiver_builder.build().save(&persister)?; - println!("Receive session established: {}", persister.session_id()); + persister.print("Session established"); let pj_uri = session.pj_uri(); - println!("Request Payjoin by sharing this Payjoin Uri:"); + persister.print("Request Payjoin by sharing this Payjoin Uri:"); println!("{pj_uri}"); self.process_receiver_session(ReceiveSession::Initialized(session.clone()), &persister) @@ -350,7 +380,7 @@ impl AppTrait for App { return Ok(()); } - let mut tasks: Vec<(String, tokio::task::JoinHandle>)> = Vec::new(); + let mut tasks: Vec<((Role, SessionId), tokio::task::JoinHandle>)> = Vec::new(); // Process receiver sessions for session_id in recv_session_ids { @@ -359,7 +389,7 @@ impl AppTrait for App { match replay_receiver_event_log(&recv_persister) { Ok((receiver_state, _)) => { tasks.push(( - session_id.to_string(), + (Role::Receiver, session_id), tokio::spawn(async move { self_clone .process_receiver_session(receiver_state, &recv_persister) @@ -376,7 +406,7 @@ impl AppTrait for App { } Err(e) => { tracing::error!("An error {:?} occurred while replaying session", e); - println!("Session failed to replay: {session_id} - {e}"); + recv_persister.print(format_args!("Session failed to replay: {e}")); Self::close_failed_session(&recv_persister, &session_id, Role::Receiver); } } @@ -389,7 +419,7 @@ impl AppTrait for App { Ok((sender_state, _)) => { let self_clone = self.clone(); tasks.push(( - session_id.clone().to_string(), + (Role::Sender, session_id), tokio::spawn(async move { self_clone.process_sender_session(sender_state, &sender_persister).await }), @@ -404,7 +434,7 @@ impl AppTrait for App { } Err(e) => { tracing::error!("An error {:?} occurred while replaying session", e); - println!("Session failed to replay: {session_id} - {e}"); + sender_persister.print(format_args!("Session failed to replay: {e}")); Self::close_failed_session(&sender_persister, &session_id, Role::Sender); } } @@ -414,16 +444,16 @@ impl AppTrait for App { tokio::select! { _ = async { - for (session_id, task) in tasks { + for ((role, id), task) in tasks { match task.await { Ok(Ok(())) => { - println!("Session {session_id} completed."); + print_session(role, &id, "Session completed."); } Ok(Err(e)) => { - println!("Session {session_id} error: {e:#}"); + print_session(role, &id, format_args!("Session error: {e:#}")); } Err(e) => { - println!("Session {session_id} panicked or was cancelled: {e:?}"); + print_session(role, &id, format_args!("Session panicked or was cancelled: {e:?}")); } } } @@ -625,32 +655,30 @@ impl App { SendSession::PendingFallback(sender) => sender, SendSession::Closed(SenderSessionOutcome::Success(proposal)) => { let txid = proposal.extract_tx_unchecked_fee_rate().compute_txid(); - println!( - "Session {session_id} already produced payjoin transaction {txid}. \ + persister.print(format_args!( + "Session already produced payjoin transaction {txid}. \ Cannot cancel a completed session." - ); + )); return Ok(()); } SendSession::Closed(SenderSessionOutcome::Aborted) => { - println!( - "Session {session_id} was already cancelled. Broadcast the original transaction manually:\n{}", - serialize_hex(&history.fallback_tx()) + persister.print( + "Session was already cancelled. Broadcast the original transaction manually:", ); + println!("{}", serialize_hex(&history.fallback_tx())); return Ok(()); } }; if no_broadcast { - println!( - "Session {session_id} cancelled. Broadcast the original transaction manually:\n{}", - serialize_hex(pending.fallback_tx()) - ); + persister.print("Session cancelled. Broadcast the original transaction manually:"); + println!("{}", serialize_hex(pending.fallback_tx())); } else { self.wallet().broadcast_tx(pending.fallback_tx())?; - println!( + persister.print(format_args!( "Broadcasted fallback transaction txid: {}", pending.fallback_tx().compute_txid() - ); + )); } pending.close().save(&persister)?; Ok(()) @@ -676,12 +704,12 @@ impl App { let pending: Receiver = match session { ReceiveSession::Initialized(receiver) => { receiver.cancel().save(&persister)?; - println!("Session {session_id} cancelled. No fallback transaction to broadcast."); + persister.print("Session 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."); + persister.print("Session cancelled. No fallback transaction to broadcast."); return Ok(()); } ReceiveSession::MaybeInputsOwned(receiver) => receiver.cancel().save(&persister)?, @@ -693,50 +721,47 @@ impl App { 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::HasReplyableError(receiver) => + match receiver.cancel().save(&persister)? { + Some(pending) => pending, + None => { + persister.print("Session 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."); + persister.print("Session already completed successfully. Cannot cancel."); return Ok(()); } ReceiveSession::Closed(ReceiverSessionOutcome::Aborted) => { match history.fallback_tx() { - Some(tx) => println!( - "Session {session_id} was already cancelled. Broadcast the fallback transaction manually:\n{}", - serialize_hex(&tx) - ), - None => println!( - "Session {session_id} is already closed. No fallback transaction available." - ), + Some(tx) => { + persister.print( + "Session was already cancelled. Broadcast the fallback transaction manually:", + ); + println!("{}", serialize_hex(&tx)); + } + None => persister + .print("Session is already closed. No fallback transaction available."), } return Ok(()); } }; if no_broadcast { - println!( - "Session {session_id} cancelled. Broadcast the fallback transaction manually:\n{}", - serialize_hex(pending.fallback_tx()) - ); + persister.print("Session cancelled. Broadcast the fallback transaction manually:"); + println!("{}", serialize_hex(pending.fallback_tx())); } else { self.wallet().broadcast_tx(pending.fallback_tx())?; - println!( + persister.print(format_args!( "Broadcasted fallback transaction txid: {}", pending.fallback_tx().compute_txid() - ); + )); } pending.close().save(&persister)?; Ok(()) @@ -770,11 +795,19 @@ impl App { P: SessionPersister, { match fallback_tx { - Some(tx) => println!( - "Session {session_id} expired. Broadcast the original transaction manually:\n{}", - serialize_hex(tx) + Some(tx) => { + print_session( + role, + session_id, + "Session expired. Broadcast the original transaction manually:", + ); + println!("{}", serialize_hex(tx)); + } + None => print_session( + role, + session_id, + "Session expired. No fallback transaction available.", ), - None => println!("Session {session_id} expired. No fallback transaction available."), } Self::close_failed_session(persister, session_id, role); } @@ -790,18 +823,19 @@ impl App { SendSession::PollingForProposal(context) => self.get_proposed_payjoin_psbt(context, persister).await?, SendSession::Closed(SenderSessionOutcome::Success(proposal)) => { - self.process_pj_response(proposal)?; + let txid = self.process_pj_response(proposal)?; + persister.print(format_args!("Payjoin sent. TXID: {txid}")); return Ok(()); } SendSession::Closed(SenderSessionOutcome::Aborted) => { - println!("Session is closed. Nothing left to do"); + persister.print("Session is closed. Nothing left to do"); return Ok(()); } SendSession::PendingFallback(_) => { let id = persister.session_id(); - println!( - "Session {id} was cancelled. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction." - ); + persister.print(format_args!( + "Session was cancelled. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction." + )); return Ok(()); } } @@ -822,7 +856,7 @@ impl App { }; match sender.process_response(&response.bytes().await?, ctx).save(persister) { Ok(sender) => { - println!("Posted Original PSBT..."); + persister.print("Posted Original PSBT..."); return self.get_proposed_payjoin_psbt(sender, persister).await; } Err(e) if e.is_transient() => { @@ -852,12 +886,13 @@ impl App { sender.clone().process_response(&response.bytes().await?, ctx).save(persister); match res { Ok(OptionalTransitionOutcome::Progress(psbt)) => { - println!("Proposal received. Processing..."); - self.process_pj_response(psbt)?; + persister.print("Proposal received. Processing..."); + let txid = self.process_pj_response(psbt)?; + persister.print(format_args!("Payjoin sent. TXID: {txid}")); return Ok(()); } Ok(OptionalTransitionOutcome::Stasis(current_state)) => { - println!("No response yet."); + persister.print("No response yet."); sender = current_state; } Err(e) if e.is_transient() => { @@ -866,7 +901,7 @@ impl App { tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; } Err(re) => { - println!("{re}"); + persister.print(&re); tracing::debug!("{re:?}"); return Err(anyhow!("Response error").context(re)); } @@ -880,7 +915,7 @@ impl App { persister: &ReceiverPersister, ) -> Result> { loop { - println!("Polling receive request..."); + persister.print("Polling receive request..."); let (ohttp_response, context) = match self.post_via_relay(|relay| session.create_poll_request(relay)).await? { RelayPost::Posted(resp, ctx) => (resp, ctx), @@ -895,7 +930,9 @@ impl App { .save(persister); match state_transition { Ok(OptionalTransitionOutcome::Progress(next_state)) => { - println!("Got a request from the sender. Responding with a Payjoin proposal."); + persister.print( + "Got a request from the sender. Responding with a Payjoin proposal.", + ); return Ok(next_state); } Ok(OptionalTransitionOutcome::Stasis(current_state)) => { @@ -944,9 +981,9 @@ impl App { 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." - ); + persister.print(format_args!( + "Session was cancelled. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction." + )); return Ok(()); } ReceiveSession::Closed(_) => return Err(anyhow!("Session closed")), @@ -965,7 +1002,10 @@ impl App { let receiver = tokio::select! { res = self.long_poll_fallback(session, persister) => res, _ = interrupt.changed() => { - println!("Interrupted. Call the `resume` command to resume all sessions."); + let session_id = persister.session_id(); + persister.print(format_args!( + "Session interrupted. Call `payjoin-cli resume --session-id {session_id}` again to resume, `payjoin-cli resume` to resume all sessions, or `payjoin-cli cancel {session_id}` to cancel the session." + )); return Err(anyhow!("Interrupted")); } }?; @@ -986,7 +1026,9 @@ impl App { }) .save(persister)?; - println!("Fallback transaction received. Consider broadcasting this to get paid if the Payjoin fails:"); + persister.print( + "Fallback transaction received. Consider broadcasting this to get paid if the Payjoin fails:", + ); println!("{}", serialize_hex(&proposal.extract_tx_to_schedule_broadcast())); self.check_inputs_not_owned(proposal, persister).await } @@ -1105,10 +1147,10 @@ impl App { let payjoin_psbt = proposal.psbt().clone(); match proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister) { Ok(session) => { - println!( + persister.print(format_args!( "Response successful. Watch mempool for successful Payjoin. TXID: {}", payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid() - ); + )); return self.monitor_payjoin_proposal(session, persister).await; } Err(e) if e.is_transient() => { @@ -1148,7 +1190,7 @@ impl App { match check_result { Ok(OptionalTransitionOutcome::Progress(())) => { - println!("Payjoin transaction detected in the mempool!"); + persister.print("Payjoin transaction detected in the mempool!"); return Ok(()); } Ok(OptionalTransitionOutcome::Stasis(current_state)) => { @@ -1195,11 +1237,10 @@ impl App { match session.process_error_response(&err_bytes, err_ctx).save(persister) { Ok(Some(pending)) => { - let session_id = persister.session_id(); - println!( - "Session {session_id} delivered error reply. Broadcast the fallback transaction manually:\n{}", - serialize_hex(pending.fallback_tx()) - ); + persister.print( + "Session delivered error reply. Broadcast the fallback transaction manually:", + ); + println!("{}", serialize_hex(pending.fallback_tx())); return Ok(()); } Ok(None) => return Ok(()), @@ -1214,11 +1255,10 @@ impl App { } match e.fatal_state() { Some(pending) => { - let session_id = persister.session_id(); - println!( - "Session {session_id} failed to deliver error reply. Broadcast the fallback transaction manually:\n{}", - serialize_hex(pending.fallback_tx()) + persister.print( + "Session failed to deliver error reply. Broadcast the fallback transaction manually:", ); + println!("{}", serialize_hex(pending.fallback_tx())); return Ok(()); } None => return Err(anyhow!("Failed to process error response")), diff --git a/payjoin-cli/tests/e2e.rs b/payjoin-cli/tests/e2e.rs index 0bc45c52c..a073d7be6 100644 --- a/payjoin-cli/tests/e2e.rs +++ b/payjoin-cli/tests/e2e.rs @@ -38,14 +38,15 @@ mod e2e { /// Helper function to extract the first session id from history stdout async fn get_session_id_from_role(mut cli_role: tokio::process::Child) -> String { let mut stdout = cli_role.stdout.take().expect("failed to take stdout of child process"); + // Session-scoped lines are prefixed with `[ ]`. let session_id = - wait_for_stdout_match(&mut stdout, |line| line.contains("session established: ")) + wait_for_stdout_match(&mut stdout, |line| line.contains("Session established")) .await .expect("payjoin-cli should output a SessionId on session startup") - .split_terminator(":") + .split_whitespace() .nth(1) .expect("could not split stdout") - .trim() + .trim_end_matches(']') .to_string(); terminate(cli_role).await.expect("Failed to kill payjoin-cli"); @@ -537,9 +538,7 @@ mod e2e { let timeout = tokio::time::Duration::from_secs(10); let res = tokio::time::timeout( timeout, - wait_for_stdout_match(&mut stdout, |line| { - line.starts_with("Session") && line.ends_with("completed.") - }), + wait_for_stdout_match(&mut stdout, |line| line.ends_with("Session completed.")), ) .await?; @@ -554,9 +553,7 @@ mod e2e { let timeout = tokio::time::Duration::from_secs(10); let res = tokio::time::timeout( timeout, - wait_for_stdout_match(&mut stdout, |line| { - line.starts_with("Session") && line.ends_with("completed.") - }), + wait_for_stdout_match(&mut stdout, |line| line.ends_with("Session completed.")), ) .await?; terminate(cli_resumer).await.expect("Failed to kill payjoin-cli"); @@ -841,7 +838,7 @@ mod e2e { .await?; terminate(cli_cancel).await.expect("Failed to kill payjoin-cli cancel"); let subcommand_output = broadcast_line.expect("cancel should broadcast fallback tx"); - let fallback_txid = subcommand_output.split_whitespace().nth(4).unwrap_or(""); + let fallback_txid = subcommand_output.split_whitespace().last().unwrap_or(""); let fallback_txid = Txid::from_str(fallback_txid).expect("valid txid"); assert!( @@ -883,8 +880,7 @@ mod e2e { .expect("Failed to kill payjoin-cli cancel on closed session"); assert!( - cancel_again_output - .contains(&format!("Session {session_id} was already cancelled")), + cancel_again_output.contains(&format!("[Sender {session_id}] Session was already cancelled")), "cancel on closed session should reference the session id and report it is already closed; got: {cancel_again_output}" ); assert!( @@ -995,7 +991,7 @@ mod e2e { cancel_line.expect("cancel should report no fallback transaction"); assert!( - subcommand_output.contains(&format!("Session {session_id} cancelled")), + subcommand_output.contains(&format!("[Receiver {session_id}] Session cancelled")), "cancel should reference the cancelled session id" ); @@ -1037,7 +1033,7 @@ mod e2e { .expect("cancel on closed session should report it is already closed"); assert!( - cancel_again_output.contains(&format!("Session {session_id} is already closed")), + cancel_again_output.contains(&format!("[Receiver {session_id}] Session is already closed")), "cancel on closed session should reference the session id and report it is already closed" ); From 3b3154d5073e0210b4e46cfe451681bcd9c4d47e Mon Sep 17 00:00:00 2001 From: xstoicunicornx Date: Tue, 14 Jul 2026 11:12:59 -0700 Subject: [PATCH 3/3] Drive CLI sessions with state-machine loops Replace the loops in the v2 sender and receiver flows with a driver: each step function performs one state transition and returns the next SendSession or ReceiveSession state, and process_sender_session/process_receiver_session loop over dispatch until the session terminates. Wrap process_receiver_session in a select against the interrupt channel in receive_payjoin, mirroring how send_payjoin guards the sender driver. This is safe at any phase: transitions persist to the event log with no await between transition and save, so an interrupted session resumes from its last saved state via the resume command. The interrupt message now includes the session id, and read_from_directory becomes a plain single-step function like the other state handlers. --- payjoin-cli/src/app/v2/mod.rs | 467 +++++++++++++++++----------------- payjoin-cli/tests/e2e.rs | 43 +--- 2 files changed, 240 insertions(+), 270 deletions(-) diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index cb2464864..d7d4054ed 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -323,6 +323,7 @@ impl AppTrait for App { } } + #[allow(clippy::incompatible_msrv)] async fn receive_payjoin(&self, amount: Amount) -> Result<()> { let address = self.wallet().get_new_address()?; let persister = ReceiverPersister::new(self.db.clone())?; @@ -358,8 +359,23 @@ impl AppTrait for App { persister.print("Request Payjoin by sharing this Payjoin Uri:"); println!("{pj_uri}"); - self.process_receiver_session(ReceiveSession::Initialized(session.clone()), &persister) - .await?; + // Interrupting is safe at any phase: every transition is saved to the + // event log before the next await, so the session resumes from its + // last saved state via the resume command. + let mut interrupt = self.interrupt.clone(); + tokio::select! { + res = self.process_receiver_session( + ReceiveSession::Initialized(session.clone()), + &persister, + ) => res?, + _ = interrupt.changed() => { + let session_id = persister.session_id(); + persister.print(format_args!( + "Session interrupted. Call `payjoin-cli resume --session-id {session_id}` again to resume, `payjoin-cli resume` to resume all sessions, or `payjoin-cli cancel {session_id}` to cancel the session." + )); + return Err(anyhow!("Interrupted")); + } + } Ok(()) } @@ -812,173 +828,140 @@ impl App { Self::close_failed_session(persister, session_id, role); } + /// Drive the sender state machine until the session terminates. Each + /// step function performs one state transition and returns the next + /// state; the driver stops on the terminal `Closed` and + /// `PendingFallback` states. async fn process_sender_session( &self, - session: SendSession, + mut session: SendSession, persister: &SenderPersister, ) -> Result<()> { - match session { - SendSession::WithReplyKey(context) => - self.post_original_proposal(context, persister).await?, - SendSession::PollingForProposal(context) => - self.get_proposed_payjoin_psbt(context, persister).await?, - SendSession::Closed(SenderSessionOutcome::Success(proposal)) => { - let txid = self.process_pj_response(proposal)?; - persister.print(format_args!("Payjoin sent. TXID: {txid}")); - return Ok(()); - } - SendSession::Closed(SenderSessionOutcome::Aborted) => { - persister.print("Session is closed. Nothing left to do"); - return Ok(()); - } - SendSession::PendingFallback(_) => { - let id = persister.session_id(); - persister.print(format_args!( - "Session was cancelled. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction." - )); - return Ok(()); - } + loop { + session = match session { + SendSession::WithReplyKey(context) => + self.post_original_proposal(context, persister).await?, + SendSession::PollingForProposal(context) => + self.get_proposed_payjoin_psbt(context, persister).await?, + SendSession::Closed(SenderSessionOutcome::Success(proposal)) => { + let txid = self.process_pj_response(proposal)?; + persister.print(format_args!("Payjoin sent. TXID: {txid}")); + return Ok(()); + } + SendSession::Closed(SenderSessionOutcome::Aborted) => return Ok(()), + SendSession::PendingFallback(_) => { + let id = persister.session_id(); + persister.print(format_args!( + "Session was cancelled. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction." + )); + return Ok(()); + } + }; } - Ok(()) } async fn post_original_proposal( &self, - mut sender: Sender, + sender: Sender, persister: &SenderPersister, - ) -> Result<()> { - loop { - let (response, ctx) = - match self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => - return self.cancel_sender_session(persister.session_id(), true), - }; - match sender.process_response(&response.bytes().await?, ctx).save(persister) { - Ok(sender) => { - persister.print("Posted Original PSBT..."); - return self.get_proposed_payjoin_psbt(sender, persister).await; + ) -> Result { + let (response, ctx) = + match self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await? { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_sender_session(persister.session_id(), true)?; + return Ok(SendSession::Closed(SenderSessionOutcome::Aborted)); } - Err(e) if e.is_transient() => { - tracing::debug!("Transient error posting original proposal, retrying: {e:?}"); - sender = e.transient_state().expect("transient error carries current state"); - tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; - } - Err(e) => return Err(e.into()), + }; + match sender.process_response(&response.bytes().await?, ctx).save(persister) { + Ok(sender) => { + persister.print("Posted Original PSBT..."); + Ok(SendSession::PollingForProposal(sender)) + } + Err(e) if e.is_transient() => { + tracing::debug!("Transient error posting original proposal, retrying: {e:?}"); + let sender = e.transient_state().expect("transient error carries current state"); + tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; + Ok(SendSession::WithReplyKey(sender)) } + Err(e) => Err(e.into()), } } async fn get_proposed_payjoin_psbt( &self, - mut sender: Sender, + sender: Sender, persister: &SenderPersister, - ) -> Result<()> { - // Long poll until we get a response - loop { - let (response, ctx) = - match self.post_via_relay(|relay| sender.create_poll_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => - return self.cancel_sender_session(persister.session_id(), true), - }; - let res = - sender.clone().process_response(&response.bytes().await?, ctx).save(persister); - match res { - Ok(OptionalTransitionOutcome::Progress(psbt)) => { - persister.print("Proposal received. Processing..."); - let txid = self.process_pj_response(psbt)?; - persister.print(format_args!("Payjoin sent. TXID: {txid}")); - return Ok(()); - } - Ok(OptionalTransitionOutcome::Stasis(current_state)) => { - persister.print("No response yet."); - sender = current_state; - } - Err(e) if e.is_transient() => { - tracing::debug!("Transient error polling for proposal, retrying: {e:?}"); - sender = e.transient_state().expect("transient error carries current state"); - tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; - } - Err(re) => { - persister.print(&re); - tracing::debug!("{re:?}"); - return Err(anyhow!("Response error").context(re)); + ) -> Result { + let (response, ctx) = + match self.post_via_relay(|relay| sender.create_poll_request(relay)).await? { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_sender_session(persister.session_id(), true)?; + return Ok(SendSession::Closed(SenderSessionOutcome::Aborted)); } + }; + let res = sender.clone().process_response(&response.bytes().await?, ctx).save(persister); + match res { + Ok(OptionalTransitionOutcome::Progress(psbt)) => { + persister.print("Proposal received. Processing..."); + Ok(SendSession::Closed(SenderSessionOutcome::Success(psbt))) } - } - } - - async fn long_poll_fallback( - &self, - mut session: Receiver, - persister: &ReceiverPersister, - ) -> Result> { - loop { - persister.print("Polling receive request..."); - let (ohttp_response, context) = - match self.post_via_relay(|relay| session.create_poll_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => { - let session_id = persister.session_id(); - self.cancel_receiver_session(session_id.clone(), true)?; - return Err(anyhow!("Receiver session expired: {session_id}")); - } - }; - let state_transition = session - .process_response(ohttp_response.bytes().await?.to_vec().as_slice(), context) - .save(persister); - match state_transition { - Ok(OptionalTransitionOutcome::Progress(next_state)) => { - persister.print( - "Got a request from the sender. Responding with a Payjoin proposal.", - ); - return Ok(next_state); - } - Ok(OptionalTransitionOutcome::Stasis(current_state)) => { - session = current_state; - } - Err(e) if e.is_transient() => { - tracing::debug!("Transient error polling for request, retrying: {e:?}"); - session = e.transient_state().expect("transient error carries current state"); - tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; - } - Err(e) => return Err(e.into()), + Ok(OptionalTransitionOutcome::Stasis(current_state)) => { + persister.print("No response yet."); + Ok(SendSession::PollingForProposal(current_state)) + } + Err(e) if e.is_transient() => { + tracing::debug!("Transient error polling for proposal, retrying: {e:?}"); + let sender = e.transient_state().expect("transient error carries current state"); + tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; + Ok(SendSession::PollingForProposal(sender)) + } + Err(re) => { + persister.print(&re); + tracing::debug!("{re:?}"); + Err(anyhow!("Response error").context(re)) } } } + /// Drive the receiver state machine until the session terminates. Each + /// step function performs one state transition and returns the next + /// state; the driver stops on the terminal `Closed` and + /// `PendingFallback` states. async fn process_receiver_session( &self, - session: ReceiveSession, + mut session: ReceiveSession, persister: &ReceiverPersister, ) -> Result<()> { - let res = { - match session { + loop { + session = match session { ReceiveSession::Initialized(proposal) => - self.read_from_directory(proposal, persister).await, + self.read_from_directory(proposal, persister).await?, ReceiveSession::UncheckedOriginalPayload(proposal) => - self.check_proposal(proposal, persister).await, + self.check_proposal(proposal, persister)?, ReceiveSession::MaybeInputsOwned(proposal) => - self.check_inputs_not_owned(proposal, persister).await, + self.check_inputs_not_owned(proposal, persister)?, ReceiveSession::MaybeInputsSeen(proposal) => - self.check_no_inputs_seen_before(proposal, persister).await, + self.check_no_inputs_seen_before(proposal, persister)?, ReceiveSession::OutputsUnknown(proposal) => - self.identify_receiver_outputs(proposal, persister).await, + self.identify_receiver_outputs(proposal, persister)?, ReceiveSession::WantsOutputs(proposal) => - self.commit_outputs(proposal, persister).await, + self.commit_outputs(proposal, persister)?, ReceiveSession::WantsInputs(proposal) => - self.contribute_inputs(proposal, persister).await, + self.contribute_inputs(proposal, persister)?, ReceiveSession::WantsFeeRange(proposal) => - self.apply_fee_range(proposal, persister).await, + self.apply_fee_range(proposal, persister)?, ReceiveSession::ProvisionalProposal(proposal) => - self.finalize_proposal(proposal, persister).await, + self.finalize_proposal(proposal, persister)?, ReceiveSession::PayjoinProposal(proposal) => - self.send_payjoin_proposal(proposal, persister).await, + self.send_payjoin_proposal(proposal, persister).await?, ReceiveSession::HasReplyableError(error) => - self.handle_error(error, persister).await, - ReceiveSession::Monitor(proposal) => - self.monitor_payjoin_proposal(proposal, persister).await, + self.handle_error(error, persister).await?, + ReceiveSession::Monitor(proposal) => { + self.monitor_payjoin_proposal(proposal, persister).await?; + return Ok(()); + } ReceiveSession::PendingFallback(_) => { let id = persister.session_id(); persister.print(format_args!( @@ -986,37 +969,52 @@ impl App { )); return Ok(()); } - ReceiveSession::Closed(_) => return Err(anyhow!("Session closed")), - } - }; - res + ReceiveSession::Closed(_) => return Ok(()), + }; + } } - #[allow(clippy::incompatible_msrv)] + /// Poll the directory once for the sender's original proposal. async fn read_from_directory( &self, session: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { - let mut interrupt = self.interrupt.clone(); - let receiver = tokio::select! { - res = self.long_poll_fallback(session, persister) => res, - _ = interrupt.changed() => { - let session_id = persister.session_id(); - persister.print(format_args!( - "Session interrupted. Call `payjoin-cli resume --session-id {session_id}` again to resume, `payjoin-cli resume` to resume all sessions, or `payjoin-cli cancel {session_id}` to cancel the session." - )); - return Err(anyhow!("Interrupted")); + ) -> Result { + persister.print("Polling receive request..."); + let (ohttp_response, context) = + match self.post_via_relay(|relay| session.create_poll_request(relay)).await? { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_receiver_session(persister.session_id(), true)?; + return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); + } + }; + let state_transition = session + .process_response(ohttp_response.bytes().await?.to_vec().as_slice(), context) + .save(persister); + match state_transition { + Ok(OptionalTransitionOutcome::Progress(next_state)) => { + persister + .print("Got a request from the sender. Responding with a Payjoin proposal."); + Ok(ReceiveSession::UncheckedOriginalPayload(next_state)) + } + Ok(OptionalTransitionOutcome::Stasis(current_state)) => + Ok(ReceiveSession::Initialized(current_state)), + Err(e) if e.is_transient() => { + tracing::debug!("Transient error polling for request, retrying: {e:?}"); + let session = e.transient_state().expect("transient error carries current state"); + tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; + Ok(ReceiveSession::Initialized(session)) } - }?; - self.check_proposal(receiver, persister).await + Err(e) => Err(e.into()), + } } - async fn check_proposal( + fn check_proposal( &self, proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { let wallet = self.wallet(); let proposal = proposal .check_broadcast_suitability(None, |tx| { @@ -1030,14 +1028,14 @@ impl App { "Fallback transaction received. Consider broadcasting this to get paid if the Payjoin fails:", ); println!("{}", serialize_hex(&proposal.extract_tx_to_schedule_broadcast())); - self.check_inputs_not_owned(proposal, persister).await + Ok(ReceiveSession::MaybeInputsOwned(proposal)) } - async fn check_inputs_not_owned( + fn check_inputs_not_owned( &self, proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { let wallet = self.wallet(); let proposal = proposal .check_inputs_not_owned(&mut |input| { @@ -1046,27 +1044,27 @@ impl App { .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error())) }) .save(persister)?; - self.check_no_inputs_seen_before(proposal, persister).await + Ok(ReceiveSession::MaybeInputsSeen(proposal)) } - async fn check_no_inputs_seen_before( + fn check_no_inputs_seen_before( &self, proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { let proposal = proposal .check_no_inputs_seen_before(&mut |input| { Ok(self.db.insert_input_seen_before(*input)?) }) .save(persister)?; - self.identify_receiver_outputs(proposal, persister).await + Ok(ReceiveSession::OutputsUnknown(proposal)) } - async fn identify_receiver_outputs( + fn identify_receiver_outputs( &self, proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { let wallet = self.wallet(); let proposal = proposal .identify_receiver_outputs(&mut |output_script| { @@ -1075,23 +1073,23 @@ impl App { .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error())) }) .save(persister)?; - self.commit_outputs(proposal, persister).await + Ok(ReceiveSession::WantsOutputs(proposal)) } - async fn commit_outputs( + fn commit_outputs( &self, proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { let proposal = proposal.commit_outputs().save(persister)?; - self.contribute_inputs(proposal, persister).await + Ok(ReceiveSession::WantsInputs(proposal)) } - async fn contribute_inputs( + fn contribute_inputs( &self, proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { let wallet = self.wallet(); let candidate_inputs = wallet.list_unspent()?; @@ -1104,23 +1102,23 @@ impl App { let selected_input = proposal.try_preserving_privacy(candidate_inputs)?; let proposal = proposal.contribute_inputs(vec![selected_input])?.commit_inputs().save(persister)?; - self.apply_fee_range(proposal, persister).await + Ok(ReceiveSession::WantsFeeRange(proposal)) } - async fn apply_fee_range( + fn apply_fee_range( &self, proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { let proposal = proposal.apply_fee_range(None, self.config.max_fee_rate).save(persister)?; - self.finalize_proposal(proposal, persister).await + Ok(ReceiveSession::ProvisionalProposal(proposal)) } - async fn finalize_proposal( + fn finalize_proposal( &self, proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { let wallet = self.wallet(); let proposal = proposal .finalize_proposal(|psbt| { @@ -1129,40 +1127,44 @@ impl App { .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error())) }) .save(persister)?; - self.send_payjoin_proposal(proposal, persister).await + Ok(ReceiveSession::PayjoinProposal(proposal)) } async fn send_payjoin_proposal( &self, - mut proposal: Receiver, + proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { - loop { - let (res, ohttp_ctx) = - match self.post_via_relay(|relay| proposal.create_post_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => - return self.cancel_receiver_session(persister.session_id(), true), - }; - let payjoin_psbt = proposal.psbt().clone(); - match proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister) { - Ok(session) => { - persister.print(format_args!( - "Response successful. Watch mempool for successful Payjoin. TXID: {}", - payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid() - )); - return self.monitor_payjoin_proposal(session, persister).await; - } - Err(e) if e.is_transient() => { - tracing::debug!("Transient error sending payjoin proposal, retrying: {e:?}"); - proposal = e.transient_state().expect("transient error carries current state"); - tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; + ) -> Result { + let (res, ohttp_ctx) = + match self.post_via_relay(|relay| proposal.create_post_request(relay)).await? { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_receiver_session(persister.session_id(), true)?; + return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); } - Err(e) => return Err(e.into()), + }; + let payjoin_psbt = proposal.psbt().clone(); + match proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister) { + Ok(session) => { + persister.print(format_args!( + "Response successful. Watch mempool for successful Payjoin. TXID: {}", + payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid() + )); + Ok(ReceiveSession::Monitor(session)) } + Err(e) if e.is_transient() => { + tracing::debug!("Transient error sending payjoin proposal, retrying: {e:?}"); + let proposal = e.transient_state().expect("transient error carries current state"); + tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; + Ok(ReceiveSession::PayjoinProposal(proposal)) + } + Err(e) => Err(e.into()), } } + /// Watch the mempool for the payjoin transaction until it appears or a + /// timeout elapses. The poll/timeout loop is one logical step from the + /// session's perspective, so it stays local rather than in the driver. async fn monitor_payjoin_proposal( &self, mut proposal: Receiver, @@ -1220,49 +1222,50 @@ impl App { /// Handle error by attempting to send an error response over the directory async fn handle_error( &self, - mut session: Receiver, + session: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { - loop { - let (err_response, err_ctx) = - match self.post_via_relay(|relay| session.create_error_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => - return self.cancel_receiver_session(persister.session_id(), true), - }; - let err_bytes = match err_response.bytes().await { - Ok(bytes) => bytes, - Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)), + ) -> Result { + let (err_response, err_ctx) = + match self.post_via_relay(|relay| session.create_error_request(relay)).await? { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_receiver_session(persister.session_id(), true)?; + return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); + } }; + let err_bytes = match err_response.bytes().await { + Ok(bytes) => bytes, + Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)), + }; - match session.process_error_response(&err_bytes, err_ctx).save(persister) { - Ok(Some(pending)) => { - persister.print( - "Session delivered error reply. Broadcast the fallback transaction manually:", - ); - println!("{}", serialize_hex(pending.fallback_tx())); - return Ok(()); - } - Ok(None) => return Ok(()), - Err(e) if e.is_transient() => { - tracing::debug!("Transient error posting error response, retrying: {e:?}"); - session = e.transient_state().expect("transient error carries current state"); - tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; + match session.process_error_response(&err_bytes, err_ctx).save(persister) { + Ok(Some(pending)) => { + persister.print( + "Session delivered error reply. Broadcast the fallback transaction manually:", + ); + println!("{}", serialize_hex(pending.fallback_tx())); + Ok(ReceiveSession::PendingFallback(pending)) + } + Ok(None) => Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)), + Err(e) if e.is_transient() => { + tracing::debug!("Transient error posting error response, retrying: {e:?}"); + let session = e.transient_state().expect("transient error carries current state"); + tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; + Ok(ReceiveSession::HasReplyableError(session)) + } + Err(e) => { + if let Some(api_err) = e.api_error_ref() { + tracing::warn!("Failed to confirm error response delivery: {api_err}"); } - Err(e) => { - if let Some(api_err) = e.api_error_ref() { - tracing::warn!("Failed to confirm error response delivery: {api_err}"); - } - match e.fatal_state() { - Some(pending) => { - persister.print( - "Session failed to deliver error reply. Broadcast the fallback transaction manually:", - ); - println!("{}", serialize_hex(pending.fallback_tx())); - return Ok(()); - } - None => return Err(anyhow!("Failed to process error response")), + match e.fatal_state() { + Some(pending) => { + persister.print( + "Session failed to deliver error reply. Broadcast the fallback transaction manually:", + ); + println!("{}", serialize_hex(pending.fallback_tx())); + Ok(ReceiveSession::PendingFallback(pending)) } + None => Err(anyhow!("Failed to process error response")), } } } diff --git a/payjoin-cli/tests/e2e.rs b/payjoin-cli/tests/e2e.rs index a073d7be6..9cebbb9b8 100644 --- a/payjoin-cli/tests/e2e.rs +++ b/payjoin-cli/tests/e2e.rs @@ -53,37 +53,6 @@ mod e2e { session_id } - /// Read lines from `child_stderr` until `match_pattern` is found and the corresponding - /// line is returned. - /// Also writes every read line to tokio::io::stdout(); - async fn wait_for_stderr_match( - child_stderr: &mut tokio::process::ChildStderr, - match_pattern: F, - ) -> Option - where - F: Fn(&str) -> bool, - { - let reader = BufReader::new(child_stderr); - let mut lines = reader.lines(); - let mut res = None; - - let mut stderr = tokio::io::stderr(); - while let Some(line) = lines.next_line().await.expect("Failed to read line from stdout") { - // Write all output to tests stdout - stderr - .write_all(format!("{line}\n").as_bytes()) - .await - .expect("Failed to write to stderr"); - - if match_pattern(&line) { - res = Some(line); - break; - } - } - - res - } - /// Read lines from `child_stdout` until `match_pattern` is found and the corresponding /// line is returned. /// Also writes every read line to tokio::io::stdout(); @@ -1208,19 +1177,17 @@ mod e2e { .arg(&ohttp_keys_path) .arg("--expire-in") .arg("5") - .stdout(Stdio::inherit()) - .stderr(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) .spawn() .expect("Failed to execute payjoin-cli receiver"); - let mut receiver_stderr = - cli_receiver.stderr.take().expect("failed to take stderr of receiver"); + let mut receiver_stdout = + cli_receiver.stdout.take().expect("failed to take stdout of receiver"); let timeout = tokio::time::Duration::from_secs(60); let expired_line = tokio::time::timeout( timeout, - wait_for_stderr_match(&mut receiver_stderr, |l| { - l.contains("Error: Receiver session expired:") - }), + wait_for_stdout_match(&mut receiver_stdout, |l| l.contains("Session expired")), ) .await?;