29031142295

Committed 09 Jul 2026 03:50PM UTC coverage: 85.932% (-0.07%) from 86.003%

Build # 29031142295

Build Type

Pull #1721

github

Committed by

web-flow

Commit Message

Merge 92e311bf4 into fb210807b

Pull Request Pull Request #1721: Make persister no longer treat transient errors as an error

Coverage Stats

51 of 89 new or added lines in 3 files covered. (57.3%)

2 existing lines in 2 files now uncovered.

13487 of 15695 relevant lines covered (85.93%)

347.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

51.59

/payjoin-cli/src/app/v2/mod.rs

use std::fmt;
use std::sync::Arc;

use anyhow::{anyhow, Context, Result};
use payjoin::bitcoin::consensus::encode::serialize_hex;
use payjoin::bitcoin::{Amount, FeeRate};
use payjoin::persist::{OptionalTransitionOutcome, SessionPersister, TerminalSuccessOutcome};
use payjoin::receive::v2::{
    replay_event_log as replay_receiver_event_log, HasReplyableError, Initialized,
    MaybeInputsOwned, MaybeInputsSeen, Monitor, OutputsUnknown, PayjoinProposal,
    PendingFallback as ReceiverPendingFallback, ProvisionalProposal, ReceiveSession, Receiver,
    ReceiverBuilder, SessionOutcome as ReceiverSessionOutcome, UncheckedOriginalPayload,
    WantsFeeRange, WantsInputs, WantsOutputs,
};
use payjoin::send::v2::{
    replay_event_log as replay_sender_event_log, PendingFallback as SenderPendingFallback,
    PollingForProposal, SendSession, Sender, SenderBuilder, SessionOutcome as SenderSessionOutcome,
    WithReplyKey,
};
use payjoin::{ImplementationError, PjParam, Uri};
use tokio::sync::watch;

use super::config::Config;
use super::wallet::BitcoindWallet;
use super::App as AppTrait;
use crate::app::v2::ohttp::MailroomManager;
use crate::app::{handle_interrupt, http_agent};
use crate::cli::Role as CliRole;
use crate::db::v2::{ReceiverPersister, SenderPersister, SessionId};
use crate::db::Database;

mod ohttp;

const W_ID: usize = 36;
const W_ROLE: usize = 25;
const W_STATUS: usize = 15;

#[derive(Clone)]
pub(crate) struct App {
    config: Config,
    db: Arc<Database>,
    wallet: BitcoindWallet,
    interrupt: watch::Receiver<()>,
    mailroom_manager: MailroomManager,
}

trait StatusText {
    fn status_text(&self) -> &'static str;
}

impl StatusText for SendSession {
    fn status_text(&self) -> &'static str {
        match self {
            SendSession::WithReplyKey(_) | SendSession::PollingForProposal(_) =>
                "Waiting for proposal",
            SendSession::Closed(session_outcome) => match session_outcome {
                SenderSessionOutcome::Aborted => "Session aborted",
                SenderSessionOutcome::Success(_) => "Session success",
            },
            SendSession::PendingFallback(_) => "Session awaiting fallback",
        }
    }
}

impl StatusText for ReceiveSession {
    fn status_text(&self) -> &'static str {
        match self {
            ReceiveSession::Initialized(_) => "Waiting for original proposal",
            ReceiveSession::UncheckedOriginalPayload(_)
            | ReceiveSession::MaybeInputsOwned(_)
            | ReceiveSession::MaybeInputsSeen(_)
            | ReceiveSession::OutputsUnknown(_)
            | ReceiveSession::WantsOutputs(_)
            | ReceiveSession::WantsInputs(_)
            | ReceiveSession::WantsFeeRange(_)
            | ReceiveSession::ProvisionalProposal(_) => "Processing original proposal",
            ReceiveSession::PayjoinProposal(_) => "Payjoin proposal sent",
            ReceiveSession::HasReplyableError(_) =>
                "Session failure, waiting to post error response",
            ReceiveSession::Monitor(_) => "Monitoring payjoin proposal",
            ReceiveSession::PendingFallback(_) => "Pending fallback handling",
            ReceiveSession::Closed(session_outcome) => match session_outcome {
                ReceiverSessionOutcome::Aborted => "Session aborted",
                ReceiverSessionOutcome::Success(_) => "Session success, Payjoin proposal was broadcasted",
                ReceiverSessionOutcome::FallbackBroadcasted => "Fallback broadcasted",
                ReceiverSessionOutcome::PayjoinProposalSent =>
                    "Payjoin proposal sent, skipping monitoring as the sender is spending non-SegWit inputs",
            },
        }
    }
}

