15767995657

Committed 19 Jun 2025 11:38PM UTC coverage: 86.087% (-0.05%) from 86.135%

Build # 15767995657

Build Type

push

github

Committed by

web-flow

Commit Message

Replace `Persister` with `SessionPersister` for `v2::Receiver` (#750)

Please take a look at individual commits for a more complete description
of the changes.
commits messages prefixed'd with "squash" indicates that they are meant
to be squash'd with the previous commit before this PR gets merged.

One open item before this is undrafted:
- [X] c7c31fcf4 introduces a In memory
receiver sesssion persister. It is accessible to the integration tests
but not to sub mod unit tests (?) . Need to move this persister impl a
common location or figure out why its not visibile to the unit tests.

Run Details

710 of 823 new or added lines in 8 files covered. (86.27%)

5 existing lines in 3 files now uncovered.

7617 of 8848 relevant lines covered (86.09%)

525.55 hits per line

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

81.02

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

use std::sync::{Arc, Mutex};

use anyhow::{anyhow, Context, Result};
use payjoin::bitcoin::consensus::encode::serialize_hex;
use payjoin::bitcoin::psbt::Psbt;
use payjoin::bitcoin::{Amount, FeeRate};
use payjoin::persist::OptionalTransitionOutcome;
use payjoin::receive::v2::{
    process_err_res, replay_event_log as replay_receiver_event_log, MaybeInputsOwned,
    MaybeInputsSeen, OutputsUnknown, PayjoinProposal, ProvisionalProposal, Receiver,
    ReceiverTypeState, SessionHistory, UncheckedProposal, WantsInputs, WantsOutputs, WithContext,
};
use payjoin::send::v2::{Sender, SenderBuilder, WithReplyKey};
use payjoin::Uri;
use tokio::sync::watch;

use super::config::Config;
use super::wallet::BitcoindWallet;
use super::App as AppTrait;
use crate::app::v2::ohttp::{unwrap_ohttp_keys_or_else_fetch, RelayManager};
use crate::app::{handle_interrupt, http_agent};
use crate::db::v2::{ReceiverPersister, SenderPersister};
use crate::db::Database;

mod ohttp;

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

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

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

    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))?;
        let uri = uri.assume_checked();
        let uri = uri.check_pj_supported().map_err(|_| anyhow!("URI does not support Payjoin"))?;
        let url = uri.extras.endpoint();
        // match bip21 to send_session public_key
        let req_ctx = match self.db.get_send_session(url)? {
            Some(send_session) => send_session,
            None => {
                let psbt = self.create_original_psbt(&uri, fee_rate)?;
                let mut persister = SenderPersister::new(self.db.clone());
                let new_sender = SenderBuilder::new(psbt, uri.clone())
                    .build_recommended(fee_rate)
                    .with_context(|| "Failed to build payjoin request")?;
                let storage_token = new_sender
                    .persist(&mut persister)
                    .map_err(|e| anyhow!("Failed to persist sender: {}", e))?;
                Sender::load(storage_token, &persister)
                    .map_err(|e| anyhow!("Failed to load sender: {}", e))?
            }
        };
        self.spawn_payjoin_sender(req_ctx).await
    }

    async fn receive_payjoin(&self, amount: Amount) -> Result<()> {
        let address = self.wallet().get_new_address()?;
        let ohttp_keys =
            unwrap_ohttp_keys_or_else_fetch(&self.config, None, self.relay_manager.clone())
                .await?
                .ohttp_keys;
        let persister = ReceiverPersister::new(self.db.clone())?;
        let session = Receiver::create_session(
            address,
            self.config.v2()?.pj_directory.clone(),
            ohttp_keys,
            None,
        )
        .save(&persister)?;
        println!("Receive session established");
        let mut pj_uri = session.pj_uri();
        pj_uri.amount = Some(amount);
        println!("Request Payjoin by sharing this Payjoin Uri:");
        println!("{}", pj_uri);

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

    #[allow(clippy::incompatible_msrv)]
    async fn resume_payjoins(&self) -> Result<()> {
        let recv_session_ids = self.db.get_recv_session_ids()?;
        let send_sessions = self.db.get_send_sessions()?;

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

        let mut tasks = Vec::new();

        for session_id in recv_session_ids {
            let self_clone = self.clone();
            let recv_persister = ReceiverPersister::from_id(self.db.clone(), session_id)?;
            let receiver_state = replay_receiver_event_log(&recv_persister)
                .map_err(|e| anyhow!("Failed to replay receiver event log: {:?}", e))?
                .0;
            tasks.push(tokio::spawn(async move {
                self_clone.process_receiver_session(receiver_state, &recv_persister).await
            }));
        }

        for session in send_sessions {
            let self_clone = self.clone();
            tasks.push(tokio::spawn(async move { self_clone.spawn_payjoin_sender(session).await }));
        }

        let mut interrupt = self.interrupt.clone();
        tokio::select! {
            _ = async {
                for task in tasks {
                    let _ = task.await;
                }
            } => {
                println!("All resumed sessions completed.");
            }
            _ = interrupt.changed() => {
                println!("Resumed sessions were interrupted.");
            }
        }
        Ok(())
    }
}

