Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 7 additions & 13 deletions payjoin-cli/src/app/v2/mod.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
use std::fmt;
use std::sync::{Arc, Mutex};
use std::sync::Arc;

use anyhow::{anyhow, Context, Result};
use payjoin::bitcoin::consensus::encode::serialize_hex;
Expand All @@ -23,7 +23,7 @@ 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::v2::ohttp::RelayManager;
use crate::app::{handle_interrupt, http_agent};
use crate::cli::Role as CliRole;
use crate::db::v2::{ReceiverPersister, SenderPersister, SessionId};
Expand All @@ -42,7 +42,7 @@ pub(crate) struct App {
db: Arc<Database>,
wallet: BitcoindWallet,
interrupt: watch::Receiver<()>,
relay_manager: Arc<Mutex<RelayManager>>,
relay_manager: RelayManager,
}

trait StatusText {
Expand Down Expand Up @@ -142,7 +142,7 @@ 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 relay_manager = Arc::new(Mutex::new(RelayManager::new(config.clone())));
let relay_manager = RelayManager::new(config.clone());
let (interrupt_tx, interrupt_rx) = watch::channel(());
tokio::spawn(handle_interrupt(interrupt_tx));
let wallet = BitcoindWallet::new(&config.bitcoind).await?;
Expand Down Expand Up @@ -278,9 +278,7 @@ impl AppTrait for App {

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, self.relay_manager.clone())
.await?
.ohttp_keys;
let ohttp_keys = self.relay_manager.unwrap_ohttp_keys_or_else_fetch().await?.ohttp_keys;
let persister = ReceiverPersister::new(self.db.clone())?;
let mut receiver_builder =
ReceiverBuilder::new(address, self.config.v2()?.pj_directory.as_str(), ohttp_keys)?
Expand Down Expand Up @@ -1059,17 +1057,13 @@ impl App {
E: Into<anyhow::Error>,
{
loop {
let relay =
self.relay_manager.lock().expect("Lock should not be poisoned").choose_relay()?;
let relay = self.relay_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.relay_manager
.lock()
.expect("Lock should not be poisoned")
.add_failed_relay(relay);
self.relay_manager.add_failed_relay(relay);
}
}
}
Expand Down
112 changes: 54 additions & 58 deletions payjoin-cli/src/app/v2/ohttp.rs
Original file line number Diff line number Diff line change
Expand Up @@ -17,19 +17,24 @@ use super::Config;
#[derive(Debug, Clone)]
pub struct RelayManager {
config: Config,
failed_relays: Vec<Url>,
failed_relays: Arc<Mutex<Vec<Url>>>,
}

impl RelayManager {
pub fn new(config: Config) -> Self { RelayManager { config, failed_relays: Vec::new() } }
pub fn new(config: Config) -> Self {
RelayManager { config, failed_relays: Arc::new(Mutex::new(Vec::new())) }
}

pub fn add_failed_relay(&mut self, relay: Url) { self.failed_relays.push(relay); }
pub fn add_failed_relay(&self, relay: Url) {
self.failed_relays.lock().expect("Lock should not be poisoned").push(relay);
}

pub fn choose_relay(&self) -> Result<Url> {
use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom;
let relays = self.config.v2()?.ohttp_relays.clone();
let relays = &self.config.v2()?.ohttp_relays;
let failed_relays = self.failed_relays.lock().expect("Lock should not be poisoned");
let remaining_relays: Vec<_> =
relays.iter().filter(|r| !self.failed_relays.contains(r)).cloned().collect();
relays.iter().filter(|r| !failed_relays.contains(r)).cloned().collect();

if remaining_relays.is_empty() {
return Err(anyhow!("No valid relays available"));
Expand All @@ -40,67 +45,58 @@ impl RelayManager {
.cloned()
.ok_or_else(|| anyhow!("Failed to select from remaining relays"))
}
}

pub(crate) struct ValidatedOhttpKeys {
pub(crate) ohttp_keys: payjoin::OhttpKeys,
}

pub(crate) async fn unwrap_ohttp_keys_or_else_fetch(
config: &Config,
relay_manager: Arc<Mutex<RelayManager>>,
) -> Result<ValidatedOhttpKeys> {
if let Some(ohttp_keys) = config.v2()?.ohttp_keys.clone() {
return Ok(ValidatedOhttpKeys { ohttp_keys });
pub async fn unwrap_ohttp_keys_or_else_fetch(&self) -> Result<ValidatedOhttpKeys> {
Comment thread
benalleng marked this conversation as resolved.
Outdated
if let Some(ohttp_keys) = self.config.v2()?.ohttp_keys.clone() {
return Ok(ValidatedOhttpKeys { ohttp_keys });
}
self.fetch_ohttp_keys().await
}
fetch_ohttp_keys(config, relay_manager).await
}

async fn fetch_ohttp_keys(
config: &Config,
relay_manager: Arc<Mutex<RelayManager>>,
) -> Result<ValidatedOhttpKeys> {
let payjoin_directory = config.v2()?.pj_directory.clone();
async fn fetch_ohttp_keys(&self) -> Result<ValidatedOhttpKeys> {
let payjoin_directory = &self.config.v2()?.pj_directory;

loop {
let selected_relay =
relay_manager.lock().expect("Lock should not be poisoned").choose_relay()?;
loop {
let selected_relay = self.choose_relay()?;

let ohttp_keys = {
#[cfg(feature = "_manual-tls")]
{
if let Some(cert_path) = config.root_certificate.as_ref() {
let cert_der = std::fs::read(cert_path)?;
payjoin::io::fetch_ohttp_keys_with_cert(
selected_relay.as_str(),
payjoin_directory.as_str(),
&cert_der,
)
.await
} else {
payjoin::io::fetch_ohttp_keys(
selected_relay.as_str(),
payjoin_directory.as_str(),
)
.await
let ohttp_keys = {
#[cfg(feature = "_manual-tls")]
{
if let Some(cert_path) = self.config.root_certificate.as_ref() {
let cert_der = std::fs::read(cert_path)?;
payjoin::io::fetch_ohttp_keys_with_cert(
selected_relay.as_str(),
payjoin_directory.as_str(),
&cert_der,
)
.await
} else {
payjoin::io::fetch_ohttp_keys(
selected_relay.as_str(),
payjoin_directory.as_str(),
)
.await
}
}
}
#[cfg(not(feature = "_manual-tls"))]
payjoin::io::fetch_ohttp_keys(selected_relay.as_str(), payjoin_directory.as_str()).await
};
#[cfg(not(feature = "_manual-tls"))]
payjoin::io::fetch_ohttp_keys(selected_relay.as_str(), payjoin_directory.as_str())
.await
};

match ohttp_keys {
Ok(keys) => return Ok(ValidatedOhttpKeys { ohttp_keys: keys }),
Err(payjoin::io::Error::UnexpectedStatusCode(e)) => {
return Err(payjoin::io::Error::UnexpectedStatusCode(e).into());
}
Err(e) => {
tracing::debug!("Failed to connect to relay: {selected_relay}, {e:?}");
relay_manager
.lock()
.expect("Lock should not be poisoned")
.add_failed_relay(selected_relay);
match ohttp_keys {
Ok(keys) => return Ok(ValidatedOhttpKeys { ohttp_keys: keys }),
Err(payjoin::io::Error::UnexpectedStatusCode(e)) => {
return Err(payjoin::io::Error::UnexpectedStatusCode(e).into());
}
Err(e) => {
tracing::debug!("Failed to connect to relay: {selected_relay}, {e:?}");
self.add_failed_relay(selected_relay);
}
}
}
}
}

pub(crate) struct ValidatedOhttpKeys {
pub(crate) ohttp_keys: payjoin::OhttpKeys,
}
Loading