fn print_header() {
    println!("{:<W_ID$} {:<W_ROLE$} {:<W_STATUS$}", "Session ID", "Sender/Receiver", "Status");
}

enum Role {
    Sender,
    Receiver,
}
impl Role {
    fn as_str(&self) -> &'static str {
        match self {
            Role::Sender => "Sender",
            Role::Receiver => "Receiver",
        }
    }
}

struct SessionHistoryRow<Status> {
    session_id: SessionId,
    role: Role,
    status: Status,
    error_message: Option<String>,
}

impl<Status: StatusText> fmt::Display for SessionHistoryRow<Status> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "{:<W_ID$} {:<W_ROLE$} {:<W_STATUS$}",
            self.session_id.to_string(),
            self.role.as_str(),
            self.error_message.as_deref().unwrap_or(self.status.status_text())
        )
    }
}

#[async_trait::async_trait]
impl AppTrait for App {
    async fn new(config: Config) -> Result<Self> {
        let db = Arc::new(Database::create(&config.db_path)?);
        let mailroom_manager = MailroomManager::new(config.clone());
        let (interrupt_tx, interrupt_rx) = watch::channel(());
        tokio::spawn(handle_interrupt(interrupt_tx));
        let wallet = BitcoindWallet::new(&config.bitcoind).await?;
        let app = Self { config, db, wallet, interrupt: interrupt_rx, mailroom_manager };
        app.wallet()
            .network()
            .context("Failed to connect to bitcoind. Check config RPC connection.")?;
        Ok(app)
    }

    fn wallet(&self) -> BitcoindWallet { self.wallet.clone() }

