diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index b1babe65c..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<()> { @@ -670,153 +678,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,33 +807,43 @@ impl App { return Ok(()); } ReceiveSession::Closed(_) => return Err(anyhow!("Session closed")), - } - }; - res + }; + } } - #[allow(clippy::incompatible_msrv)] 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() => { - println!("Interrupted. Call the `resume` command to resume all sessions."); - return Err(anyhow!("Interrupted")); + ) -> Result { + 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)) } - }?; - self.check_proposal(receiver, persister).await + Err(e) => Err(e.into()), + } } 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 +855,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 +871,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 +900,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 +929,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 +954,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 +1033,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")), } } }