impl App {
    #[allow(clippy::incompatible_msrv)]
    async fn spawn_payjoin_sender(&self, mut req_ctx: Sender<WithReplyKey>) -> Result<()> {
        let mut interrupt = self.interrupt.clone();
        tokio::select! {
            res = self.long_poll_post(&mut req_ctx) => {
                self.process_pj_response(res?)?;
                self.db.clear_send_session(req_ctx.endpoint())?;
            }
            _ = interrupt.changed() => {
                println!("Interrupted. Call `send` with the same arguments to resume this session or `resume` to resume all sessions.");
            }
        }
        Ok(())
    }

    async fn long_poll_post(&self, req_ctx: &mut Sender<WithReplyKey>) -> Result<Psbt> {
        let ohttp_relay = self.unwrap_relay_or_else_fetch(Some(req_ctx.endpoint().clone())).await?;

        match req_ctx.extract_v2(ohttp_relay.clone()) {
            Ok((req, ctx)) => {
                println!("Posting Original PSBT Payload request...");
                let response = post_request(req).await?;
                println!("Sent fallback transaction");
                let v2_ctx = Arc::new(ctx.process_response(&response.bytes().await?)?);
                loop {
                    let (req, ohttp_ctx) = v2_ctx.extract_req(&ohttp_relay)?;
                    let response = post_request(req).await?;
                    match v2_ctx.process_response(&response.bytes().await?, ohttp_ctx) {
                        Ok(Some(psbt)) => return Ok(psbt),
                        Ok(None) => {
                            println!("No response yet.");
                        }
                        Err(re) => {
                            println!("{re}");
                            log::debug!("{re:?}");
                            return Err(anyhow!("Response error").context(re));
                        }
                    }
                }
            }
            Err(_) => {
                let (req, v1_ctx) = req_ctx.extract_v1();
                println!("Posting Original PSBT Payload request...");
                let response = post_request(req).await?;
                println!("Sent fallback transaction");
                match v1_ctx.process_response(&response.bytes().await?) {
                    Ok(psbt) => Ok(psbt),
                    Err(re) => {
                        println!("{re}");
                        log::debug!("{re:?}");
                        Err(anyhow!("Response error").context(re))
                    }
                }
            }
        }
    }

    async fn long_poll_fallback(
        &self,
        session: Receiver<WithContext>,
        persister: &ReceiverPersister,
    ) -> Result<Receiver<UncheckedProposal>> {
        let ohttp_relay = self
            .unwrap_relay_or_else_fetch(Some(session.pj_uri().extras.endpoint().clone()))
            .await?;

        let mut session = session;
        loop {
            let (req, context) = session.extract_req(&ohttp_relay)?;
            println!("Polling receive request...");
            let ohttp_response = post_request(req).await?;
            let state_transition = session
                .process_res(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: ReceiverTypeState,
        persister: &ReceiverPersister,
    ) -> Result<()> {
        let res = {
            match session {
                ReceiverTypeState::WithContext(proposal) =>
                    self.read_from_directory(proposal, persister).await,
                ReceiverTypeState::UncheckedProposal(proposal) =>
                    self.check_proposal(proposal, persister).await,
                ReceiverTypeState::MaybeInputsOwned(proposal) =>
                    self.check_inputs_not_owned(proposal, persister).await,
                ReceiverTypeState::MaybeInputsSeen(proposal) =>
                    self.check_no_inputs_seen_before(proposal, persister).await,
                ReceiverTypeState::OutputsUnknown(proposal) =>
                    self.identify_receiver_outputs(proposal, persister).await,
                ReceiverTypeState::WantsOutputs(proposal) =>
                    self.commit_outputs(proposal, persister).await,
                ReceiverTypeState::WantsInputs(proposal) =>
                    self.contribute_inputs(proposal, persister).await,
                ReceiverTypeState::ProvisionalProposal(proposal) =>
                    self.finalize_proposal(proposal, persister).await,
                ReceiverTypeState::PayjoinProposal(proposal) =>
                    self.send_payjoin_proposal(proposal, persister).await,
                ReceiverTypeState::Uninitialized(_) =>
                    return Err(anyhow!("Uninitialized receiver session")),
                ReceiverTypeState::TerminalState =>
                    return Err(anyhow!("Terminal receiver session")),
            }
        };

        match res {
            Ok(_) => Ok(()),
            Err(e) => {
                let (_, session_history) = replay_receiver_event_log(persister)?;
                let pj_uri = match session_history.pj_uri() {
                    Some(uri) => Some(uri.extras.endpoint().clone()),
                    None => None,
                };
                let ohttp_relay = self.unwrap_relay_or_else_fetch(pj_uri).await?;
                handle_recoverable_error(&ohttp_relay, &session_history).await?;

                Err(e)
            }
        }
    }

    #[allow(clippy::incompatible_msrv)]
    async fn read_from_directory(
        &self,
        session: Receiver<WithContext>,
        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"));
            }
        }?;
        println!("Fallback transaction received. Consider broadcasting this to get paid if the Payjoin fails:");
        println!("{}", serialize_hex(&receiver.extract_tx_to_schedule_broadcast()));
        self.check_proposal(receiver, persister).await
    }