    #[allow(clippy::incompatible_msrv)]
    async fn send_payjoin(&self, bip21: &str, fee_rate: FeeRate) -> Result<()> {
        use payjoin::UriExt;
        let uri = Uri::try_from(bip21)
            .map_err(|e| anyhow!("Failed to create URI from BIP21: {}", e))?
            .assume_checked()
            .check_pj_supported()
            .map_err(|_| anyhow!("URI does not support Payjoin"))?;
        let address = uri.address;
        let amount = uri.amount.ok_or_else(|| anyhow!("please specify the amount in the Uri"))?;
        match uri.extras.pj_param() {
            #[cfg(feature = "v1")]
            PjParam::V1(pj_param) => {
                let psbt = self.create_original_psbt(&address, amount, fee_rate)?;
                let fallback_tx = psbt.clone().extract_tx()?;
                let (req, ctx) = payjoin::send::v1::SenderBuilder::from_parts(
                    psbt,
                    pj_param,
                    &address,
                    Some(amount),
                )
                .build_recommended(fee_rate)
                .with_context(|| "Failed to build payjoin request")?
                .create_v1_post_request();
                let http = http_agent(&self.config)?;
                let body = String::from_utf8(req.body.clone()).unwrap();
                println!("Sending Original PSBT to {}", req.url);
                let response = match http
                    .post(req.url)
                    .header("Content-Type", req.content_type)
                    .body(body.clone())
                    .send()
                    .await
                {
                    Ok(response) => response,
                    Err(e) => {
                        tracing::error!("HTTP request failed: {e}");
                        println!("Payjoin failed. To broadcast the fallback transaction, run:");
                        println!(
                            "  bitcoin-cli -rpcwallet=<wallet> sendrawtransaction {:#}",
                            payjoin::bitcoin::consensus::encode::serialize_hex(&fallback_tx)
                        );
                        return Err(anyhow!("HTTP request failed: {e}"));
                    }
                };
                let psbt = match ctx.process_response(&response.bytes().await?) {
                    Ok(psbt) => psbt,
                    Err(e) => {
                        tracing::error!("Error processing response: {e:?}");
                        println!("Payjoin failed. To broadcast the fallback transaction, run:");
                        println!(
                            "  bitcoin-cli -rpcwallet=<wallet> sendrawtransaction {:#}",
                            payjoin::bitcoin::consensus::encode::serialize_hex(&fallback_tx)
                        );
                        return Err(anyhow!("Failed to process response {e}"));
                    }
                };

                self.process_pj_response(psbt)?;
                Ok(())
            }
            PjParam::V2(pj_param) => {
                let receiver_pubkey = pj_param.receiver_pubkey();
                let sender_state = self
                    .db
                    .get_send_session_id_by_receiver_pk(receiver_pubkey)?
                    .and_then(|session_id| {
                        let sender_persister =
                            SenderPersister::from_id(self.db.clone(), session_id);
                        let (send_session, _) = replay_sender_event_log(&sender_persister)
                            .map_err(|e| anyhow!("Failed to replay sender event log: {:?}", e))
                            .ok()?;
                        Some((send_session, sender_persister))
                    });

                let (sender_state, persister) = match sender_state {
                    Some((sender_state, persister)) => (sender_state, persister),
                    None => {
                        let psbt = self.create_original_psbt(&address, amount, fee_rate)?;
                        let persister =
                            SenderPersister::new(self.db.clone(), bip21, receiver_pubkey)?;
                        let sender =
                            SenderBuilder::from_parts(psbt, pj_param, &address, Some(amount))
                                .build_recommended(fee_rate)?
                                .save(&persister)?;

                        (SendSession::WithReplyKey(sender), persister)
                    }
                };
                println!("Send session established: {}", persister.session_id());
                let mut interrupt = self.interrupt.clone();
                tokio::select! {
                    res = self.process_sender_session(sender_state, &persister) => {
                        match res {
                            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.");
                                return Err(err);
                            }
                        }
                    },
                    _ = interrupt.changed() => {
                        let id = persister.session_id();
                        println!(
                            "Session {id} interrupted. Call `send` again to resume, `resume` to resume all sessions, or `payjoin-cli cancel {id}` to cancel and broadcast the original transaction."
                        );
                        return Err(anyhow!("Interrupted"))
                    }
                }
            }
            _ => unimplemented!("Unrecognized payjoin version"),
        }
    }

    async fn receive_payjoin(&self, amount: Amount) -> Result<()> {
        let address = self.wallet().get_new_address()?;
        let persister = ReceiverPersister::new(self.db.clone())?;
        let (directory, ohttp_keys) = loop {
            let directory = self.mailroom_manager.choose_directory()?;
            match self
                .mailroom_manager
                .unwrap_ohttp_keys_or_else_fetch_from_directory(&directory)
                .await
            {
                Ok(keys) => break (directory, keys.ohttp_keys),
                Err(e) => {
                    tracing::debug!("Directory {directory} failed: {e:#}");
                    self.mailroom_manager.add_failed_directory(directory);
                    self.mailroom_manager.clear_failed_relays();
                    continue;
                }
            }
        };
        let mut receiver_builder =
            ReceiverBuilder::new(address, directory.as_str(), ohttp_keys)?.with_amount(amount);
        if let Some(max_fee_rate) = self.config.max_fee_rate {
            receiver_builder = receiver_builder.with_max_fee_rate(max_fee_rate);
        }
        let session = receiver_builder.build().save(&persister)?;

        println!("Receive session established: {}", persister.session_id());
        let pj_uri = session.pj_uri();
        println!("Request Payjoin by sharing this Payjoin Uri:");
        println!("{pj_uri}");

        self.process_receiver_session(ReceiveSession::Initialized(session.clone()), &persister)
            .await?;
        Ok(())
    }

    async fn resume_payjoins(&self, session_id: Option<SessionId>) -> Result<()> {
        let mut recv_session_ids = self.db.get_recv_session_ids()?;
        let mut send_session_ids = self.db.get_send_session_ids()?;

        if let Some(ref target_id) = session_id {
            recv_session_ids.retain(|id| id == target_id);
            send_session_ids.retain(|id| id == target_id);
            if recv_session_ids.is_empty() && send_session_ids.is_empty() {
                anyhow::bail!("Session {target_id} not found or already completed");
            }
        }

        if recv_session_ids.is_empty() && send_session_ids.is_empty() {
            println!("No sessions to resume.");
            return Ok(());
        }

        let mut tasks: Vec<(String, tokio::task::JoinHandle<Result<()>>)> = Vec::new();

        // Process receiver sessions
        for session_id in recv_session_ids {
            let self_clone = self.clone();
            let recv_persister = ReceiverPersister::from_id(self.db.clone(), session_id.clone());
            match replay_receiver_event_log(&recv_persister) {
                Ok((receiver_state, _)) => {
                    tasks.push((
                        session_id.to_string(),
                        tokio::spawn(async move {
                            self_clone
                                .process_receiver_session(receiver_state, &recv_persister)
                                .await
                        }),
                    ));
                }
                Err(e) => {
                    if e.is_expired() {
                        println!("Session {session_id} receiver expired.");
                    } else {
                        tracing::error!(
                            "An error {:?} occurred while replaying receiver session",
                            e
                        );
                        println!("Session {session_id} receiver failed to replay -  {e}");
                    }
                    Self::close_failed_session(&recv_persister, &session_id, "receiver");
                }
            }
        }

        // Process sender sessions
        for session_id in send_session_ids {
            let sender_persister = SenderPersister::from_id(self.db.clone(), session_id.clone());
            match replay_sender_event_log(&sender_persister) {
                Ok((sender_state, _)) => {
                    let self_clone = self.clone();
                    tasks.push((
                        session_id.clone().to_string(),
                        tokio::spawn(async move {
                            self_clone.process_sender_session(sender_state, &sender_persister).await
                        }),
                    ));
                }
                Err(e) => {
                    if e.is_expired() {
                        println!("Session {session_id} sender expired.");
                    } else {
                        tracing::error!("An error {:?} occurred while replaying Sender session", e);
                        println!("Session {session_id} sender failed to replay -  {e}");
                    }
                    Self::close_failed_session(&sender_persister, &session_id, "sender");
                }
            }
        }

        let mut interrupt = self.interrupt.clone();
        tokio::select! {
            _ = async {

                for (session_id, task) in tasks {
                    match task.await {
                        Ok(Ok(())) => {
                            println!("Session {session_id} completed.");
                        }
                        Ok(Err(e)) => {
                            println!("Session {session_id} error: {e:#}");
                        }
                        Err(e) => {
                            println!("Session {session_id} panicked or was cancelled: {e:?}");
                        }
                    }
                }
        } => {}
            _ = interrupt.changed() => {
                println!("Resumed sessions were interrupted.");
            }
        }
        Ok(())
    }

    #[cfg(feature = "v2")]
    async fn history(&self) -> Result<()> {
        print_header();
        let mut send_rows = vec![];
        let mut recv_rows = vec![];
        self.db.get_send_session_ids()?.into_iter().for_each(|session_id| {
            let persister = SenderPersister::from_id(self.db.clone(), session_id.clone());
            match replay_sender_event_log(&persister) {
                Ok((sender_state, _)) => {
                    let row = SessionHistoryRow {
                        session_id,
                        role: Role::Sender,
                        status: sender_state.clone(),
                        error_message: None,
                    };
                    send_rows.push(row);
                }
                Err(e) => {
                    let row = SessionHistoryRow {
                        session_id,
                        role: Role::Sender,
                        status: SendSession::Closed(SenderSessionOutcome::Aborted),
                        error_message: Some(e.to_string()),
                    };
                    send_rows.push(row);
                }
            }
        });

        self.db.get_recv_session_ids()?.into_iter().for_each(|session_id| {
            let persister = ReceiverPersister::from_id(self.db.clone(), session_id.clone());
            match replay_receiver_event_log(&persister) {
                Ok((receiver_state, _)) => {
                    let row = SessionHistoryRow {
                        session_id,
                        role: Role::Receiver,
                        status: receiver_state.clone(),
                        error_message: None,
                    };
                    recv_rows.push(row);
                }
                Err(e) => {
                    let row = SessionHistoryRow {
                        session_id,
                        role: Role::Receiver,
                        status: ReceiveSession::Closed(ReceiverSessionOutcome::Aborted),
                        error_message: Some(e.to_string()),
                    };
                    recv_rows.push(row);
                }
            }
        });

        self.db.get_inactive_send_session_ids()?.into_iter().for_each(|(session_id, _)| {
            let persister = SenderPersister::from_id(self.db.clone(), session_id.clone());
            match replay_sender_event_log(&persister) {
                Ok((sender_state, _)) => {
                    let row = SessionHistoryRow {
                        session_id,
                        role: Role::Sender,
                        status: sender_state.clone(),
                        error_message: None,
                    };
                    send_rows.push(row);
                }
                Err(e) => {
                    let row = SessionHistoryRow {
                        session_id,
                        role: Role::Sender,
                        status: SendSession::Closed(SenderSessionOutcome::Aborted),
                        error_message: Some(e.to_string()),
                    };
                    send_rows.push(row);
                }
            }
        });

        self.db.get_inactive_recv_session_ids()?.into_iter().for_each(|(session_id, _)| {
            let persister = ReceiverPersister::from_id(self.db.clone(), session_id.clone());
            match replay_receiver_event_log(&persister) {
                Ok((receiver_state, _)) => {
                    let row = SessionHistoryRow {
                        session_id,
                        role: Role::Receiver,
                        status: receiver_state.clone(),
                        error_message: None,
                    };
                    recv_rows.push(row);
                }
                Err(e) => {
                    let row = SessionHistoryRow {
                        session_id,
                        role: Role::Receiver,
                        status: ReceiveSession::Closed(ReceiverSessionOutcome::Aborted),
                        error_message: Some(e.to_string()),
                    };
                    recv_rows.push(row);
                }
            }
        });

        // Collect every row (active + inactive, sender + receiver), sort by
        // session id, and print in a single stable order regardless of which
        // table or active/inactive bucket the row came from.
        let mut rows: Vec<(SessionId, String)> = Vec::new();
        for row in send_rows {
            rows.push((row.session_id.clone(), format!("{row}")));
        }
        for row in recv_rows {
            rows.push((row.session_id.clone(), format!("{row}")));
        }
        rows.sort_by_key(|(id, _)| id.clone());
        for (_, line) in rows {
            println!("{line}");
        }

        Ok(())
    }

    async fn cancel(
        &self,
        session_id: SessionId,
        no_broadcast: bool,
        role: Option<CliRole>,
    ) -> Result<()> {
        if let Some(role) = role {
            return match role {
                CliRole::Sender => self.cancel_sender_session(session_id, no_broadcast),
                CliRole::Receiver => self.cancel_receiver_session(session_id, no_broadcast),
            };
        }

        let is_sender = self.db.send_session_exists(&session_id)?;
        let is_receiver = self.db.recv_session_exists(&session_id)?;

        match (is_sender, is_receiver) {
            (true, false) => self.cancel_sender_session(session_id, no_broadcast),
            (false, true) => self.cancel_receiver_session(session_id, no_broadcast),
            (true, true) => anyhow::bail!(
                "Session {session_id} exists as both a sender and receiver session. \
                 Pass `--role sender` or `--role receiver`."
            ),
            (false, false) => anyhow::bail!("Session {session_id} not found"),
        }
    }
}

