Repository navigation
Split resume session relays #1399
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,5 +1,5 @@ | ||
| use std::fmt; | ||
| use std::sync::Arc; | ||
|
|
||
| use anyhow::{anyhow, Context, Result}; | ||
| use payjoin::bitcoin::consensus::encode::serialize_hex; | ||
|
|
@@ -40,7 +40,6 @@ pub(crate) struct App { | |
| db: Arc<Database>, | ||
| wallet: BitcoindWallet, | ||
| interrupt: watch::Receiver<()>, | ||
| } | ||
|
|
||
| trait StatusText { | ||
|
|
@@ -141,11 +140,10 @@ impl<Status: StatusText> fmt::Display for SessionHistoryRow<Status> { | |
| impl AppTrait for App { | ||
| async fn new(config: Config) -> Result<Self> { | ||
| let db = Arc::new(Database::create(&config.db_path)?); | ||
| 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 }; | ||
| app.wallet() | ||
| .network() | ||
| .context("Failed to connect to bitcoind. Check config RPC connection.")?; | ||
|
|
@@ -154,7 +152,6 @@ impl AppTrait for 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) | ||
|
|
@@ -277,10 +274,10 @@ impl AppTrait for App { | |
|
|
||
| async fn receive_payjoin(&self, amount: Amount) -> Result<()> { | ||
| let address = self.wallet().get_new_address()?; | ||
| let mut relay_manager = RelayManager::new(); | ||
| let ohttp_keys = unwrap_ohttp_keys_or_else_fetch(&self.config, None, &mut relay_manager) | ||
| .await? | ||
| .ohttp_keys; | ||
| let persister = ReceiverPersister::new(self.db.clone())?; | ||
| let session = | ||
| ReceiverBuilder::new(address, self.config.v2()?.pj_directory.as_str(), ohttp_keys)? | ||
|
|
@@ -559,11 +556,12 @@ impl App { | |
| session: SendSession, | ||
| persister: &SenderPersister, | ||
| ) -> Result<()> { | ||
| let mut relay_manager = RelayManager::new(); | ||
| match session { | ||
| SendSession::WithReplyKey(context) => | ||
| self.post_original_proposal(context, persister, &mut relay_manager).await?, | ||
| SendSession::PollingForProposal(context) => | ||
| self.get_proposed_payjoin_psbt(context, persister, &mut relay_manager).await?, | ||
| SendSession::Closed(SenderSessionOutcome::Success(proposal)) => { | ||
| self.process_pj_response(proposal)?; | ||
| return Ok(()); | ||
|
|
@@ -588,21 +586,27 @@ impl App { | |
| &self, | ||
| sender: Sender<WithReplyKey>, | ||
| persister: &SenderPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<()> { | ||
| let (req, ctx) = sender.create_v2_post_request( | ||
| self.unwrap_relay_or_else_fetch(Some(&sender.endpoint()), relay_manager) | ||
| .await? | ||
| .as_str(), | ||
| )?; | ||
| let response = self.post_request(req).await?; | ||
| let sender = sender.process_response(&response.bytes().await?, ctx).save(persister)?; | ||
| println!("Posted Original PSBT..."); | ||
| self.get_proposed_payjoin_psbt(sender, persister, relay_manager).await | ||
| } | ||
|
|
||
| async fn get_proposed_payjoin_psbt( | ||
| &self, | ||
| sender: Sender<PollingForProposal>, | ||
| persister: &SenderPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<()> { | ||
| let ohttp_relay = | ||
| self.unwrap_relay_or_else_fetch(Some(&sender.endpoint()), relay_manager).await?; | ||
| let mut session = sender.clone(); | ||
| // Long poll until we get a response | ||
| loop { | ||
|
|
@@ -633,9 +637,11 @@ impl App { | |
| &self, | ||
| session: Receiver<Initialized>, | ||
| persister: &ReceiverPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<Receiver<UncheckedOriginalPayload>> { | ||
| let ohttp_relay = self | ||
| .unwrap_relay_or_else_fetch(Some(&session.pj_uri().extras.endpoint()), relay_manager) | ||
| .await?; | ||
|
Comment on lines
+642
to
+644
Collaborator
There was a problem hiding this comment. Unrelated to this PR, but shouldn't this relay selection be inside the loop so that a different relay can be used for each poll?
Collaborator
Author
There was a problem hiding this comment. Yes, though with a lot of the discussion from #919 I do want to be sure that this is not going to be immediately overwritten or cause some additional complexity in the refactor so I am trying to keep these PRs as minimal as possible before we transition to the AS-aware relay selection. |
||
|
|
||
| let mut session = session; | ||
| loop { | ||
|
|
@@ -664,30 +670,31 @@ impl App { | |
| session: ReceiveSession, | ||
| persister: &ReceiverPersister, | ||
| ) -> Result<()> { | ||
| let mut relay_manager = RelayManager::new(); | ||
| let res = { | ||
| match session { | ||
| ReceiveSession::Initialized(proposal) => | ||
| self.read_from_directory(proposal, persister, &mut relay_manager).await, | ||
| ReceiveSession::UncheckedOriginalPayload(proposal) => | ||
| self.check_proposal(proposal, persister, &mut relay_manager).await, | ||
| ReceiveSession::MaybeInputsOwned(proposal) => | ||
| self.check_inputs_not_owned(proposal, persister, &mut relay_manager).await, | ||
| ReceiveSession::MaybeInputsSeen(proposal) => | ||
| self.check_no_inputs_seen_before(proposal, persister, &mut relay_manager).await, | ||
| ReceiveSession::OutputsUnknown(proposal) => | ||
| self.identify_receiver_outputs(proposal, persister, &mut relay_manager).await, | ||
| ReceiveSession::WantsOutputs(proposal) => | ||
| self.commit_outputs(proposal, persister, &mut relay_manager).await, | ||
| ReceiveSession::WantsInputs(proposal) => | ||
| self.contribute_inputs(proposal, persister, &mut relay_manager).await, | ||
| ReceiveSession::WantsFeeRange(proposal) => | ||
| self.apply_fee_range(proposal, persister, &mut relay_manager).await, | ||
| ReceiveSession::ProvisionalProposal(proposal) => | ||
| self.finalize_proposal(proposal, persister, &mut relay_manager).await, | ||
| ReceiveSession::PayjoinProposal(proposal) => | ||
| self.send_payjoin_proposal(proposal, persister, &mut relay_manager).await, | ||
| ReceiveSession::HasReplyableError(error) => | ||
| self.handle_error(error, persister, &mut relay_manager).await, | ||
| ReceiveSession::Monitor(proposal) => | ||
| self.monitor_payjoin_proposal(proposal, persister).await, | ||
| ReceiveSession::Closed(_) => return Err(anyhow!("Session closed")), | ||
|
|
@@ -701,22 +708,24 @@ impl App { | |
| &self, | ||
| session: Receiver<Initialized>, | ||
| persister: &ReceiverPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<()> { | ||
| let mut interrupt = self.interrupt.clone(); | ||
| let receiver = tokio::select! { | ||
| res = self.long_poll_fallback(session, persister, relay_manager) => res, | ||
| _ = interrupt.changed() => { | ||
| println!("Interrupted. Call the `resume` command to resume all sessions."); | ||
| return Err(anyhow!("Interrupted")); | ||
| } | ||
| }?; | ||
| self.check_proposal(receiver, persister, relay_manager).await | ||
| } | ||
|
|
||
| async fn check_proposal( | ||
| &self, | ||
| proposal: Receiver<UncheckedOriginalPayload>, | ||
| persister: &ReceiverPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<()> { | ||
| let wallet = self.wallet(); | ||
| let proposal = proposal | ||
|
|
@@ -729,13 +738,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, relay_manager).await | ||
| } | ||
|
|
||
| async fn check_inputs_not_owned( | ||
| &self, | ||
| proposal: Receiver<MaybeInputsOwned>, | ||
| persister: &ReceiverPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<()> { | ||
| let wallet = self.wallet(); | ||
| let proposal = proposal | ||
|
|
@@ -745,26 +755,28 @@ impl App { | |
| .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error())) | ||
| }) | ||
| .save(persister)?; | ||
| self.check_no_inputs_seen_before(proposal, persister, relay_manager).await | ||
| } | ||
|
|
||
| async fn check_no_inputs_seen_before( | ||
| &self, | ||
| proposal: Receiver<MaybeInputsSeen>, | ||
| persister: &ReceiverPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> 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, relay_manager).await | ||
| } | ||
|
|
||
| async fn identify_receiver_outputs( | ||
| &self, | ||
| proposal: Receiver<OutputsUnknown>, | ||
| persister: &ReceiverPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<()> { | ||
| let wallet = self.wallet(); | ||
| let proposal = proposal | ||
|
|
@@ -774,22 +786,24 @@ impl App { | |
| .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error())) | ||
| }) | ||
| .save(persister)?; | ||
| self.commit_outputs(proposal, persister, relay_manager).await | ||
| } | ||
|
|
||
| async fn commit_outputs( | ||
| &self, | ||
| proposal: Receiver<WantsOutputs>, | ||
| persister: &ReceiverPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<()> { | ||
| let proposal = proposal.commit_outputs().save(persister)?; | ||
| self.contribute_inputs(proposal, persister, relay_manager).await | ||
| } | ||
|
|
||
| async fn contribute_inputs( | ||
| &self, | ||
| proposal: Receiver<WantsInputs>, | ||
| persister: &ReceiverPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<()> { | ||
| let wallet = self.wallet(); | ||
| let candidate_inputs = wallet.list_unspent()?; | ||
|
|
@@ -803,22 +817,24 @@ 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, relay_manager).await | ||
| } | ||
|
|
||
| async fn apply_fee_range( | ||
| &self, | ||
| proposal: Receiver<WantsFeeRange>, | ||
| persister: &ReceiverPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<()> { | ||
| let proposal = proposal.apply_fee_range(None, self.config.max_fee_rate).save(persister)?; | ||
| self.finalize_proposal(proposal, persister, relay_manager).await | ||
| } | ||
|
|
||
| async fn finalize_proposal( | ||
| &self, | ||
| proposal: Receiver<ProvisionalProposal>, | ||
| persister: &ReceiverPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<()> { | ||
| let wallet = self.wallet(); | ||
| let proposal = proposal | ||
|
|
@@ -828,16 +844,19 @@ impl App { | |
| .map_err(|e| ImplementationError::from(e.into_boxed_dyn_error())) | ||
| }) | ||
| .save(persister)?; | ||
| self.send_payjoin_proposal(proposal, persister, relay_manager).await | ||
| } | ||
|
|
||
| async fn send_payjoin_proposal( | ||
| &self, | ||
| proposal: Receiver<PayjoinProposal>, | ||
| persister: &ReceiverPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<()> { | ||
| let (req, ohttp_ctx) = proposal | ||
| .create_post_request( | ||
| self.unwrap_relay_or_else_fetch(None::<&str>, relay_manager).await?.as_str(), | ||
| ) | ||
| .map_err(|e| anyhow!("v2 req extraction failed {}", e))?; | ||
| let res = self.post_request(req).await?; | ||
| let payjoin_psbt = proposal.psbt().clone(); | ||
|
|
@@ -898,14 +917,13 @@ impl App { | |
| async fn unwrap_relay_or_else_fetch( | ||
| &self, | ||
| directory: Option<impl payjoin::IntoUrl>, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<payjoin::Url> { | ||
| let directory = directory.map(|url| url.into_url()).transpose()?; | ||
| let ohttp_relay = match relay_manager.get_selected_relay() { | ||
| Some(relay) => relay, | ||
| None => | ||
| unwrap_ohttp_keys_or_else_fetch(&self.config, directory, relay_manager) | ||
| .await? | ||
| .relay_url, | ||
| }; | ||
|
|
@@ -917,9 +935,11 @@ impl App { | |
| &self, | ||
| session: Receiver<HasReplyableError>, | ||
| persister: &ReceiverPersister, | ||
| relay_manager: &mut RelayManager, | ||
| ) -> Result<()> { | ||
| let (err_req, err_ctx) = session.create_error_request( | ||
| self.unwrap_relay_or_else_fetch(None::<&str>, relay_manager).await?.as_str(), | ||
| )?; | ||
|
|
||
| let err_response = match self.post_request(err_req).await { | ||
| Ok(response) => response, | ||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why do we need to pass a mut reference if we can get a relay manager from
self.relay_manager()?Not requesting a change, just a question
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
No you're right the current behavior would I think just return back to caching the relay when it finds one that works during any long pollingNo actually I think I am getting mixed up, I'm jumping ahead towards per-request logic. I don't think the logic fundamentally changes though I think its unnecessary to bubble it up to these calls instead of right beforeunwrap_ohttp_keys_or_else_fetchThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@benalleng per your last comment. Are you still making this change?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think I misspoke earlier. the failures I was having the other day were due to the RelayManager::new() not being given a default to work without any context of the
unwrap_ohttp_keys_or_else_fetchlevel the RelayManager state was falling back to the payjo.in default thus causing a failure in our nix checksThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Final followup, the mut reference is needed here as long as we are still using the same relay per session. This can be removed but really only once we move to per-request relay selection. Though I want to move towards that I think this is not the PR to do it yet.