Skip to content

Add choose_relay method in RelayManager #1628

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
benalleng merged 3 commits into from
Jun 9, 2026
Merged

Add choose_relay method in RelayManager #1628

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
77b0f32
Add one relay warning during randomized selection
benalleng Apr 13, 2026
c5f9fb1
Use choose relay instead of holding any relay manager state
benalleng Jun 8, 2026
be38713
Check each state update to add failed relay response
benalleng Jun 8, 2026
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
9 changes: 8 additions & 1 deletion payjoin-cli/src/app/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,7 +179,14 @@ impl Config {
#[cfg(feature = "v2")]
{
match built_config.get::<V2Config>("v2") {
Ok(v2) => config.version = Some(VersionConfig::V2(v2)),
Ok(v2) => {
if v2.ohttp_relays.len() < 2 {
tracing::warn!(
"Only one OHTTP relay is configured. Add more ohttp_relays to improve privacy."
);
}
config.version = Some(VersionConfig::V2(v2))
}
Err(e) =>
return Err(ConfigError::Message(format!(
"Valid V2 configuration is required for BIP77 mode: {e}"
Expand Down
90 changes: 46 additions & 44 deletions payjoin-cli/src/app/v2/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
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()));
let relay_manager = Arc::new(Mutex::new(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,10 +278,9 @@ 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, None, self.relay_manager.clone())
.await?
.ohttp_keys;
let ohttp_keys = unwrap_ohttp_keys_or_else_fetch(&self.config, self.relay_manager.clone())
.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 @@ -695,9 +694,8 @@ impl App {
sender: Sender<WithReplyKey>,
persister: &SenderPersister,
) -> Result<()> {
let relay = self.unwrap_relay_or_else_fetch(Some(&sender.endpoint())).await?;
let (req, ctx) = sender.create_v2_post_request(relay.as_str())?;
let response = self.post_request(req).await?;
let (response, ctx) =
self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await?;
let sender = sender.process_response(&response.bytes().await?, ctx).save(persister)?;
println!("Posted Original PSBT...");
self.get_proposed_payjoin_psbt(sender, persister).await
Expand All @@ -708,12 +706,11 @@ impl App {
sender: Sender<PollingForProposal>,
persister: &SenderPersister,
) -> Result<()> {
let ohttp_relay = self.unwrap_relay_or_else_fetch(Some(&sender.endpoint())).await?;
let mut session = sender.clone();
// Long poll until we get a response
loop {
let (req, ctx) = session.create_poll_request(ohttp_relay.as_str())?;
let response = self.post_request(req).await?;
let (response, ctx) =
self.post_via_relay(|relay| session.create_poll_request(relay)).await?;
let res = session.process_response(&response.bytes().await?, ctx).save(persister);
match res {
Ok(OptionalTransitionOutcome::Progress(psbt)) => {
Expand All @@ -740,14 +737,11 @@ impl App {
session: Receiver<Initialized>,
persister: &ReceiverPersister,
) -> Result<Receiver<UncheckedOriginalPayload>> {
let ohttp_relay =
self.unwrap_relay_or_else_fetch(Some(&session.pj_uri().extras.endpoint())).await?;

let mut session = session;
loop {
let (req, context) = session.create_poll_request(ohttp_relay.as_str())?;
println!("Polling receive request...");
let ohttp_response = self.post_request(req).await?;
let (ohttp_response, context) =
self.post_via_relay(|relay| session.create_poll_request(relay)).await?;
let state_transition = session
.process_response(ohttp_response.bytes().await?.to_vec().as_slice(), context)
.save(persister);
Expand Down Expand Up @@ -949,10 +943,13 @@ impl App {
proposal: Receiver<PayjoinProposal>,
persister: &ReceiverPersister,
) -> Result<()> {
let (req, ohttp_ctx) = proposal
.create_post_request(self.unwrap_relay_or_else_fetch(None::<&str>).await?.as_str())
.map_err(|e| anyhow!("v2 req extraction failed {}", e))?;
let res = self.post_request(req).await?;
let (res, ohttp_ctx) = self
.post_via_relay(|relay| {
proposal
.create_post_request(relay)
.map_err(|e| anyhow!("v2 req extraction failed {}", e))
})
.await?;
let payjoin_psbt = proposal.psbt().clone();
let session = proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister)?;
println!(
Expand Down Expand Up @@ -1008,36 +1005,19 @@ impl App {
}
}

async fn unwrap_relay_or_else_fetch(
&self,
directory: Option<impl payjoin::IntoUrl>,
) -> Result<payjoin::Url> {
let directory = directory.map(|url| url.into_url()).transpose()?;
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 error by attempting to send an error response over the directory
async fn handle_error(
&self,
session: Receiver<HasReplyableError>,
persister: &ReceiverPersister,
) -> Result<()> {
let (err_req, err_ctx) = session
.create_error_request(self.unwrap_relay_or_else_fetch(None::<&str>).await?.as_str())?;

let err_response = match self.post_request(err_req).await {
Ok(response) => response,
Err(e) => return Err(anyhow!("Failed to post error request: {}", e)),
};
let (err_response, err_ctx) = self
.post_via_relay(|relay| {
session
.create_error_request(relay)
.map_err(|e| anyhow!("Failed to post error request: {}", e))
})
.await?;

let err_bytes = match err_response.bytes().await {
Ok(bytes) => bytes,
Expand Down Expand Up @@ -1072,4 +1052,26 @@ impl App {
.and_then(|r| r.error_for_status())
.context("HTTP request failed")
}

async fn post_via_relay<F, T, E>(&self, mut build: F) -> Result<(reqwest::Response, T)>
where
F: FnMut(&str) -> std::result::Result<(payjoin::Request, T), E>,
E: Into<anyhow::Error>,
{
loop {
let relay =
self.relay_manager.lock().expect("Lock should not be poisoned").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);
}
}
}
}
}
70 changes: 25 additions & 45 deletions payjoin-cli/src/app/v2/ohttp.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
//! OHTTP relay selection and key bootstrapping for the payjoin-cli.
//!
//! [`RelayManager`] tracks the currently selected relay and any relays that
//! have failed, excluding them from future selections for the lifetime of
//! the [`RelayManager`].
//! [`RelayManager`] tracks relays that have failed, excluding them from
//! future selections for the lifetime of the [`RelayManager`].
//!
//! `fetch_ohttp_keys` selects a relay at random from the configured list,
//! `unwrap_ohttp_keys_or_else_fetch` returns user-supplied keys when present,
//! otherwise selects a relay at random from the configured list,

@xstoicunicornx xstoicunicornx Jun 9, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
//! otherwise selects a relay at random from the configured list,
//! otherwise fetches the keys from the config directory using a
//! relay selected at random from the configured list,

Nit: this comment doesn't read clearly as is

@benalleng benalleng Jun 9, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Going to put this in the followup with @va-an in moving this under the RelayManager

//! excluding relays that [`RelayManager`] has marked as failed,
//! to avoid a fixed contact pattern at the network layer.
use std::sync::{Arc, Mutex};
Expand All @@ -16,74 +16,55 @@ use super::Config;

#[derive(Debug, Clone)]
pub struct RelayManager {
selected_relay: Option<Url>,
config: Config,
failed_relays: Vec<Url>,
}

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

pub fn set_selected_relay(&mut self, relay: Url) { self.selected_relay = Some(relay); }
pub fn add_failed_relay(&mut self, relay: Url) { self.failed_relays.push(relay); }

pub fn get_selected_relay(&self) -> Option<Url> { self.selected_relay.clone() }
pub fn choose_relay(&self) -> Result<Url> {
use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom;
let relays = self.config.v2()?.ohttp_relays.clone();
let remaining_relays: Vec<_> =
relays.iter().filter(|r| !self.failed_relays.contains(r)).cloned().collect();

pub fn add_failed_relay(&mut self, relay: Url) { self.failed_relays.push(relay); }
if remaining_relays.is_empty() {
return Err(anyhow!("No valid relays available"));
}

pub fn get_failed_relays(&self) -> Vec<Url> { self.failed_relays.clone() }
remaining_relays
.choose(&mut payjoin::bitcoin::key::rand::thread_rng())
.cloned()
.ok_or_else(|| anyhow!("Failed to select from remaining relays"))
}
}

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

pub(crate) async fn unwrap_ohttp_keys_or_else_fetch(
config: &Config,
directory: Option<Url>,
relay_manager: Arc<Mutex<RelayManager>>,
) -> Result<ValidatedOhttpKeys> {
if let Some(ohttp_keys) = config.v2()?.ohttp_keys.clone() {
println!("Using OHTTP Keys from config");
let validated = fetch_ohttp_keys(config, directory, relay_manager).await?;
Ok(ValidatedOhttpKeys { ohttp_keys, relay_url: validated.relay_url })
} else {
println!("Bootstrapping private network transport over Oblivious HTTP");
let fetched_keys = fetch_ohttp_keys(config, directory, relay_manager).await?;

Ok(fetched_keys)
return Ok(ValidatedOhttpKeys { ohttp_keys });
}
fetch_ohttp_keys(config, relay_manager).await
}

async fn fetch_ohttp_keys(
config: &Config,
directory: Option<Url>,
relay_manager: Arc<Mutex<RelayManager>>,
) -> Result<ValidatedOhttpKeys> {
use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom;
let payjoin_directory = directory.unwrap_or(config.v2()?.pj_directory.clone());
let relays = config.v2()?.ohttp_relays.clone();
let payjoin_directory = config.v2()?.pj_directory.clone();

loop {
let failed_relays =
relay_manager.lock().expect("Lock should not be poisoned").get_failed_relays();

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

let selected_relay =
match remaining_relays.choose(&mut payjoin::bitcoin::key::rand::thread_rng()) {
Some(relay) => relay.clone(),
None => return Err(anyhow!("Failed to select from remaining relays")),
};

relay_manager
.lock()
.expect("Lock should not be poisoned")
.set_selected_relay(selected_relay.clone());
relay_manager.lock().expect("Lock should not be poisoned").choose_relay()?;

let ohttp_keys = {
#[cfg(feature = "_manual-tls")]
Expand All @@ -109,8 +90,7 @@ async fn fetch_ohttp_keys(
};

match ohttp_keys {
Ok(keys) =>
return Ok(ValidatedOhttpKeys { ohttp_keys: keys, relay_url: selected_relay }),
Ok(keys) => return Ok(ValidatedOhttpKeys { ohttp_keys: keys }),
Err(payjoin::io::Error::UnexpectedStatusCode(e)) => {
return Err(payjoin::io::Error::UnexpectedStatusCode(e).into());
}
Expand Down
Loading