    async fn check_proposal(
        &self,
        proposal: Receiver<UncheckedProposal>,
        persister: &ReceiverPersister,
    ) -> Result<()> {
        let wallet = self.wallet();
        let proposal = proposal
            .check_broadcast_suitability(None, |tx| Ok(wallet.can_broadcast(tx)?))
            .save(persister)?;

        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(|input| Ok(wallet.is_mine(input)?)).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(|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(|output_script| Ok(wallet.is_mine(output_script)?))
            .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()?;

        let selected_input = proposal.try_preserving_privacy(candidate_inputs)?;
        let proposal =
            proposal.contribute_inputs(vec![selected_input])?.commit_inputs().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| Ok(wallet.process_psbt(psbt)?),
                None,
                self.config.max_fee_rate,
            )
            .save(persister)?;
        self.send_payjoin_proposal(proposal, persister).await
    }

    async fn send_payjoin_proposal(
        &self,
        mut proposal: Receiver<PayjoinProposal>,
        persister: &ReceiverPersister,
    ) -> Result<()> {
        let (req, ohttp_ctx) = proposal
            .extract_req(&self.unwrap_relay_or_else_fetch(None).await?)
            .map_err(|e| anyhow!("v2 req extraction failed {}", e))?;
        let res = post_request(req).await?;
        let payjoin_psbt = proposal.psbt().clone();
        proposal.process_res(&res.bytes().await?, ohttp_ctx).save(persister)?;
        println!(
            "Response successful. Watch mempool for successful Payjoin. TXID: {}",
            payjoin_psbt.extract_tx_unchecked_fee_rate().clone().compute_txid()
        );
        Ok(())
    }

    async fn unwrap_relay_or_else_fetch(
        &self,
        directory: Option<payjoin::Url>,
    ) -> Result<payjoin::Url> {
        let selected_relay =
            self.relay_manager.lock().expect("Lock should not be poisoned").get_selected_relay();
        let ohttp_relay = match selected_relay {
            Some(relay) => relay,
            None =>
                unwrap_ohttp_keys_or_else_fetch(&self.config, directory, self.relay_manager.clone())
                    .await?
                    .relay_url,
        };
        Ok(ohttp_relay)
    }
}

/// Handle request error by sending an error response over the directory
async fn handle_recoverable_error(
    ohttp_relay: &payjoin::Url,
    session_history: &SessionHistory,
) -> Result<()> {
    let e = match session_history.terminal_error() {
        Some((_, Some(e))) => e,
        _ => return Ok(()),
    };
    let (err_req, err_ctx) = session_history
        .extract_err_req(ohttp_relay)?
        .expect("If JsonReply is Some, then err_req and err_ctx should be Some");
    let to_return = anyhow!("Replied with error: {}", e.to_json().to_string());

    let err_response = match post_request(err_req).await {
        Ok(response) => response,
        Err(e) => return Err(anyhow!("Failed to post error request: {}", e)),
    };

    let err_bytes = match err_response.bytes().await {
        Ok(bytes) => bytes,
        Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)),
    };

    if let Err(e) = process_err_res(&err_bytes, err_ctx) {
        return Err(anyhow!("Failed to process error response: {}", e));
    }

    Err(to_return)
}

async fn post_request(req: payjoin::Request) -> Result<reqwest::Response> {
    let http = http_agent()?;
    http.post(req.url)
        .header("Content-Type", req.content_type)
        .body(req.body)
        .send()
        .await
        .map_err(map_reqwest_err)
}

