Add core RelaySelector

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

@@ -130,7 +130,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 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?;

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

@@ -14,53 +14,46 @@
 use std::sync::{Arc, Mutex};
 
 use anyhow::{anyhow, Result};
+use payjoin::relay::RelaySelector;
 use payjoin::Url;
 
 use super::Config;
 
 #[derive(Debug, Clone)]
 pub struct MailroomManager {
     config: Config,
-    failed_relays: Arc<Mutex<Vec<Url>>>,
+    relay_selector: Arc<Mutex<RelaySelector>>,
     failed_directories: Arc<Mutex<Vec<Url>>>,
 }
 
 impl MailroomManager {
-    pub fn new(config: Config) -> Self {
-        MailroomManager {
+    pub fn new(config: Config) -> Result<Self> {
+        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) {
         self.failed_directories.lock().expect("Lock should not be poisoned").push(directory);
     }
 
     pub fn choose_relay(&self) -> Result<Url> {
-        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<Url> {

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

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};

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<Url>,
+    failed: Vec<Url>,
+}
+
+impl RelaySelector {
+    /// Deduplicates `relays` (preserving order) so uniform selection isn't
+    /// skewed by a relay listed more than once.
+    pub fn new(relays: Vec<Url>) -> Self {
+        let mut deduped: Vec<Url> = 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<R: rand::Rng>(&self, rng: &mut R) -> Option<Url> {
+        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<Url> {
+        ["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");
+    }
+}