diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index 99d53e030..81d4ab9aa 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -130,7 +130,7 @@ impl fmt::Display for SessionHistoryRow { impl AppTrait for App { async fn new(config: Config) -> Result { let db = Arc::new(Database::create(&config.db_path)?); - let mailroom_manager = MailroomManager::new(config.clone()); + 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?; diff --git a/payjoin-cli/src/app/v2/ohttp.rs b/payjoin-cli/src/app/v2/ohttp.rs index e438540ec..d561e7c4a 100644 --- a/payjoin-cli/src/app/v2/ohttp.rs +++ b/payjoin-cli/src/app/v2/ohttp.rs @@ -14,6 +14,7 @@ use std::sync::{Arc, Mutex}; use anyhow::{anyhow, Result}; +use payjoin::relay::RelaySelector; use payjoin::Url; use super::Config; @@ -21,25 +22,26 @@ use super::Config; #[derive(Debug, Clone)] pub struct MailroomManager { config: Config, - failed_relays: Arc>>, + relay_selector: Arc>, failed_directories: Arc>>, } impl MailroomManager { - pub fn new(config: Config) -> Self { - MailroomManager { + pub fn new(config: Config) -> Result { + let relay_selector = RelaySelector::new(config.v2()?.ohttp_relays.clone()); + Ok(MailroomManager { config, - failed_relays: Arc::new(Mutex::new(Vec::new())), + relay_selector: Arc::new(Mutex::new(relay_selector)), failed_directories: Arc::new(Mutex::new(Vec::new())), - } + }) } pub fn add_failed_relay(&self, relay: Url) { - self.failed_relays.lock().expect("Lock should not be poisoned").push(relay); + self.relay_selector.lock().expect("Lock should not be poisoned").mark_failed(&relay); } pub fn clear_failed_relays(&self) { - self.failed_relays.lock().expect("Lock should not be poisoned").clear(); + self.relay_selector.lock().expect("Lock should not be poisoned").clear_failed(); } pub fn add_failed_directory(&self, directory: Url) { @@ -47,20 +49,11 @@ impl MailroomManager { } pub fn choose_relay(&self) -> Result { - use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom; - 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| !failed_relays.contains(r)).cloned().collect(); - - if remaining_relays.is_empty() { - return Err(anyhow!("No valid relays available")); - } - - remaining_relays - .choose(&mut payjoin::bitcoin::key::rand::thread_rng()) - .cloned() - .ok_or_else(|| anyhow!("Failed to select from remaining relays")) + self.relay_selector + .lock() + .expect("Lock should not be poisoned") + .select(&mut payjoin::bitcoin::key::rand::thread_rng()) + .ok_or_else(|| anyhow!("No valid relays available")) } pub fn choose_directory(&self) -> Result { diff --git a/payjoin/CHANGELOG.md b/payjoin/CHANGELOG.md index 46018da10..81d3b71c9 100644 --- a/payjoin/CHANGELOG.md +++ b/payjoin/CHANGELOG.md @@ -44,6 +44,10 @@ Selected Improvements: - Document BIP77 v1 fallback behavior in `create_post_request` (#1593) - Remove redundant language from `finalize_proposal` rustdocs (#1567) +### Relay Selection + +- Add `payjoin::relay::RelaySelector`, a reusable OHTTP relay-selection primitive (uniform-random over non-failed relays; callers mark failures to fail over), so integrators share one selection policy instead of each diverging + ## 0.25.0 Introduce monitoring typestates, replyable error handling, async diff --git a/payjoin/src/core/mod.rs b/payjoin/src/core/mod.rs index ec64e9963..efccf74d5 100644 --- a/payjoin/src/core/mod.rs +++ b/payjoin/src/core/mod.rs @@ -19,6 +19,8 @@ pub use into_url::{Error as IntoUrlError, IntoUrl}; pub(crate) mod url; pub use url::{ParseError as UrlParseError, Url}; #[cfg(feature = "v2")] +pub mod relay; +#[cfg(feature = "v2")] pub mod time; pub mod uri; pub use uri::{PjParam, PjParseError, PjUri, Uri, UriExt}; diff --git a/payjoin/src/core/relay.rs b/payjoin/src/core/relay.rs new file mode 100644 index 000000000..94ca1aede --- /dev/null +++ b/payjoin/src/core/relay.rs @@ -0,0 +1,126 @@ +use bitcoin::secp256k1::rand::seq::IteratorRandom; +use bitcoin::secp256k1::rand::{self}; + +use crate::Url; + +/// Picks an OHTTP relay, excluding relays marked failed, so clients share one +/// selection policy instead of each diverging. +#[derive(Clone, Debug)] +pub struct RelaySelector { + relays: Vec, + failed: Vec, +} + +impl RelaySelector { + /// Deduplicates `relays` (preserving order) so uniform selection isn't + /// skewed by a relay listed more than once. + pub fn new(relays: Vec) -> Self { + let mut deduped: Vec = Vec::new(); + for relay in relays { + if !deduped.contains(&relay) { + deduped.push(relay); + } + } + Self { relays: deduped, failed: Vec::new() } + } + + /// Pick a relay, never one marked failed, or `None` when none remain. + pub fn select(&self, rng: &mut R) -> Option { + self.relays.iter().filter(|r| !self.failed.contains(r)).choose(rng).cloned() + } + + /// Record a relay transport failure so `select` avoids it. + pub fn mark_failed(&mut self, relay: &Url) { + if !self.failed.contains(relay) { + self.failed.push(relay.clone()); + } + } + + /// Clear all recorded failures so every configured relay is selectable again. + pub fn clear_failed(&mut self) { self.failed.clear(); } +} + +#[cfg(test)] +mod tests { + use bitcoin::secp256k1::rand::rngs::StdRng; + use bitcoin::secp256k1::rand::SeedableRng; + + use super::*; + + fn relays() -> Vec { + ["https://a.example", "https://b.example", "https://c.example"] + .iter() + .map(|s| Url::parse(s).unwrap()) + .collect() + } + + #[test] + fn select_returns_a_configured_relay() { + let selector = RelaySelector::new(relays()); + let mut rng = StdRng::seed_from_u64(1); + let picked = selector.select(&mut rng).expect("a relay"); + assert!(relays().contains(&picked)); + } + + #[test] + fn select_never_returns_a_failed_relay() { + let mut selector = RelaySelector::new(relays()); + let mut rng = StdRng::seed_from_u64(2); + let failed = Url::parse("https://a.example").unwrap(); + selector.mark_failed(&failed); + for _ in 0..50 { + assert_ne!(selector.select(&mut rng), Some(failed.clone())); + } + } + + #[test] + fn select_is_none_when_all_failed() { + let mut selector = RelaySelector::new(relays()); + for r in relays() { + selector.mark_failed(&r); + } + let mut rng = StdRng::seed_from_u64(3); + assert_eq!(selector.select(&mut rng), None); + } + + #[test] + fn clear_failed_restores_all_relays() { + let mut selector = RelaySelector::new(relays()); + for r in relays() { + selector.mark_failed(&r); + } + let mut rng = StdRng::seed_from_u64(6); + assert_eq!(selector.select(&mut rng), None); + selector.clear_failed(); + assert!(selector.select(&mut rng).is_some()); + } + + #[test] + fn new_dedups_relays_preserving_order() { + let a = Url::parse("https://a.example").unwrap(); + let b = Url::parse("https://b.example").unwrap(); + let selector = RelaySelector::new(vec![a.clone(), a.clone(), b.clone()]); + assert_eq!(selector.relays, vec![a, b]); + } + + #[test] + fn select_is_none_when_empty() { + let selector = RelaySelector::new(Vec::new()); + let mut rng = StdRng::seed_from_u64(4); + assert_eq!(selector.select(&mut rng), None); + } + + // Selection is uniform across all relays. + #[test] + fn select_is_uniform_across_relays() { + let selector = RelaySelector::new(relays()); + let mut seen = std::collections::BTreeSet::new(); + let mut rng = StdRng::seed_from_u64(5); + for _ in 0..200 { + if let Some(r) = selector.select(&mut rng) { + seen.insert(r.to_string()); + } + } + assert_eq!(seen.len(), relays().len(), "random must reach every relay"); + } +}