fn map_reqwest_err(e: reqwest::Error) -> anyhow::Error {
    match e.status() {
        Some(status_code) => anyhow!("HTTP request failed: {} {}", status_code, e),
        None => anyhow!("No HTTP response: {}", e),
    }
}

1	use std::sync::{Arc, Mutex};
2
3	use anyhow::{anyhow, Context, Result};
4	use payjoin::bitcoin::consensus::encode::serialize_hex;
5	use payjoin::bitcoin::psbt::Psbt;
6	use payjoin::bitcoin::{Amount, FeeRate};
7	use payjoin::persist::OptionalTransitionOutcome;
8	use payjoin::receive::v2::{
9	process_err_res, replay_event_log as replay_receiver_event_log, MaybeInputsOwned,
10	MaybeInputsSeen, OutputsUnknown, PayjoinProposal, ProvisionalProposal, Receiver,
11	ReceiverTypeState, SessionHistory, UncheckedProposal, WantsInputs, WantsOutputs, WithContext,
12	};
13	use payjoin::send::v2::{Sender, SenderBuilder, WithReplyKey};
14	use payjoin::Uri;
15	use tokio::sync::watch;
16
17	use super::config::Config;
18	use super::wallet::BitcoindWallet;
19	use super::App as AppTrait;
20	use crate::app::v2::ohttp::{unwrap_ohttp_keys_or_else_fetch, RelayManager};
21	use crate::app::{handle_interrupt, http_agent};
22	use crate::db::v2::{ReceiverPersister, SenderPersister};
23	use crate::db::Database;
24
25	mod ohttp;
26
27	#[derive(Clone)]
28	pub(crate) struct App {
29	config: Config,
30	db: Arc<Database>,
31	wallet: BitcoindWallet,
32	interrupt: watch::Receiver<()>,
33	relay_manager: Arc<Mutex<RelayManager>>,
34	}
35
36	#[async_trait::async_trait]
37	impl AppTrait for App {
38	fn new(config: Config) -> Result<Self> {	4✔
39	let db = Arc::new(Database::create(&config.db_path)?);	4✔
40	let relay_manager = Arc::new(Mutex::new(RelayManager::new()));	4✔
41	let (interrupt_tx, interrupt_rx) = watch::channel(());	4✔
42	tokio::spawn(handle_interrupt(interrupt_tx));	4✔
43	let wallet = BitcoindWallet::new(&config.bitcoind)?;	4✔
44	let app = Self { config, db, wallet, interrupt: interrupt_rx, relay_manager };	4✔
45	app.wallet()	4✔
46	.network()	4✔
47	.context("Failed to connect to bitcoind. Check config RPC connection.")?;	4✔
48	Ok(app)	4✔
49	}	4✔
50
51	fn wallet(&self) -> BitcoindWallet { self.wallet.clone() }	14✔
52
53	async fn send_payjoin(&self, bip21: &str, fee_rate: FeeRate) -> Result<()> {	2✔
54	use payjoin::UriExt;
55	let uri =	2✔
56	Uri::try_from(bip21).map_err(\|e\| anyhow!("Failed to create URI from BIP21: {}", e))?;	2✔
57	let uri = uri.assume_checked();	2✔
58	let uri = uri.check_pj_supported().map_err(\|_\| anyhow!("URI does not support Payjoin"))?;	2✔
59	let url = uri.extras.endpoint();	2✔
60	// match bip21 to send_session public_key
61	let req_ctx = match self.db.get_send_session(url)? {	2✔
62	Some(send_session) => send_session,	1✔
63	None => {
64	let psbt = self.create_original_psbt(&uri, fee_rate)?;	1✔
65	let mut persister = SenderPersister::new(self.db.clone());	1✔
66	let new_sender = SenderBuilder::new(psbt, uri.clone())	1✔
67	.build_recommended(fee_rate)	1✔
68	.with_context(\|\| "Failed to build payjoin request")?;	1✔
69	let storage_token = new_sender	1✔
70	.persist(&mut persister)	1✔
71	.map_err(\|e\| anyhow!("Failed to persist sender: {}", e))?;	1✔
72	Sender::load(storage_token, &persister)	1✔
73	.map_err(\|e\| anyhow!("Failed to load sender: {}", e))?	1✔
74	}
75	};
76	self.spawn_payjoin_sender(req_ctx).await	2✔
77	}	4✔
78
79	async fn receive_payjoin(&self, amount: Amount) -> Result<()> {	1✔
80	let address = self.wallet().get_new_address()?;	1✔
81	let ohttp_keys =	1✔
82	unwrap_ohttp_keys_or_else_fetch(&self.config, None, self.relay_manager.clone())	1✔
83	.await?	1✔
84	.ohttp_keys;
85	let persister = ReceiverPersister::new(self.db.clone())?;	1✔
86	let session = Receiver::create_session(	1✔
87	address,	1✔
88	self.config.v2()?.pj_directory.clone(),	1✔
89	ohttp_keys,	1✔
90	None,	1✔
91	)	1✔
92	.save(&persister)?;	1✔
93	println!("Receive session established");	1✔
94	let mut pj_uri = session.pj_uri();	1✔
95	pj_uri.amount = Some(amount);	1✔
96	println!("Request Payjoin by sharing this Payjoin Uri:");	1✔
97	println!("{}", pj_uri);	1✔
98		1✔
99	self.process_receiver_session(ReceiverTypeState::WithContext(session.clone()), &persister)	1✔
100	.await?;	1✔
NEW 101	Ok(())	×
102	}	2✔
103
104	#[allow(clippy::incompatible_msrv)]
105	async fn resume_payjoins(&self) -> Result<()> {	1✔
106	let recv_session_ids = self.db.get_recv_session_ids()?;	1✔
107	let send_sessions = self.db.get_send_sessions()?;	1✔
108
109	if recv_session_ids.is_empty() && send_sessions.is_empty() {	1✔
110	println!("No sessions to resume.");	×
111	return Ok(());	×
112	}	1✔
113		1✔
114	let mut tasks = Vec::new();	1✔
115
116	for session_id in recv_session_ids {	2✔
117	let self_clone = self.clone();	1✔
118	let recv_persister = ReceiverPersister::from_id(self.db.clone(), session_id)?;	1✔
119	let receiver_state = replay_receiver_event_log(&recv_persister)	1✔
120	.map_err(\|e\| anyhow!("Failed to replay receiver event log: {:?}", e))?	1✔
121	.0;
122	tasks.push(tokio::spawn(async move {	1✔
123	self_clone.process_receiver_session(receiver_state, &recv_persister).await	1✔
124	}));	1✔
125	}	1✔
126
127	for session in send_sessions {	1✔
128	let self_clone = self.clone();	×
129	tasks.push(tokio::spawn(async move { self_clone.spawn_payjoin_sender(session).await }));	×
130	}	×
131
132	let mut interrupt = self.interrupt.clone();	1✔
133	tokio::select! {	1✔
134	_ = async {	1✔
135	for task in tasks {	2✔
136	let _ = task.await;	1✔
137	}
138	} => {	1✔
139	println!("All resumed sessions completed.");	1✔
140	}	1✔
141	_ = interrupt.changed() => {	1✔
UNCOV 142	println!("Resumed sessions were interrupted.");	×
UNCOV 143	}	×
144	}
145	Ok(())	1✔
146	}	2✔
147	}
148
149	impl App {
150	#[allow(clippy::incompatible_msrv)]
151	async fn spawn_payjoin_sender(&self, mut req_ctx: Sender<WithReplyKey>) -> Result<()> {	2✔
152	let mut interrupt = self.interrupt.clone();	2✔
153	tokio::select! {	2✔
154	res = self.long_poll_post(&mut req_ctx) => {	2✔
155	self.process_pj_response(res?)?;	1✔
156	self.db.clear_send_session(req_ctx.endpoint())?;	1✔
157	}
158	_ = interrupt.changed() => {	2✔
159	println!("Interrupted. Call `send` with the same arguments to resume this session or `resume` to resume all sessions.");	1✔
160	}	1✔
161	}
162	Ok(())	2✔
163	}	2✔
164
165	async fn long_poll_post(&self, req_ctx: &mut Sender<WithReplyKey>) -> Result<Psbt> {	2✔
166	let ohttp_relay = self.unwrap_relay_or_else_fetch(Some(req_ctx.endpoint().clone())).await?;	2✔
167
168	match req_ctx.extract_v2(ohttp_relay.clone()) {	2✔
169	Ok((req, ctx)) => {	2✔
170	println!("Posting Original PSBT Payload request...");	2✔
171	let response = post_request(req).await?;	2✔
172	println!("Sent fallback transaction");	2✔
173	let v2_ctx = Arc::new(ctx.process_response(&response.bytes().await?)?);	2✔
174	loop {
175	let (req, ohttp_ctx) = v2_ctx.extract_req(&ohttp_relay)?;	3✔
176	let response = post_request(req).await?;	3✔
177	match v2_ctx.process_response(&response.bytes().await?, ohttp_ctx) {	2✔
178	Ok(Some(psbt)) => return Ok(psbt),	1✔
179	Ok(None) => {	1✔
180	println!("No response yet.");	1✔
181	}	1✔
182	Err(re) => {	×
183	println!("{re}");	×
184	log::debug!("{re:?}");	×
185	return Err(anyhow!("Response error").context(re));	×
186	}
187	}
188	}
189	}
190	Err(_) => {
191	let (req, v1_ctx) = req_ctx.extract_v1();	×
192	println!("Posting Original PSBT Payload request...");	×
193	let response = post_request(req).await?;	×
194	println!("Sent fallback transaction");	×
195	match v1_ctx.process_response(&response.bytes().await?) {	×
196	Ok(psbt) => Ok(psbt),	×
197	Err(re) => {	×
198	println!("{re}");	×
199	log::debug!("{re:?}");	×
200	Err(anyhow!("Response error").context(re))	×
201	}
202	}
203	}
204	}
205	}	1✔
206
207	async fn long_poll_fallback(	2✔
208	&self,	2✔
209	session: Receiver<WithContext>,	2✔
210	persister: &ReceiverPersister,	2✔
211	) -> Result<Receiver<UncheckedProposal>> {	2✔
212	let ohttp_relay = self	2✔
213	.unwrap_relay_or_else_fetch(Some(session.pj_uri().extras.endpoint().clone()))	2✔
214	.await?;	2✔
215
216	let mut session = session;	2✔
217	loop {
218	let (req, context) = session.extract_req(&ohttp_relay)?;	2✔
219	println!("Polling receive request...");	2✔
220	let ohttp_response = post_request(req).await?;	2✔
221	let state_transition = session	1✔
222	.process_res(ohttp_response.bytes().await?.to_vec().as_slice(), context)	1✔
223	.save(persister);	1✔
224	match state_transition {	1✔
225	Ok(OptionalTransitionOutcome::Progress(next_state)) => {	1✔
226	println!("Got a request from the sender. Responding with a Payjoin proposal.");	1✔
227	return Ok(next_state);	1✔
228	}
NEW 229	Ok(OptionalTransitionOutcome::Stasis(current_state)) => {	×
NEW 230	session = current_state;	×
NEW 231	continue;	×
232	}
NEW 233	Err(e) => return Err(e.into()),	×
234	}
235	}
236	}	1✔
237
238	async fn process_receiver_session(	2✔
239	&self,	2✔
240	session: ReceiverTypeState,	2✔
241	persister: &ReceiverPersister,	2✔
242	) -> Result<()> {	2✔
243	let res = {	2✔
244	match session {	2✔
245	ReceiverTypeState::WithContext(proposal) =>	2✔
246	self.read_from_directory(proposal, persister).await,	2✔
NEW 247	ReceiverTypeState::UncheckedProposal(proposal) =>	×
NEW 248	self.check_proposal(proposal, persister).await,	×
NEW 249	ReceiverTypeState::MaybeInputsOwned(proposal) =>	×
NEW 250	self.check_inputs_not_owned(proposal, persister).await,	×
NEW 251	ReceiverTypeState::MaybeInputsSeen(proposal) =>	×
NEW 252	self.check_no_inputs_seen_before(proposal, persister).await,	×
NEW 253	ReceiverTypeState::OutputsUnknown(proposal) =>	×
NEW 254	self.identify_receiver_outputs(proposal, persister).await,	×
NEW 255	ReceiverTypeState::WantsOutputs(proposal) =>	×
NEW 256	self.commit_outputs(proposal, persister).await,	×
NEW 257	ReceiverTypeState::WantsInputs(proposal) =>	×
NEW 258	self.contribute_inputs(proposal, persister).await,	×
NEW 259	ReceiverTypeState::ProvisionalProposal(proposal) =>	×
NEW 260	self.finalize_proposal(proposal, persister).await,	×
NEW 261	ReceiverTypeState::PayjoinProposal(proposal) =>	×
NEW 262	self.send_payjoin_proposal(proposal, persister).await,	×
263	ReceiverTypeState::Uninitialized(_) =>
NEW 264	return Err(anyhow!("Uninitialized receiver session")),	×
265	ReceiverTypeState::TerminalState =>
NEW 266	return Err(anyhow!("Terminal receiver session")),	×
267	}
268	};
269
270	match res {	2✔
271	Ok(_) => Ok(()),	1✔
272	Err(e) => {	1✔
273	let (_, session_history) = replay_receiver_event_log(persister)?;	1✔
274	let pj_uri = match session_history.pj_uri() {	1✔
275	Some(uri) => Some(uri.extras.endpoint().clone()),	1✔
NEW 276	None => None,	×
277	};
278	let ohttp_relay = self.unwrap_relay_or_else_fetch(pj_uri).await?;	1✔
279	handle_recoverable_error(&ohttp_relay, &session_history).await?;	1✔
280
281	Err(e)	1✔
282	}
283	}
284	}	2✔
285
286	#[allow(clippy::incompatible_msrv)]
287	async fn read_from_directory(	2✔
288	&self,	2✔
289	session: Receiver<WithContext>,	2✔
290	persister: &ReceiverPersister,	2✔
291	) -> Result<()> {	2✔
292	let mut interrupt = self.interrupt.clone();	2✔
293	let receiver = tokio::select! {	2✔
294	res = self.long_poll_fallback(session, persister) => res,	2✔
295	_ = interrupt.changed() => {	2✔
296	println!("Interrupted. Call the `resume` command to resume all sessions.");	1✔
297	return Err(anyhow!("Interrupted"));	1✔
298	}
NEW 299	}?;	×
300	println!("Fallback transaction received. Consider broadcasting this to get paid if the Payjoin fails:");	1✔
301	println!("{}", serialize_hex(&receiver.extract_tx_to_schedule_broadcast()));	1✔
302	self.check_proposal(receiver, persister).await	1✔
303	}	2✔
304
305	async fn check_proposal(	1✔
306	&self,	1✔
307	proposal: Receiver<UncheckedProposal>,	1✔
308	persister: &ReceiverPersister,	1✔
309	) -> Result<()> {	1✔
310	let wallet = self.wallet();	1✔
311	let proposal = proposal	1✔
312	.check_broadcast_suitability(None, \|tx\| Ok(wallet.can_broadcast(tx)?))	1✔
313	.save(persister)?;	1✔
314
315	self.check_inputs_not_owned(proposal, persister).await	1✔
316	}	1✔
317
318	async fn check_inputs_not_owned(	1✔
319	&self,	1✔
320	proposal: Receiver<MaybeInputsOwned>,	1✔
321	persister: &ReceiverPersister,	1✔
322	) -> Result<()> {	1✔
323	let wallet = self.wallet();	1✔
324	let proposal =	1✔
325	proposal.check_inputs_not_owned(\|input\| Ok(wallet.is_mine(input)?)).save(persister)?;	1✔
326	self.check_no_inputs_seen_before(proposal, persister).await	1✔
327	}	1✔
328
329	async fn check_no_inputs_seen_before(	1✔
330	&self,	1✔
331	proposal: Receiver<MaybeInputsSeen>,	1✔
332	persister: &ReceiverPersister,	1✔
333	) -> Result<()> {	1✔
334	let proposal = proposal	1✔
335	.check_no_inputs_seen_before(\|input\| Ok(self.db.insert_input_seen_before(*input)?))	1✔
336	.save(persister)?;	1✔
337	self.identify_receiver_outputs(proposal, persister).await	1✔
338	}	1✔
339
340	async fn identify_receiver_outputs(	1✔
341	&self,	1✔
342	proposal: Receiver<OutputsUnknown>,	1✔
343	persister: &ReceiverPersister,	1✔
344	) -> Result<()> {	1✔
345	let wallet = self.wallet();	1✔
346	let proposal = proposal	1✔
347	.identify_receiver_outputs(\|output_script\| Ok(wallet.is_mine(output_script)?))	2✔
348	.save(persister)?;	1✔
349	self.commit_outputs(proposal, persister).await	1✔
350	}	1✔
351
352	async fn commit_outputs(	1✔
353	&self,	1✔
354	proposal: Receiver<WantsOutputs>,	1✔
355	persister: &ReceiverPersister,	1✔
356	) -> Result<()> {	1✔
357	let proposal = proposal.commit_outputs().save(persister)?;	1✔
358	self.contribute_inputs(proposal, persister).await	1✔
359	}	1✔
360
361	async fn contribute_inputs(	1✔
362	&self,	1✔
363	proposal: Receiver<WantsInputs>,	1✔
364	persister: &ReceiverPersister,	1✔
365	) -> Result<()> {	1✔
366	let wallet = self.wallet();	1✔
367	let candidate_inputs = wallet.list_unspent()?;	1✔
368
369	let selected_input = proposal.try_preserving_privacy(candidate_inputs)?;	1✔
370	let proposal =	1✔
371	proposal.contribute_inputs(vec![selected_input])?.commit_inputs().save(persister)?;	1✔
372	self.finalize_proposal(proposal, persister).await	1✔
373	}	1✔
374
375	async fn finalize_proposal(	1✔
376	&self,	1✔
377	proposal: Receiver<ProvisionalProposal>,	1✔
378	persister: &ReceiverPersister,	1✔
379	) -> Result<()> {	1✔
380	let wallet = self.wallet();	1✔
381	let proposal = proposal	1✔
382	.finalize_proposal(	1✔
383	\|psbt\| Ok(wallet.process_psbt(psbt)?),	1✔
384	None,	1✔
385	self.config.max_fee_rate,	1✔
386	)	1✔
387	.save(persister)?;	1✔
388	self.send_payjoin_proposal(proposal, persister).await	1✔
389	}	1✔
390
391	async fn send_payjoin_proposal(	1✔
392	&self,	1✔
393	mut proposal: Receiver<PayjoinProposal>,	1✔
394	persister: &ReceiverPersister,	1✔
395	) -> Result<()> {	1✔
396	let (req, ohttp_ctx) = proposal	1✔
397	.extract_req(&self.unwrap_relay_or_else_fetch(None).await?)	1✔
398	.map_err(\|e\| anyhow!("v2 req extraction failed {}", e))?;	1✔
399	let res = post_request(req).await?;	1✔
400	let payjoin_psbt = proposal.psbt().clone();	1✔
401	proposal.process_res(&res.bytes().await?, ohttp_ctx).save(persister)?;	1✔
402	println!(	1✔
403	"Response successful. Watch mempool for successful Payjoin. TXID: {}",	1✔
404	payjoin_psbt.extract_tx_unchecked_fee_rate().clone().compute_txid()	1✔
405	);	1✔
406	Ok(())	1✔
407	}	1✔
408
409	async fn unwrap_relay_or_else_fetch(	6✔
410	&self,	6✔
411	directory: Option<payjoin::Url>,	6✔
412	) -> Result<payjoin::Url> {	6✔
413	let selected_relay =	6✔
414	self.relay_manager.lock().expect("Lock should not be poisoned").get_selected_relay();	6✔
415	let ohttp_relay = match selected_relay {	6✔
416	Some(relay) => relay,	1✔
417	None =>
418	unwrap_ohttp_keys_or_else_fetch(&self.config, directory, self.relay_manager.clone())	5✔
419	.await?	5✔
420	.relay_url,
421	};
422	Ok(ohttp_relay)	6✔
423	}	6✔
424	}
425
426	/// Handle request error by sending an error response over the directory
427	async fn handle_recoverable_error(	1✔
428	ohttp_relay: &payjoin::Url,	1✔
429	session_history: &SessionHistory,	1✔
430	) -> Result<()> {	1✔
431	let e = match session_history.terminal_error() {	1✔
NEW 432	Some((_, Some(e))) => e,	×
433	_ => return Ok(()),	1✔
434	};
NEW 435	let (err_req, err_ctx) = session_history	×
NEW 436	.extract_err_req(ohttp_relay)?	×
NEW 437	.expect("If JsonReply is Some, then err_req and err_ctx should be Some");	×
NEW 438	let to_return = anyhow!("Replied with error: {}", e.to_json().to_string());	×
439
440	let err_response = match post_request(err_req).await {	×
441	Ok(response) => response,	×
NEW 442	Err(e) => return Err(anyhow!("Failed to post error request: {}", e)),	×
443	};
444
445	let err_bytes = match err_response.bytes().await {	×
446	Ok(bytes) => bytes,	×
NEW 447	Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)),	×
448	};
449
NEW 450	if let Err(e) = process_err_res(&err_bytes, err_ctx) {	×
NEW 451	return Err(anyhow!("Failed to process error response: {}", e));	×
452	}	×
453		×
NEW 454	Err(to_return)	×
455	}	1✔
456
457	async fn post_request(req: payjoin::Request) -> Result<reqwest::Response> {	8✔
458	let http = http_agent()?;	8✔
459	http.post(req.url)	8✔
460	.header("Content-Type", req.content_type)	8✔
461	.body(req.body)	8✔
462	.send()	8✔
463	.await	8✔
464	.map_err(map_reqwest_err)	6✔
465	}	6✔
466
467	fn map_reqwest_err(e: reqwest::Error) -> anyhow::Error {	×
468	match e.status() {	×
469	Some(status_code) => anyhow!("HTTP request failed: {} {}", status_code, e),	×
470	None => anyhow!("No HTTP response: {}", e),	×
471	}
472	}	×

payjoin / rust-payjoin / 15767995657

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