impl App {
    fn cancel_sender_session(&self, session_id: SessionId, no_broadcast: bool) -> Result<()> {
        let persister = SenderPersister::from_id(self.db.clone(), session_id.clone());
        let (session, history) = replay_sender_event_log(&persister)?;

        let pending: Sender<SenderPendingFallback> = match session {
            SendSession::WithReplyKey(sender) => sender.cancel().save(&persister)?,
            SendSession::PollingForProposal(sender) => sender.cancel().save(&persister)?,
            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}. \
                     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())
                );
                return Ok(());
            }
        };

        if no_broadcast {
            println!(
                "Session {session_id} cancelled. Broadcast the original transaction manually:\n{}",
                serialize_hex(pending.fallback_tx())
            );
        } else {
            self.wallet().broadcast_tx(pending.fallback_tx())?;
            println!(
                "Broadcasted fallback transaction txid: {}",
                pending.fallback_tx().compute_txid()
            );
        }
        pending.close().save(&persister)?;
        Ok(())
    }

    fn cancel_receiver_session(&self, session_id: SessionId, no_broadcast: bool) -> Result<()> {
        let persister = ReceiverPersister::from_id(self.db.clone(), session_id.clone());
        let (session, history) = replay_receiver_event_log(&persister)?;

        let pending: Receiver<ReceiverPendingFallback> = match session {
            ReceiveSession::Initialized(receiver) => {
                receiver.cancel().save(&persister)?;
                println!("Session {session_id} cancelled. No fallback transaction to broadcast.");
                return Ok(());
            }
            ReceiveSession::UncheckedOriginalPayload(receiver) => {
                receiver.cancel().save(&persister)?;
                println!("Session {session_id} cancelled. No fallback transaction to broadcast.");
                return Ok(());
            }
            ReceiveSession::MaybeInputsOwned(receiver) => receiver.cancel().save(&persister)?,
            ReceiveSession::MaybeInputsSeen(receiver) => receiver.cancel().save(&persister)?,
            ReceiveSession::OutputsUnknown(receiver) => receiver.cancel().save(&persister)?,
            ReceiveSession::WantsOutputs(receiver) => receiver.cancel().save(&persister)?,
            ReceiveSession::WantsInputs(receiver) => receiver.cancel().save(&persister)?,
            ReceiveSession::WantsFeeRange(receiver) => receiver.cancel().save(&persister)?,
            ReceiveSession::ProvisionalProposal(receiver) => receiver.cancel().save(&persister)?,
            ReceiveSession::PayjoinProposal(receiver) => receiver.cancel().save(&persister)?,
            ReceiveSession::Monitor(receiver) => receiver.cancel().save(&persister)?,
            ReceiveSession::HasReplyableError(receiver) => match receiver
                .cancel()
                .save(&persister)?
            {
                Some(pending) => pending,
                None => {
                    println!("Session {session_id} cancelled. No fallback transaction available.");
                    return Ok(());
                }
            },
            ReceiveSession::PendingFallback(receiver) => receiver,
            ReceiveSession::Closed(
                ReceiverSessionOutcome::Success(_)
                | ReceiverSessionOutcome::FallbackBroadcasted
                | ReceiverSessionOutcome::PayjoinProposalSent,
            ) => {
                println!("Session {session_id} already completed successfully. Cannot cancel.");
                return Ok(());
            }
            ReceiveSession::Closed(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."
                    ),
                }
                return Ok(());
            }
        };

        if no_broadcast {
            println!(
                "Session {session_id} cancelled. Broadcast the fallback transaction manually:\n{}",
                serialize_hex(pending.fallback_tx())
            );
        } else {
            self.wallet().broadcast_tx(pending.fallback_tx())?;
            println!(
                "Broadcasted fallback transaction txid: {}",
                pending.fallback_tx().compute_txid()
            );
        }
        pending.close().save(&persister)?;
        Ok(())
    }

    fn close_failed_session<P>(persister: &P, session_id: &SessionId, role: &str)
    where
        P: SessionPersister,
    {
        if let Err(close_err) = SessionPersister::close(persister) {
            tracing::error!("Failed to close {} session {}: {:?}", role, session_id, close_err);
        } else {
            tracing::error!("Closed failed {} session: {}", role, session_id);
        }
    }

    async fn process_sender_session(
        &self,
        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(());
            }
        }
        Ok(())
    }

    async fn post_original_proposal(
        &self,
        sender: Sender<WithReplyKey>,
        persister: &SenderPersister,
    ) -> Result<()> {
        let (response, ctx) =
            self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await?;
        let sender = sender.process_response(&response.bytes().await?, ctx).save(persister)?;
        println!("Posted Original PSBT...");
        self.get_proposed_payjoin_psbt(sender, persister).await
    }

    async fn get_proposed_payjoin_psbt(
        &self,
        sender: Sender<PollingForProposal>,
        persister: &SenderPersister,
    ) -> Result<()> {
        let mut session = sender.clone();
        // Long poll until we get a response
        loop {
            let (response, ctx) =
                self.post_via_relay(|relay| session.create_poll_request(relay)).await?;
            let res = session.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.");
                    session = current_state;
                    continue;
                }
                Err(re) => {
                    println!("{re}");
                    tracing::debug!("{re:?}");
                    return Err(anyhow!("Response error").context(re));
                }
            }
        }
    }

    async fn long_poll_fallback(
        &self,
        session: Receiver<Initialized>,
        persister: &ReceiverPersister,
    ) -> Result<Receiver<UncheckedOriginalPayload>> {
        let mut session = session;
        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;
                    continue;
                }
                Err(e) => return Err(e.into()),
            }
        }
    }

    async fn process_receiver_session(
        &self,
        session: ReceiveSession,
        persister: &ReceiverPersister,
    ) -> Result<()> {
        let res = {
            match session {
                ReceiveSession::Initialized(proposal) =>
                    self.read_from_directory(proposal, persister).await,
                ReceiveSession::UncheckedOriginalPayload(proposal) =>
                    self.check_proposal(proposal, persister).await,
                ReceiveSession::MaybeInputsOwned(proposal) =>
                    self.check_inputs_not_owned(proposal, persister).await,
                ReceiveSession::MaybeInputsSeen(proposal) =>
                    self.check_no_inputs_seen_before(proposal, persister).await,
                ReceiveSession::OutputsUnknown(proposal) =>
                    self.identify_receiver_outputs(proposal, persister).await,
                ReceiveSession::WantsOutputs(proposal) =>
                    self.commit_outputs(proposal, persister).await,
                ReceiveSession::WantsInputs(proposal) =>
                    self.contribute_inputs(proposal, persister).await,
                ReceiveSession::WantsFeeRange(proposal) =>
                    self.apply_fee_range(proposal, persister).await,
                ReceiveSession::ProvisionalProposal(proposal) =>
                    self.finalize_proposal(proposal, persister).await,
                ReceiveSession::PayjoinProposal(proposal) =>
                    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,
                ReceiveSession::PendingFallback(_) => {
                    let id = persister.session_id();
                    println!(
                        "Session {id} was cancelled. Run `payjoin-cli cancel {id}` to cancel and broadcast the fallback transaction."
                    );
                    return Ok(());
                }
                ReceiveSession::Closed(_) => return Err(anyhow!("Session closed")),
            }
        };
        res
    }

    #[allow(clippy::incompatible_msrv)]
    async fn read_from_directory(
        &self,
        session: Receiver<Initialized>,
        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"));
            }
        }?;
        self.check_proposal(receiver, persister).await
    }

    async fn check_proposal(
        &self,
        proposal: Receiver<UncheckedOriginalPayload>,
        persister: &ReceiverPersister,
    ) -> Result<()> {
        let wallet = self.wallet();
        let proposal = proposal
            .check_broadcast_suitability(None, |tx| {
                wallet
                    .can_broadcast(tx)
                    .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
            })
            .save(persister)?;

        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
    }

    async fn check_inputs_not_owned(
        &self,
        proposal: Receiver<MaybeInputsOwned>,
        persister: &ReceiverPersister,
    ) -> Result<()> {
        let wallet = self.wallet();
        let proposal = proposal
            .check_inputs_not_owned(&mut |input| {
                wallet
                    .is_mine(input)
                    .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
            })
            .save(persister)?;
        self.check_no_inputs_seen_before(proposal, persister).await
    }

    async fn check_no_inputs_seen_before(
        &self,
        proposal: Receiver<MaybeInputsSeen>,
        persister: &ReceiverPersister,
    ) -> 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
    }

    async fn identify_receiver_outputs(
        &self,
        proposal: Receiver<OutputsUnknown>,
        persister: &ReceiverPersister,
    ) -> Result<()> {
        let wallet = self.wallet();
        let proposal = proposal
            .identify_receiver_outputs(&mut |output_script| {
                wallet
                    .is_mine(output_script)
                    .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
            })
            .save(persister)?;
        self.commit_outputs(proposal, persister).await
    }

    async fn commit_outputs(
        &self,
        proposal: Receiver<WantsOutputs>,
        persister: &ReceiverPersister,
    ) -> Result<()> {
        let proposal = proposal.commit_outputs().save(persister)?;
        self.contribute_inputs(proposal, persister).await
    }

    async fn contribute_inputs(
        &self,
        proposal: Receiver<WantsInputs>,
        persister: &ReceiverPersister,
    ) -> Result<()> {
        let wallet = self.wallet();
        let candidate_inputs = wallet.list_unspent()?;

        if candidate_inputs.is_empty() {
            return Err(anyhow::anyhow!(
                "No spendable UTXOs available in wallet. Cannot contribute inputs to payjoin."
            ));
        }

        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
    }

    async fn apply_fee_range(
        &self,
        proposal: Receiver<WantsFeeRange>,
        persister: &ReceiverPersister,
    ) -> Result<()> {
        let proposal = proposal.apply_fee_range(None, self.config.max_fee_rate).save(persister)?;
        self.finalize_proposal(proposal, persister).await
    }

    async fn finalize_proposal(
        &self,
        proposal: Receiver<ProvisionalProposal>,
        persister: &ReceiverPersister,
    ) -> Result<()> {
        let wallet = self.wallet();
        let proposal = proposal
            .finalize_proposal(|psbt| {
                wallet
                    .process_psbt(psbt)
                    .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
            })
            .save(persister)?;
        self.send_payjoin_proposal(proposal, persister).await
    }

    async fn send_payjoin_proposal(
        &self,
        proposal: Receiver<PayjoinProposal>,
        persister: &ReceiverPersister,
    ) -> 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();
        let session = proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister)?;
        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;
    }

    async fn monitor_payjoin_proposal(
        &self,
        proposal: Receiver<Monitor>,
        persister: &ReceiverPersister,
    ) -> Result<()> {
        // On a session resumption, the receiver will resume again in this state.
        let poll_interval = tokio::time::Duration::from_millis(200);
        let timeout_duration = tokio::time::Duration::from_secs(5);

        let mut interval = tokio::time::interval(poll_interval);
        interval.tick().await;

        tracing::debug!("Polling for payment confirmation");

        let result = tokio::time::timeout(timeout_duration, async {
            loop {
                interval.tick().await;
                let check_result = proposal
                    .check_for_transaction(|txid| {
                        self.wallet()
                            .get_raw_transaction(&txid)
                            .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error()))
                    })
                    .save(persister);

                match check_result {
                    Ok(OptionalTransitionOutcome::Progress(())) => {
                        println!("Payjoin transaction detected in the mempool!");
                        return Ok(());
                    }
                    Ok(OptionalTransitionOutcome::Stasis(_)) => continue,
                    Err(_) => continue,
                }
            }
        })
        .await;

        match result {
            Ok(ok) => ok,
            Err(_) => Err(anyhow!(
                "No payjoin transaction detected in mempool within {timeout_duration:?}, stopping."
            )),
        }
    }

    /// Handle error by attempting to send an error response over the directory
    async fn handle_error(
        &self,
        mut session: Receiver<HasReplyableError>,
        persister: &ReceiverPersister,
    ) -> Result<()> {
        let session_id = persister.session_id();
        for retries in 0..3 {
            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)),
            };

            match session.process_error_response(&err_bytes, err_ctx).save(persister) {
                Ok(TerminalSuccessOutcome::Advance(_)) => {
                    println!("Error reply delivered. Run `payjoin-cli cancel {session_id}` to broadcast the fallback transaction.");
                    return Ok(());
                }
                Ok(TerminalSuccessOutcome::Terminate) => {
                    println!("Session {session_id} aborted. No fallback transaction available.");
                    return Ok(());
                }
                Ok(TerminalSuccessOutcome::Transient(current_session, err)) if retries < 2 => {
                    tracing::warn!("Transient error posting error reply: {err}; retrying");
                    session = current_session;
                    continue;
                }
                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_ref() {
                        Some(_) => {
                            println!("Session {session_id} failed. Run `payjoin-cli cancel {session_id}` to cancel and broadcast the fallback transaction.");
                            return Ok(());
                        }
                        None => return Err(e.into()),
                    }
                }
                _ => {
                    tracing::error!("Failed to post error reply after 3 retries.");
                    return Err(anyhow!("Retry error"));
                }
            }
        }
        Ok(())
    }

    async fn post_request(&self, req: payjoin::Request) -> Result<reqwest::Response> {
        let http = http_agent(&self.config)?;
        http.post(req.url)
            .header("Content-Type", req.content_type)
            .body(req.body)
            .send()
            .await
            .and_then(|r| r.error_for_status())
            .context("HTTP request failed")
    }

    async fn post_via_relay<F, T, E>(&self, mut build: F) -> Result<(reqwest::Response, T)>
    where
        F: FnMut(&str) -> std::result::Result<(payjoin::Request, T), E>,
        E: Into<anyhow::Error>,
    {
        loop {
            let relay = self.mailroom_manager.choose_relay()?;
            let (req, ctx) = build(relay.as_str()).map_err(Into::into)?;
            match self.post_request(req).await {
                Ok(resp) => return Ok((resp, ctx)),
                Err(e) => {
                    tracing::debug!("Request to relay {relay} failed: {e:?}");
                    self.mailroom_manager.add_failed_relay(relay);
                }
            }
        }
    }
}

payjoin / rust-payjoin / 29031142295

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous