From 5bbd0bf4b68e8d0315661982410feb10718ffbd9 Mon Sep 17 00:00:00 2001 From: xstoicunicornx Date: Fri, 10 Jul 2026 15:53:54 -0700 Subject: [PATCH 1/2] 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. --- payjoin-cli/src/app/v2/mod.rs | 382 +++++++++++++++++----------------- 1 file changed, 188 insertions(+), 194 deletions(-) diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index b1babe65c..da86ef4af 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -670,153 +670,127 @@ impl App { } } + /// Drive the sender state machine to completion. Each step function + /// performs a single state transition and returns the next session + /// state, so retries and polling loop through here rather than + /// recursing. 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)) => { - self.process_pj_response(proposal)?; - return Ok(()); - } - SendSession::Closed(SenderSessionOutcome::Aborted) => { - println!("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." - ); - 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)) => { + self.process_pj_response(proposal)?; + return Ok(()); + } + SendSession::Closed(SenderSessionOutcome::Aborted) => { + println!("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." + ); + return Ok(()); + } + }; } - Ok(()) } async fn post_original_proposal( &self, - mut sender: Sender, + sender: Sender, persister: &SenderPersister, - ) -> Result<()> { - loop { - let (response, ctx) = - self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await?; - match sender.process_response(&response.bytes().await?, ctx).save(persister) { - Ok(sender) => { - println!("Posted Original PSBT..."); - return self.get_proposed_payjoin_psbt(sender, persister).await; - } - Err(e) if e.is_transient() => { - tracing::debug!("Transient error posting original proposal, retrying: {e:?}"); - sender = e.transient_state().expect("transient error carries current state"); - tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; - } - Err(e) => return Err(e.into()), + ) -> Result { + let (response, ctx) = + self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await?; + match sender.process_response(&response.bytes().await?, ctx).save(persister) { + Ok(sender) => { + println!("Posted Original PSBT..."); + 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) = - self.post_via_relay(|relay| sender.create_poll_request(relay)).await?; - let res = sender.process_response(&response.bytes().await?, ctx).save(persister); - match res { - Ok(OptionalTransitionOutcome::Progress(psbt)) => { - println!("Proposal received. Processing..."); - self.process_pj_response(psbt)?; - return Ok(()); - } - Ok(OptionalTransitionOutcome::Stasis(current_state)) => { - println!("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) => { - println!("{re}"); - tracing::debug!("{re:?}"); - return Err(anyhow!("Response error").context(re)); - } + ) -> Result { + let (response, ctx) = + self.post_via_relay(|relay| sender.create_poll_request(relay)).await?; + match sender.process_response(&response.bytes().await?, ctx).save(persister) { + Ok(OptionalTransitionOutcome::Progress(psbt)) => { + println!("Proposal received. Processing..."); + Ok(SendSession::Closed(SenderSessionOutcome::Success(psbt))) } - } - } - - async fn long_poll_fallback( - &self, - mut session: Receiver, - persister: &ReceiverPersister, - ) -> Result> { - loop { - println!("Polling receive request..."); - let (ohttp_response, context) = - self.post_via_relay(|relay| session.create_poll_request(relay)).await?; - 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)) => { - println!("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)) => { + println!("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) => { + println!("{re}"); + tracing::debug!("{re:?}"); + Err(anyhow!("Response error").context(re)) } } } + /// Drive the receiver state machine to completion. Each step function + /// performs a single state transition and returns the next session + /// state, so retries and polling loop through here rather than + /// recursing. 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).await?, ReceiveSession::MaybeInputsOwned(proposal) => - self.check_inputs_not_owned(proposal, persister).await, + self.check_inputs_not_owned(proposal, persister).await?, ReceiveSession::MaybeInputsSeen(proposal) => - self.check_no_inputs_seen_before(proposal, persister).await, + self.check_no_inputs_seen_before(proposal, persister).await?, ReceiveSession::OutputsUnknown(proposal) => - self.identify_receiver_outputs(proposal, persister).await, + self.identify_receiver_outputs(proposal, persister).await?, ReceiveSession::WantsOutputs(proposal) => - self.commit_outputs(proposal, persister).await, + self.commit_outputs(proposal, persister).await?, ReceiveSession::WantsInputs(proposal) => - self.contribute_inputs(proposal, persister).await, + self.contribute_inputs(proposal, persister).await?, ReceiveSession::WantsFeeRange(proposal) => - self.apply_fee_range(proposal, persister).await, + self.apply_fee_range(proposal, persister).await?, ReceiveSession::ProvisionalProposal(proposal) => - self.finalize_proposal(proposal, persister).await, + self.finalize_proposal(proposal, persister).await?, 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, + self.handle_error(error, persister).await?, ReceiveSession::Monitor(proposal) => - self.monitor_payjoin_proposal(proposal, persister).await, + return self.monitor_payjoin_proposal(proposal, persister).await, ReceiveSession::PendingFallback(_) => { let id = persister.session_id(); println!( @@ -825,9 +799,8 @@ impl App { return Ok(()); } ReceiveSession::Closed(_) => return Err(anyhow!("Session closed")), - } - }; - res + }; + } } #[allow(clippy::incompatible_msrv)] @@ -835,23 +808,46 @@ impl App { &self, session: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { + let poll_once = async { + println!("Polling receive request..."); + let (ohttp_response, context) = + self.post_via_relay(|relay| session.create_poll_request(relay)).await?; + 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)) => { + println!("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)) + } + Err(e) => Err(e.into()), + } + }; let mut interrupt = self.interrupt.clone(); - let receiver = tokio::select! { - res = self.long_poll_fallback(session, persister) => res, + tokio::select! { + res = poll_once => res, _ = interrupt.changed() => { println!("Interrupted. Call the `resume` command to resume all sessions."); - return Err(anyhow!("Interrupted")); + Err(anyhow!("Interrupted")) } - }?; - self.check_proposal(receiver, persister).await + } } async fn check_proposal( &self, proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { let wallet = self.wallet(); let proposal = proposal .check_broadcast_suitability(None, |tx| { @@ -863,14 +859,14 @@ impl App { println!("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( &self, proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { let wallet = self.wallet(); let proposal = proposal .check_inputs_not_owned(&mut |input| { @@ -879,27 +875,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( &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( &self, proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { let wallet = self.wallet(); let proposal = proposal .identify_receiver_outputs(&mut |output_script| { @@ -908,23 +904,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( &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( &self, proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { let wallet = self.wallet(); let candidate_inputs = wallet.list_unspent()?; @@ -937,23 +933,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( &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( &self, proposal: Receiver, persister: &ReceiverPersister, - ) -> Result<()> { + ) -> Result { let wallet = self.wallet(); let proposal = proposal .finalize_proposal(|psbt| { @@ -962,38 +958,37 @@ 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) = self - .post_via_relay(|relay| { - proposal - .create_post_request(relay) - .map_err(|e| anyhow!("v2 req extraction failed {}", e)) - }) - .await?; - let payjoin_psbt = proposal.psbt().clone(); - match proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister) { - Ok(session) => { - println!( - "Response successful. Watch mempool for successful Payjoin. TXID: {}", - payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid() - ); - return self.monitor_payjoin_proposal(session, persister).await; - } - Err(e) if e.is_transient() => { - tracing::debug!("Transient error sending payjoin proposal, retrying: {e:?}"); - proposal = e.transient_state().expect("transient error carries current state"); - tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; - } - Err(e) => return Err(e.into()), + ) -> Result { + let (res, ohttp_ctx) = self + .post_via_relay(|relay| { + proposal + .create_post_request(relay) + .map_err(|e| anyhow!("v2 req extraction failed {}", e)) + }) + .await?; + let payjoin_psbt = proposal.psbt().clone(); + match proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister) { + Ok(session) => { + println!( + "Response successful. Watch mempool for successful Payjoin. TXID: {}", + payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid() + ); + 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()), } } @@ -1042,47 +1037,46 @@ impl App { } } - /// Handle error by attempting to send an error response over the directory + /// 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) = self - .post_via_relay(|relay| { - session - .create_error_request(relay) - .map_err(|e| anyhow!("Failed to post error request: {}", e)) - }) - .await?; - - let err_bytes = match err_response.bytes().await { - Ok(bytes) => bytes, - Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)), - }; + ) -> Result { + let (err_response, err_ctx) = self + .post_via_relay(|relay| { + session + .create_error_request(relay) + .map_err(|e| anyhow!("Failed to post error request: {}", e)) + }) + .await?; - match session.process_error_response(&err_bytes, err_ctx).save(persister) { - Ok(_) => return Ok(()), - Err(e) if e.is_transient() => { - tracing::debug!("Transient error posting error response, retrying: {e:?}"); - session = e.transient_state().expect("transient error carries current state"); - tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; + 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(_) => 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.error_state() { - Some(_) => { - let id = persister.session_id(); - println!( - "Session {id} failed. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction." - ); - return Ok(()); - } - None => return Err(anyhow!("Failed to process error response")), + match e.error_state() { + Some(session) => { + let id = persister.session_id(); + println!( + "Session {id} failed. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction." + ); + Ok(ReceiveSession::PendingFallback(session)) } + None => Err(anyhow!("Failed to process error response")), } } } From 32630ff897d410adb796a96c0462e39203b368f3 Mon Sep 17 00:00:00 2001 From: xstoicunicornx Date: Mon, 13 Jul 2026 08:26:10 -0700 Subject: [PATCH 2/2] Handle interrupts across the whole receive session 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 | 62 ++++++++++++++++------------------- 1 file changed, 29 insertions(+), 33 deletions(-) diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index da86ef4af..f68bc765d 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -293,9 +293,17 @@ impl AppTrait for App { println!("Request Payjoin by sharing this Payjoin Uri:"); println!("{pj_uri}"); - self.process_receiver_session(ReceiveSession::Initialized(session.clone()), &persister) - .await?; - Ok(()) + let mut interrupt = self.interrupt.clone(); + tokio::select! { + res = self.process_receiver_session(ReceiveSession::Initialized(session.clone()), &persister) => res, + _ = interrupt.changed() => { + let id = persister.session_id(); + println!( + "Session {id} interrupted. Call the `resume` command to resume all sessions, or `payjoin-cli cancel {id}` to cancel the session." + ); + Err(anyhow!("Interrupted")) + } + } } async fn resume_payjoins(&self, session_id: Option) -> Result<()> { @@ -803,43 +811,31 @@ impl App { } } - #[allow(clippy::incompatible_msrv)] async fn read_from_directory( &self, session: Receiver, persister: &ReceiverPersister, ) -> Result { - let poll_once = async { - println!("Polling receive request..."); - let (ohttp_response, context) = - self.post_via_relay(|relay| session.create_poll_request(relay)).await?; - 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)) => { - println!("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)) - } - Err(e) => Err(e.into()), + println!("Polling receive request..."); + let (ohttp_response, context) = + self.post_via_relay(|relay| session.create_poll_request(relay)).await?; + 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)) => { + println!("Got a request from the sender. Responding with a Payjoin proposal."); + Ok(ReceiveSession::UncheckedOriginalPayload(next_state)) } - }; - let mut interrupt = self.interrupt.clone(); - tokio::select! { - res = poll_once => res, - _ = interrupt.changed() => { - println!("Interrupted. Call the `resume` command to resume all sessions."); - Err(anyhow!("Interrupted")) + 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)) } + Err(e) => Err(e.into()), } }