diff --git a/payjoin-cli/CHANGELOG.md b/payjoin-cli/CHANGELOG.md index be97ca13f..991ef4955 100644 --- a/payjoin-cli/CHANGELOG.md +++ b/payjoin-cli/CHANGELOG.md @@ -5,6 +5,9 @@ - Fix panic when spending taproot UTXOs. They are now contributed as key-path spends with an assumed default input weight. - Any unsupported coin is simply skipped instead of crashing the cli. +- Replace the CLI's relay/directory bookkeeping and failover loops with + `payjoin::RequestHandler`, shared behind an `Arc` so concurrent sessions + no longer serialize their relay posts ## 1.0.0 diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index 924129b13..f6bce8f89 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -13,6 +13,7 @@ use payjoin::receive::v2::{ ReceiverBuilder, SessionOutcome as ReceiverSessionOutcome, UncheckedOriginalPayload, WantsFeeRange, WantsInputs, WantsOutputs, }; +use payjoin::request_handler::{PostError, RelayPost, RequestExpiry}; use payjoin::send::v2::{ replay_event_log as replay_sender_event_log, PendingFallback as SenderPendingFallback, PollingForProposal, SendSession, Sender, SenderBuilder, SessionOutcome as SenderSessionOutcome, @@ -40,30 +41,6 @@ const W_STATUS: usize = 15; /// misbehaving directory or relay is not hammered in a tight loop. const TRANSIENT_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(5); -/// A request-construction error that can report whether it was caused by -/// session expiry. Implemented by the sender/receiver request-building error -/// types so `post_via_relay` can hand expiry back to the caller (which owns the -/// typestate needed to react) instead of flattening it into `anyhow::Error`. -trait RequestExpiry { - fn expired(&self) -> bool; -} - -impl RequestExpiry for payjoin::send::v2::CreateRequestError { - fn expired(&self) -> bool { self.is_expired() } -} - -impl RequestExpiry for payjoin::receive::v2::CreateRequestError { - fn expired(&self) -> bool { self.is_expired() } -} - -/// Outcome of building and posting a request via `post_via_relay`. HTTP -/// failures are retried against other relays inside the helper; only session -/// expiry and fatal build errors escape to the caller. -enum RelayPost { - Posted(reqwest::Response, T), - Expired, -} - #[derive(Clone)] pub(crate) struct App { config: Config, @@ -197,7 +174,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?; @@ -327,22 +304,7 @@ impl AppTrait for App { async fn receive_payjoin(&self, amount: Amount) -> Result<()> { let address = self.wallet().get_new_address()?; let persister = ReceiverPersister::new(self.db.clone())?; - let (directory, ohttp_keys) = loop { - let directory = self.mailroom_manager.choose_directory()?; - match self - .mailroom_manager - .unwrap_ohttp_keys_or_else_fetch_from_directory(&directory) - .await - { - Ok(keys) => break (directory, keys.ohttp_keys), - Err(e) => { - tracing::debug!("Directory {directory} failed: {e:#}"); - self.mailroom_manager.add_failed_directory(directory); - self.mailroom_manager.clear_failed_relays(); - continue; - } - } - }; + let (directory, ohttp_keys) = self.mailroom_manager.fetch_ohttp_keys().await?; let mut receiver_builder = ReceiverBuilder::new(address, directory.as_str(), ohttp_keys)?.with_amount(amount); if let Some(max_fee_rate) = self.config.max_fee_rate { @@ -871,14 +833,18 @@ impl App { sender: Sender, persister: &SenderPersister, ) -> Result { - let (response, ctx) = - match self.post_via_relay(|relay| sender.create_v2_post_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => { - self.cancel_sender_session(persister.session_id(), true)?; - return Ok(SendSession::Closed(SenderSessionOutcome::Aborted)); - } - }; + let (response, ctx) = match self + .post_via_relay(sender.directory(), |relay| { + sender.create_v2_post_request(relay.as_str()) + }) + .await? + { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_sender_session(persister.session_id(), true)?; + return Ok(SendSession::Closed(SenderSessionOutcome::Aborted)); + } + }; match sender.process_response(&response.bytes().await?, ctx).save(persister) { Ok(sender) => { persister.print("Posted Original PSBT..."); @@ -899,14 +865,16 @@ impl App { sender: Sender, persister: &SenderPersister, ) -> Result { - let (response, ctx) = - match self.post_via_relay(|relay| sender.create_poll_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => { - self.cancel_sender_session(persister.session_id(), true)?; - return Ok(SendSession::Closed(SenderSessionOutcome::Aborted)); - } - }; + let (response, ctx) = match self + .post_via_relay(sender.directory(), |relay| sender.create_poll_request(relay.as_str())) + .await? + { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_sender_session(persister.session_id(), true)?; + return Ok(SendSession::Closed(SenderSessionOutcome::Aborted)); + } + }; let res = sender.clone().process_response(&response.bytes().await?, ctx).save(persister); match res { Ok(OptionalTransitionOutcome::Progress(psbt)) => { @@ -987,14 +955,18 @@ impl App { persister: &ReceiverPersister, ) -> Result { persister.print("Polling receive request..."); - let (ohttp_response, context) = - match self.post_via_relay(|relay| session.create_poll_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => { - self.cancel_receiver_session(persister.session_id(), true)?; - return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); - } - }; + let (ohttp_response, context) = match self + .post_via_relay(session.directory(), |relay| { + session.create_poll_request(relay.as_str()) + }) + .await? + { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_receiver_session(persister.session_id(), true)?; + return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); + } + }; let state_transition = session .process_response(ohttp_response.bytes().await?.to_vec().as_slice(), context) .save(persister); @@ -1138,14 +1110,18 @@ impl App { proposal: Receiver, persister: &ReceiverPersister, ) -> Result { - let (res, ohttp_ctx) = - match self.post_via_relay(|relay| proposal.create_post_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => { - self.cancel_receiver_session(persister.session_id(), true)?; - return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); - } - }; + let (res, ohttp_ctx) = match self + .post_via_relay(proposal.directory(), |relay| { + proposal.create_post_request(relay.as_str()) + }) + .await? + { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_receiver_session(persister.session_id(), true)?; + return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); + } + }; let payjoin_psbt = proposal.psbt().clone(); match proposal.process_response(&res.bytes().await?, ohttp_ctx).save(persister) { Ok(session) => { @@ -1228,14 +1204,18 @@ impl App { session: Receiver, persister: &ReceiverPersister, ) -> Result { - let (err_response, err_ctx) = - match self.post_via_relay(|relay| session.create_error_request(relay)).await? { - RelayPost::Posted(resp, ctx) => (resp, ctx), - RelayPost::Expired => { - self.cancel_receiver_session(persister.session_id(), true)?; - return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); - } - }; + let (err_response, err_ctx) = match self + .post_via_relay(session.directory(), |relay| { + session.create_error_request(relay.as_str()) + }) + .await? + { + RelayPost::Posted(resp, ctx) => (resp, ctx), + RelayPost::Expired => { + self.cancel_receiver_session(persister.session_id(), true)?; + return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); + } + }; let err_bytes = match err_response.bytes().await { Ok(bytes) => bytes, Err(e) => return Err(anyhow!("Failed to get error response bytes: {}", e)), @@ -1285,25 +1265,21 @@ impl App { .context("HTTP request failed") } - async fn post_via_relay(&self, mut build: F) -> Result> + async fn post_via_relay( + &self, + directory: &payjoin::Url, + build: F, + ) -> Result> where - F: FnMut(&str) -> std::result::Result<(payjoin::Request, T), E>, + F: FnMut(&payjoin::Url) -> std::result::Result<(payjoin::Request, T), E>, E: RequestExpiry + Into, { - loop { - let relay = self.mailroom_manager.choose_relay()?; - let (req, ctx) = match build(relay.as_str()) { - Ok(r) => r, - Err(e) if e.expired() => return Ok(RelayPost::Expired), - Err(e) => return Err(e.into()), - }; - match self.post_request(req).await { - Ok(resp) => return Ok(RelayPost::Posted(resp, ctx)), - Err(e) => { - tracing::debug!("Request to relay {relay} failed: {e:?}"); - self.mailroom_manager.add_failed_relay(relay); - } - } + let post = |req: payjoin::Request| async move { self.post_request(req).await }; + match self.mailroom_manager.post_via_relay(directory, build, post).await { + Ok(RelayPost::Posted(resp, ctx)) => Ok(RelayPost::Posted(resp, ctx)), + Ok(RelayPost::Expired) => Ok(RelayPost::Expired), + Err(PostError::NoRelaysAvailable { .. }) => Err(anyhow!("No valid relays available")), + Err(PostError::Build(e)) => Err(e.into()), } } } diff --git a/payjoin-cli/src/app/v2/ohttp.rs b/payjoin-cli/src/app/v2/ohttp.rs index 82fa4e29a..e6ec9ccbe 100644 --- a/payjoin-cli/src/app/v2/ohttp.rs +++ b/payjoin-cli/src/app/v2/ohttp.rs @@ -1,147 +1,66 @@ //! OHTTP relay and payjoin directory selection / key bootstrapping for the payjoin-cli. //! -//! [`MailroomManager`] tracks relays and directories that have failed, -//! excluding them from future selections for the lifetime of the [`MailroomManager`]. +//! Thin wrapper around [`payjoin::request_handler::RequestHandler`] sharing one +//! instance across sessions and tasks. The `RequestHandler` tracks relays and +//! directories that have failed, excluding them from future selection, and +//! fetches OHTTP keys with automatic failover. It takes `&self` and holds its +//! own lock only for synchronous accesses, so it is shared directly (behind an +//! `Arc`) without an external async lock — concurrent sessions never serialize +//! their relay posts. //! -//! `unwrap_ohttp_keys_or_else_fetch_from_directory` returns user-supplied keys -//! when present, otherwise selects a relay at random from the configured list -//! (excluding failed relays) to fetch OHTTP keys from the given directory. -//! -//! `fetch_ohttp_keys_from_directory` retries on relay failures (e.g. connection -//! errors) by selecting another relay. Once a directory is chosen for a session -//! it must not change — the directory is embedded in the BIP21 URI at session -//! creation and recovered from the session event log on resume. -use std::sync::{Arc, Mutex}; - -use anyhow::{anyhow, Result}; +//! Once a directory is chosen for a session it must not change — the +//! directory is embedded in the BIP21 URI at session creation and recovered +//! from the session event log on resume. +use std::sync::Arc; + +use anyhow::Result; +use payjoin::bitcoin::secp256k1::rand::rngs::OsRng; +use payjoin::request_handler::RequestHandler; use payjoin::Url; use super::Config; #[derive(Debug, Clone)] -pub struct MailroomManager { - config: Config, - failed_relays: Arc>>, - failed_directories: Arc>>, -} +pub struct MailroomManager(Arc); impl MailroomManager { - pub fn new(config: Config) -> Self { - MailroomManager { - config, - failed_relays: Arc::new(Mutex::new(Vec::new())), - failed_directories: Arc::new(Mutex::new(Vec::new())), + pub fn new(config: Config) -> Result { + let v2 = config.v2()?; + let mailroom = RequestHandler::new(v2.ohttp_relays.clone(), v2.pj_directories.clone()); + if let Some(ohttp_keys) = v2.ohttp_keys.clone() { + mailroom.set_ohttp_keys(ohttp_keys); } - } - - pub fn add_failed_relay(&self, relay: Url) { - self.failed_relays.lock().expect("Lock should not be poisoned").push(relay); - } - - pub fn clear_failed_relays(&self) { - self.failed_relays.lock().expect("Lock should not be poisoned").clear(); - } - - 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 { - 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")); + #[cfg(feature = "_manual-tls")] + if let Some(cert_path) = config.root_certificate.as_ref() { + mailroom.set_cert_der(std::fs::read(cert_path)?); } - - remaining_relays - .choose(&mut payjoin::bitcoin::key::rand::thread_rng()) - .cloned() - .ok_or_else(|| anyhow!("Failed to select from remaining relays")) + Ok(Self(Arc::new(mailroom))) } - pub fn choose_directory(&self) -> Result { - use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom; - let directories = &self.config.v2()?.pj_directories; - let failed_directories = - self.failed_directories.lock().expect("Lock should not be poisoned"); - let remaining_directories: Vec<_> = - directories.iter().filter(|d| !failed_directories.contains(d)).cloned().collect(); - - if remaining_directories.is_empty() { - return Err(anyhow!("No valid directories available")); - } - - remaining_directories - .choose(&mut payjoin::bitcoin::key::rand::thread_rng()) - .cloned() - .ok_or_else(|| anyhow!("Failed to select from remaining directories")) + pub async fn fetch_ohttp_keys(&self) -> Result<(Url, payjoin::OhttpKeys)> { + // OsRng rather than thread_rng: ThreadRng is !Send and must not be + // held across the fetch's await points. + self.0.fetch_ohttp_keys(&mut OsRng).await.map_err(Into::into) } - pub(crate) async fn unwrap_ohttp_keys_or_else_fetch_from_directory( + pub async fn post_via_relay( &self, directory: &Url, - ) -> Result { - if let Some(ohttp_keys) = self.config.v2()?.ohttp_keys.clone() { - return Ok(ValidatedOhttpKeys { ohttp_keys }); - } - self.fetch_ohttp_keys_from_directory(directory).await - } - - async fn fetch_ohttp_keys_from_directory(&self, directory: &Url) -> Result { - loop { - let selected_relay = self.choose_relay()?; - - 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(), - directory.as_str(), - &cert_der, - ) - .await - } else { - payjoin::io::fetch_ohttp_keys(selected_relay.as_str(), directory.as_str()) - .await - } - } - #[cfg(not(feature = "_manual-tls"))] - payjoin::io::fetch_ohttp_keys(selected_relay.as_str(), directory.as_str()).await - }; - - match ohttp_keys { - Ok(keys) => return Ok(ValidatedOhttpKeys { ohttp_keys: keys }), - Err(payjoin::io::Error::UnexpectedStatusCode(e)) => { - tracing::debug!( - "Directory {directory} returned unexpected status via relay {selected_relay}: {e:?}" - ); - self.add_failed_directory(directory.clone()); - return Err(anyhow!("Directory {directory} returned unexpected status: {e}")); - } - Err(payjoin::io::Error::OhttpKeysBodyTooLarge(len)) => { - tracing::debug!( - "Directory {directory} returned an oversized OHTTP keys body via relay {selected_relay}: {len} bytes" - ); - self.add_failed_directory(directory.clone()); - return Err(anyhow!( - "Directory {directory} returned an oversized OHTTP keys body: {len} bytes" - )); - } - Err(e) => { - tracing::debug!("Failed to connect to relay: {selected_relay}, {e:?}"); - self.add_failed_relay(selected_relay); - } - } - } + build: F, + post: impl Fn(payjoin::Request) -> Fut, + ) -> std::result::Result< + payjoin::request_handler::RelayPost, + payjoin::request_handler::PostError, + > + where + F: FnMut(&payjoin::Url) -> std::result::Result<(payjoin::Request, T), E>, + E: payjoin::request_handler::RequestExpiry, + TransportError: std::fmt::Display, + Fut: std::future::Future>, + { + // The directory is never reached through itself: a server serving as + // both relay and directory rejects the self-loop request. + let transport = payjoin::request_handler::FnRelayTransport::new(build, post); + self.0.post_via_relay(&mut OsRng, directory, transport).await } } - -pub(crate) struct ValidatedOhttpKeys { - pub(crate) ohttp_keys: payjoin::OhttpKeys, -} diff --git a/payjoin-ffi/CHANGELOG.md b/payjoin-ffi/CHANGELOG.md index 78ea889c6..6d2eb2624 100644 --- a/payjoin-ffi/CHANGELOG.md +++ b/payjoin-ffi/CHANGELOG.md @@ -1,3 +1,23 @@ +## Unreleased + +### RequestHandler + +- Export `RequestHandler` for relay and directory selection with + failure tracking, driven by the shared core failover loops +- Add `RequestHandler::post_via_relay`, posting over a + binding-supplied `RelayTransport` (`build`/`post`) with relay + failover +- Add `RequestHandler::fetch_ohttp_keys`, fetching a directory's OHTTP + keys over a binding-supplied `OhttpKeysFetcher` with relay and + directory failover +- `RequestHandlerError::NoDirectoriesAvailable` and + `RelayPostError::NoRelaysAvailable` carry a `last_error` describing + the most recent failed attempt +- `RequestHandler::set_ohttp_keys` and `set_cert_der` pin configuration + on the handler itself, so every reference observes it +- Failed endpoints are cleared on entry to each failover loop, so a + transient outage no longer persists to the next call + ## [0.25.0] - Publish the Python, JavaScript, C#, and Dart packages from one signed diff --git a/payjoin-ffi/src/lib.rs b/payjoin-ffi/src/lib.rs index 63c67ae04..eca37e5a9 100644 --- a/payjoin-ffi/src/lib.rs +++ b/payjoin-ffi/src/lib.rs @@ -5,6 +5,7 @@ pub mod ohttp; pub mod output_substitution; pub mod receive; pub mod request; +pub mod request_handler; pub mod send; #[cfg(feature = "_test-utils")] pub mod test_utils; @@ -17,6 +18,7 @@ pub use crate::ohttp::*; pub use crate::output_substitution::*; pub use crate::receive::*; pub use crate::request::Request; +pub use crate::request_handler::*; pub use crate::send::*; #[cfg(feature = "_test-utils")] pub use crate::test_utils::*; diff --git a/payjoin-ffi/src/ohttp.rs b/payjoin-ffi/src/ohttp.rs index c3d8edb79..071e54e10 100644 --- a/payjoin-ffi/src/ohttp.rs +++ b/payjoin-ffi/src/ohttp.rs @@ -40,3 +40,12 @@ impl From<&ClientResponse> for payjoin::OhttpResponse { impl From for ClientResponse { fn from(value: payjoin::OhttpResponse) -> Self { Self(Mutex::new(Some(value))) } } + +#[cfg(all(test, feature = "_test-utils"))] +impl ClientResponse { + /// A placeholder context for tests that never process a response. + /// + /// Consuming it panics like any other context that has already been moved + /// out of memory, so it must only be used where the response is discarded. + pub(crate) fn dangling() -> Self { Self(Mutex::new(None)) } +} diff --git a/payjoin-ffi/src/request_handler.rs b/payjoin-ffi/src/request_handler.rs new file mode 100644 index 000000000..0a578aecf --- /dev/null +++ b/payjoin-ffi/src/request_handler.rs @@ -0,0 +1,638 @@ +use std::sync::{Arc, Mutex}; + +use payjoin::bitcoin::secp256k1::rand::rngs::OsRng; +use payjoin::request_handler; + +use crate::ohttp::{ClientResponse, OhttpKeys}; +pub use crate::uri::error::UrlParseError; +use crate::Url; + +/// Errors returned by [`RequestHandler`] key fetching. +#[derive(Debug, thiserror::Error, uniffi::Error)] +#[non_exhaustive] +pub enum RequestHandlerError { + /// Every configured directory has been marked failed. Carries the last + /// fetch failure observed while trying them, when there was one. + #[error("No valid directories available")] + NoDirectoriesAvailable { + /// The most recent failure seen while exhausting the directories. + last_error: Option, + }, + /// Catch-all for unhandled error variants + #[error("An unexpected error occurred")] + Unexpected, +} + +impl From for RequestHandlerError { + fn from(value: request_handler::Error) -> Self { + match value { + request_handler::Error::NoDirectoriesAvailable => + Self::NoDirectoriesAvailable { last_error: None }, + _ => Self::Unexpected, + } + } +} + +/// The result of a successful [`RequestHandler::fetch_ohttp_keys`]. +#[derive(Debug, uniffi::Record)] +pub struct FetchedOhttpKeys { + /// The directory the keys belong to. Pin this in the BIP21 URI when + /// creating a session, since a session's directory must not change. + pub directory: Arc, + /// The directory's OHTTP keys. + pub ohttp_keys: Arc, +} + +/// Why a single attempt to fetch a directory's OHTTP keys failed. +/// +/// The variant an [`OhttpKeysFetcher`] returns tells [`RequestHandler`] how to +/// fail over: a `Transient` failure retires the relay and retries the same +/// directory over another relay, while a `Directory` failure retires the +/// directory and moves on to another directory. +#[derive(Debug, thiserror::Error, uniffi::Error)] +#[non_exhaustive] +pub enum DirectoryFetchError { + /// The relay (or the connection through it) is at fault; retry over + /// another relay. + #[error("relay transport failure: {detail}")] + Transient { + /// Human-readable detail. + detail: String, + }, + /// The directory itself is at fault (for example a non-2xx status); do not + /// retry it over another relay. + #[error("directory failure: {detail}")] + Directory { + /// Human-readable detail. + detail: String, + }, +} + +impl request_handler::DirectoryFatal for DirectoryFetchError { + fn is_directory_fatal(&self) -> bool { matches!(self, DirectoryFetchError::Directory { .. }) } +} + +/// Fetches a payjoin directory's OHTTP keys through an OHTTP relay. +/// +/// Implemented by the binding, which owns the HTTP transport and any proxy or +/// TLS configuration. [`RequestHandler::fetch_ohttp_keys`] selects the relay +/// and directory, drives failover, and tracks failures; the fetcher performs a +/// single HTTP attempt per call. +#[uniffi::export(with_foreign)] +#[async_trait::async_trait] +pub trait OhttpKeysFetcher: Send + Sync { + /// Fetch OHTTP keys from `directory` tunnelled through `relay`. + /// + /// Return [`DirectoryFetchError::Transient`] when the relay is at fault so + /// the handler retries over another relay, and + /// [`DirectoryFetchError::Directory`] when the directory itself is at + /// fault so the handler retries over another directory. + async fn fetch( + &self, + relay: Arc, + directory: Arc, + ) -> Result, DirectoryFetchError>; +} + +/// A posted request and the OHTTP context needed to process its response. +#[derive(uniffi::Record)] +pub struct RelayedResponse { + /// The response body returned through the relay. + pub body: Vec, + /// The OHTTP context to pass to the session's `process_response`. + pub ohttp_context: Arc, +} + +/// A request built for a specific relay, ready to POST. +#[derive(uniffi::Record)] +pub struct BuiltRequest { + /// The request to POST to the relay. + pub request: crate::Request, + /// The OHTTP context needed to process the eventual response. + pub ohttp_context: Arc, +} + +/// Why building a request for a relay failed. +/// +/// [`RequestHandler::post_via_relay`] stops failover on either variant and +/// surfaces it to the caller: `Expired` as [`RelayPostError::Expired`] and +/// `Fatal` as [`RelayPostError::Fatal`]. +#[derive(Debug, thiserror::Error, uniffi::Error)] +#[non_exhaustive] +pub enum RelayBuildError { + /// The session expired while building the request; stop failover. + #[error("session expired")] + Expired, + /// The request could not be built for another reason; stop failover. + #[error("could not build request: {detail}")] + Fatal { + /// Human-readable detail. + detail: String, + }, +} + +impl request_handler::RequestExpiry for RelayBuildError { + fn expired(&self) -> bool { matches!(self, RelayBuildError::Expired) } +} + +/// Why a single attempt to POST a request over a relay failed. +/// +/// The variant a [`RelayTransport`] returns tells [`RequestHandler`] how to +/// fail over: a `Transient` failure retires the relay and retries over another +/// relay. Build-time failures (expiry, fatal) are reported by +/// [`RelayTransport::build`], not here. +#[derive(Debug, thiserror::Error, uniffi::Error)] +#[non_exhaustive] +pub enum RelayTransportError { + /// The relay could not be reached; retire it and retry over another relay. + #[error("relay transport failure: {detail}")] + Transient { + /// Human-readable detail. + detail: String, + }, +} + +/// Errors from [`RequestHandler::post_via_relay`]. +#[derive(Debug, thiserror::Error, uniffi::Error)] +#[non_exhaustive] +pub enum RelayPostError { + /// Every configured relay has been marked failed or excluded. Carries the + /// last transport failure observed while trying them, when there was one. + #[error("No valid relays available")] + NoRelaysAvailable { + /// The most recent transport failure seen while exhausting the relays. + last_error: Option, + }, + /// The session expired while building the request. + #[error("session expired")] + Expired, + /// The request could not be built for another reason. + #[error("could not build request: {detail}")] + Fatal { + /// Human-readable detail. + detail: String, + }, +} + +/// Builds and posts a request through a relay, implemented by the binding. +/// +/// [`RequestHandler::post_via_relay`] selects the relay, drives failover, and +/// tracks failures; the transport builds a fresh request for the selected +/// relay and performs the HTTP request. +#[uniffi::export(with_foreign)] +#[async_trait::async_trait] +pub trait RelayTransport: Send + Sync { + /// Build a fresh OHTTP request addressed to `relay`. + /// + /// Build a new request (and OHTTP context) on every call: reusing or + /// retransmitting identical ciphertext breaks OHTTP privacy, since a relay + /// could observe that a request is being retried. Return + /// [`RelayBuildError::Expired`] if the session expired while building. + fn build(&self, relay: Arc) -> Result; + + /// POST a request produced by [`RelayTransport::build`], returning the + /// response body. Return [`RelayTransportError::Transient`] to retire the + /// relay and retry over another. + async fn post(&self, request: crate::Request) -> Result, RelayTransportError>; +} + +/// Selects OHTTP relays and payjoin directories, excluding endpoints marked +/// failed, and drives failover over a caller-supplied transport. +/// +/// Endpoints are configured as URL strings and deduplicated. Failures are +/// scoped to a single failover loop: `post_via_relay` and `fetch_ohttp_keys` +/// clear recorded relay failures on entry, and `fetch_ohttp_keys` clears +/// directory failures on entry, so a transient outage that retires every +/// endpoint does not persist to the next call. +/// +/// Failure state is synchronized internally, so the handler is safe to share +/// across tasks. +/// +/// A directory chosen for a session must not change: select one when creating +/// a new session and keep it for the session's lifetime. Resuming a session +/// does not consult this handler. +#[derive(Debug, uniffi::Object)] +pub struct RequestHandler { + inner: request_handler::RequestHandler, +} + +fn parse_urls(urls: Vec) -> Result, UrlParseError> { + urls.into_iter().map(|url| payjoin::Url::parse(&url).map_err(Into::into)).collect() +} + +#[uniffi::export] +impl RequestHandler { + /// Create a `RequestHandler` from relay and directory URL strings. + /// + /// Duplicates are removed so uniform selection isn't skewed by an + /// endpoint listed more than once. + #[uniffi::constructor] + pub fn new(relays: Vec, directories: Vec) -> Result { + Ok(Self { + inner: request_handler::RequestHandler::new( + parse_urls(relays)?, + parse_urls(directories)?, + ), + }) + } + + /// Pin user-supplied OHTTP keys so a key fetch never contacts a + /// directory. + /// + /// Observed by every reference to this handler rather than creating a + /// forked copy with separate failure tracking. + pub fn set_ohttp_keys(&self, ohttp_keys: Arc) { + let keys: payjoin::OhttpKeys = Arc::unwrap_or_clone(ohttp_keys).into(); + self.inner.set_ohttp_keys(keys); + } + + /// Return the pinned OHTTP keys, or fetch them with automatic failover. + /// + /// Relay transport failures retry over another relay. A directory failure + /// retires that directory and retries over another, giving the next + /// directory a clean relay set. Failures are scoped to this call: the + /// failed-directory and failed-relay sets are cleared on entry, so an + /// endpoint that failed a previous call is tried again. + /// + /// Returns the directory alongside the keys: a receiver pins the directory + /// in the BIP21 URI at session creation, so the directory is reported even + /// when the keys are pinned and no fetch happened. + pub async fn fetch_ohttp_keys( + &self, + fetcher: Arc, + ) -> Result { + // Core drives the directory/relay failover loop; the closure records + // the last attempt's failure so it can be reported once every + // directory is exhausted. + let last_error = Arc::new(Mutex::new(None::)); + let fetch = { + let fetcher = fetcher.clone(); + let last_error = last_error.clone(); + move |relay: payjoin::Url, directory: payjoin::Url| { + let fetcher = fetcher.clone(); + let last_error = last_error.clone(); + async move { + let result = + fetcher.fetch(Arc::new(relay.into()), Arc::new(directory.into())).await; + match result { + Ok(ohttp_keys) => Ok(Arc::unwrap_or_clone(ohttp_keys).into()), + Err(e) => { + *last_error.lock().unwrap() = Some(e.to_string()); + Err(e) + } + } + } + } + }; + match self.inner.fetch_ohttp_keys_with(&mut OsRng, fetch).await { + Ok((directory, ohttp_keys)) => Ok(FetchedOhttpKeys { + directory: Arc::new(directory.into()), + ohttp_keys: Arc::new(ohttp_keys.into()), + }), + Err(request_handler::Error::NoDirectoriesAvailable) => + Err(RequestHandlerError::NoDirectoriesAvailable { + last_error: last_error.lock().unwrap().clone(), + }), + Err(e) => Err(e.into()), + } + } + + /// Build and POST a request over a relay, failing over to another relay on + /// transport error. + /// + /// `exclude` is the target the request ultimately reaches — for Payjoin v2 + /// the session directory. A relay equal to it is never selected: a server + /// serving as both relay and directory rejects the resulting self-loop + /// request. + /// + /// `transport` builds and POSTs the request, letting callers keep their own + /// proxy and TLS client configuration while this method owns relay + /// selection and failover. `build` runs once per relay attempt and a + /// transport error marks that relay failed and retries over another. + /// + /// Returns [`RelayPostError::NoRelaysAvailable`] once every configured + /// relay has been marked failed or excluded. Relay failures are scoped to + /// this call: the failed-relay set is cleared on entry, so an outage that + /// retires every relay in one call leaves the next call free to try them + /// again. + pub async fn post_via_relay( + &self, + exclude: Arc, + transport: Arc, + ) -> Result { + let exclude: payjoin::Url = Arc::unwrap_or_clone(exclude).into(); + struct CoreRelayTransport { + inner: Arc, + } + impl payjoin::request_handler::RelayTransport for CoreRelayTransport { + type Request = crate::Request; + type Context = Arc; + type Response = Vec; + type BuildError = RelayBuildError; + type Error = RelayTransportError; + fn build( + &mut self, + relay: &payjoin::Url, + ) -> Result<(crate::Request, Arc), RelayBuildError> { + let built = self.inner.build(Arc::new(relay.clone().into()))?; + Ok((built.request, built.ohttp_context)) + } + async fn post( + &mut self, + request: crate::Request, + ) -> Result, RelayTransportError> { + self.inner.post(request).await + } + } + let transport = CoreRelayTransport { inner: transport }; + match self.inner.post_via_relay(&mut OsRng, &exclude, transport).await { + Ok(request_handler::RelayPost::Posted(body, ohttp_context)) => + Ok(RelayedResponse { body, ohttp_context }), + Ok(request_handler::RelayPost::Expired) => Err(RelayPostError::Expired), + Err(request_handler::PostError::NoRelaysAvailable { last_error }) => + Err(RelayPostError::NoRelaysAvailable { last_error }), + Err(request_handler::PostError::Build(RelayBuildError::Expired)) => + Err(RelayPostError::Expired), + Err(request_handler::PostError::Build(RelayBuildError::Fatal { detail })) => + Err(RelayPostError::Fatal { detail }), + } + } +} + +#[cfg(feature = "_manual-tls")] +#[uniffi::export] +impl RequestHandler { + /// Use a DER-encoded certificate for local HTTPS connections when + /// fetching OHTTP keys. + /// + /// Observed by every reference to this handler rather than creating a + /// forked copy with separate failure tracking. + pub fn set_cert_der(&self, cert_der: Vec) { self.inner.set_cert_der(cert_der); } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn urls() -> Vec { + vec![ + "https://a.example".to_string(), + "https://b.example".to_string(), + "https://c.example".to_string(), + ] + } + + fn canonical_urls() -> Vec { + urls().into_iter().map(|url| Url::parse(url).unwrap().as_string()).collect() + } + + #[test] + fn new_rejects_invalid_url() { + let err = RequestHandler::new(vec!["not a url".to_string()], urls()).unwrap_err(); + assert_eq!(err.to_string(), "invalid scheme"); + } + + // Failover loops are async and need a runtime; they run under the + // `_test-utils` feature, which enables tokio. + #[cfg(feature = "_test-utils")] + mod failover { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Mutex; + + use super::*; + + fn unconfigured() -> Arc { + Arc::new(Url::parse("https://unconfigured.example".to_string()).unwrap()) + } + + fn test_keys() -> Arc { + Arc::new(OhttpKeys::decode(payjoin_test_utils::ohttp_key_config_bytes()).unwrap()) + } + + fn post_request() -> crate::Request { + crate::Request { + url: "https://relay.example".to_string(), + content_type: "message/ohttp-req".to_string(), + body: vec![1, 2, 3], + } + } + + fn built() -> BuiltRequest { + BuiltRequest { + request: post_request(), + ohttp_context: Arc::new(ClientResponse::dangling()), + } + } + + struct AlwaysTransient { + calls: AtomicUsize, + } + + #[async_trait::async_trait] + impl RelayTransport for AlwaysTransient { + fn build(&self, _relay: Arc) -> Result { + Ok(built()) + } + + async fn post(&self, _request: crate::Request) -> Result, RelayTransportError> { + self.calls.fetch_add(1, Ordering::SeqCst); + Err(RelayTransportError::Transient { detail: "down".to_string() }) + } + } + + #[tokio::test] + async fn post_via_relay_retires_relay_and_exhausts() { + let mailroom = RequestHandler::new(urls(), Vec::new()).unwrap(); + let transport = Arc::new(AlwaysTransient { calls: AtomicUsize::new(0) }); + let result = mailroom.post_via_relay(unconfigured(), transport.clone()).await; + match result { + Err(RelayPostError::NoRelaysAvailable { last_error }) => + assert_eq!(last_error.as_deref(), Some("relay transport failure: down")), + Err(other) => panic!("expected relays to be exhausted, got {other:?}"), + Ok(_) => panic!("expected relays to be exhausted, got a response"), + } + assert_eq!(transport.calls.load(Ordering::SeqCst), 3); + } + + struct FixedBuildError { + calls: AtomicUsize, + error: fn() -> RelayBuildError, + } + + #[async_trait::async_trait] + impl RelayTransport for FixedBuildError { + fn build(&self, _relay: Arc) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + Err((self.error)()) + } + + async fn post(&self, _request: crate::Request) -> Result, RelayTransportError> { + unreachable!("post must not run when build fails") + } + } + + #[tokio::test] + async fn post_via_relay_stops_on_expiry() { + let mailroom = RequestHandler::new(urls(), Vec::new()).unwrap(); + let transport = Arc::new(FixedBuildError { + calls: AtomicUsize::new(0), + error: || RelayBuildError::Expired, + }); + let result = mailroom.post_via_relay(unconfigured(), transport.clone()).await; + assert!(matches!(result, Err(RelayPostError::Expired))); + assert_eq!(transport.calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn post_via_relay_stops_on_fatal_build_error() { + let mailroom = RequestHandler::new(urls(), Vec::new()).unwrap(); + let transport = Arc::new(FixedBuildError { + calls: AtomicUsize::new(0), + error: || RelayBuildError::Fatal { detail: "bad".to_string() }, + }); + let result = mailroom.post_via_relay(unconfigured(), transport.clone()).await; + match result { + Err(RelayPostError::Fatal { detail }) => assert_eq!(detail, "bad"), + Err(other) => panic!("expected a fatal build error, got {other:?}"), + Ok(_) => panic!("expected a fatal build error, got a response"), + } + assert_eq!(transport.calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn post_via_relay_never_selects_excluded_relay() { + // The only relay is the excluded directory, so no relay is + // selectable and the transport must never be called. + let mailroom = RequestHandler::new(vec![urls()[0].clone()], Vec::new()).unwrap(); + let transport = Arc::new(AlwaysTransient { calls: AtomicUsize::new(0) }); + let exclude = Arc::new(Url::parse(urls()[0].clone()).unwrap()); + let result = mailroom.post_via_relay(exclude, transport.clone()).await; + match result { + Err(RelayPostError::NoRelaysAvailable { last_error }) => + assert_eq!(last_error, None), + Err(other) => panic!("expected relays to be exhausted, got {other:?}"), + Ok(_) => panic!("expected relays to be exhausted, got a response"), + } + assert_eq!(transport.calls.load(Ordering::SeqCst), 0); + } + + struct TransientThenKeys { + calls: AtomicUsize, + keys: Arc, + } + + #[async_trait::async_trait] + impl OhttpKeysFetcher for TransientThenKeys { + async fn fetch( + &self, + _relay: Arc, + _directory: Arc, + ) -> Result, DirectoryFetchError> { + if self.calls.fetch_add(1, Ordering::SeqCst) == 0 { + return Err(DirectoryFetchError::Transient { detail: "down".to_string() }); + } + Ok(self.keys.clone()) + } + } + + #[tokio::test] + async fn fetch_retries_over_another_relay() { + let mailroom = RequestHandler::new(urls(), urls()).unwrap(); + let fetcher = + Arc::new(TransientThenKeys { calls: AtomicUsize::new(0), keys: test_keys() }); + let result = mailroom.fetch_ohttp_keys(fetcher.clone()).await.unwrap(); + assert!(canonical_urls().contains(&result.directory.as_string())); + // One transient failure, then a success. + assert_eq!(fetcher.calls.load(Ordering::SeqCst), 2); + } + + struct AlwaysDirectoryFails { + calls: AtomicUsize, + directories: Mutex>, + } + + #[async_trait::async_trait] + impl OhttpKeysFetcher for AlwaysDirectoryFails { + async fn fetch( + &self, + _relay: Arc, + directory: Arc, + ) -> Result, DirectoryFetchError> { + self.calls.fetch_add(1, Ordering::SeqCst); + self.directories.lock().unwrap().push(directory.as_string()); + Err(DirectoryFetchError::Directory { detail: "404".to_string() }) + } + } + + #[tokio::test] + async fn fetch_retires_directory_and_gives_up() { + let mailroom = RequestHandler::new(urls(), urls()).unwrap(); + let fetcher = Arc::new(AlwaysDirectoryFails { + calls: AtomicUsize::new(0), + directories: Mutex::new(Vec::new()), + }); + let result = mailroom.fetch_ohttp_keys(fetcher.clone()).await; + match result { + Err(RequestHandlerError::NoDirectoriesAvailable { last_error }) => { + let last_error = last_error.expect("a directory failure should be reported"); + assert!(last_error.contains("404"), "unexpected detail: {last_error}"); + } + Err(other) => panic!("expected directories to be exhausted, got {other:?}"), + Ok(_) => panic!("expected directories to be exhausted, got keys"), + } + // Every directory is attempted exactly once. + assert_eq!(fetcher.calls.load(Ordering::SeqCst), 3); + let mut tried = fetcher.directories.lock().unwrap().clone(); + tried.sort(); + tried.dedup(); + assert_eq!(tried.len(), 3); + } + + struct NeverCalled { + calls: AtomicUsize, + } + + #[async_trait::async_trait] + impl OhttpKeysFetcher for NeverCalled { + async fn fetch( + &self, + _relay: Arc, + _directory: Arc, + ) -> Result, DirectoryFetchError> { + self.calls.fetch_add(1, Ordering::SeqCst); + unreachable!("pinned keys must not be fetched") + } + } + + #[tokio::test] + async fn pinned_ohttp_keys_short_circuit_fetch() { + let mailroom = RequestHandler::new(Vec::new(), urls()).unwrap(); + mailroom.set_ohttp_keys(test_keys()); + let fetcher = Arc::new(NeverCalled { calls: AtomicUsize::new(0) }); + let result = mailroom.fetch_ohttp_keys(fetcher.clone()).await.unwrap(); + assert!(canonical_urls().contains(&result.directory.as_string())); + assert_eq!(fetcher.calls.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn fetch_resets_directory_failures() { + // The first fetch retires every directory (each reports a + // directory-fatal 404); the second starts from a clean directory + // set and succeeds rather than reporting every directory retired + // for the handler's lifetime. + let mailroom = RequestHandler::new(urls(), urls()).unwrap(); + let fetcher = Arc::new(AlwaysDirectoryFails { + calls: AtomicUsize::new(0), + directories: Mutex::new(Vec::new()), + }); + assert!(matches!( + mailroom.fetch_ohttp_keys(fetcher).await, + Err(RequestHandlerError::NoDirectoriesAvailable { .. }) + )); + let fetcher = + Arc::new(TransientThenKeys { calls: AtomicUsize::new(0), keys: test_keys() }); + let result = mailroom.fetch_ohttp_keys(fetcher).await.unwrap(); + assert!(canonical_urls().contains(&result.directory.as_string())); + } + } +} diff --git a/payjoin-ffi/src/uri/mod.rs b/payjoin-ffi/src/uri/mod.rs index cbc6e72b1..bafbcbe39 100644 --- a/payjoin-ffi/src/uri/mod.rs +++ b/payjoin-ffi/src/uri/mod.rs @@ -79,7 +79,7 @@ impl From for payjoin::Url { fn from(value: Url) -> Self { value.0 } } -#[derive(Debug, uniffi::Object)] +#[derive(Clone, Debug, uniffi::Object)] pub struct Url(payjoin::Url); #[uniffi::export] diff --git a/payjoin/CHANGELOG.md b/payjoin/CHANGELOG.md index 055cbfffe..4ef38943e 100644 --- a/payjoin/CHANGELOG.md +++ b/payjoin/CHANGELOG.md @@ -2,6 +2,32 @@ ## Unreleased +### URL Selection + +- Add `payjoin::request_handler::RequestHandler`, which owns relay and + directory selection, tracks failed endpoints, and fails over to + another URL on error. With the `io` feature it also fetches OHTTP + keys +- Add `RequestHandler::post_via_relay` to post a request over a + caller-supplied `RelayTransport`, failing over to another relay on + transport error. Takes only `v2`, not `io` +- Add `RequestHandler::fetch_ohttp_keys_with`, the transport-generic + form of the OHTTP key fetch, parameterized over the new + `DirectoryFatal` trait +- `RequestHandler` takes `&self` and no longer implements `Clone`, so + it can be shared across tasks via `Arc` and concurrent relay posts + do not serialize +- `PostError::NoRelaysAvailable` carries a `last_error` describing the + most recent transport failure +- Relay and directory selection skip the session directory, which a + server serving as both relay and directory would reject as a + self-loop. `post_via_relay` takes the directory for this reason; + `Sender::directory`, `Receiver::directory`, and `PjParam::directory` + expose it +- Failed endpoints are cleared on entry to each failover loop, so a + transient outage no longer retires every relay until the handler is + dropped + ## 1.2.0 This release adds a non-blocking interface to the receiver state machine. diff --git a/payjoin/src/core/mod.rs b/payjoin/src/core/mod.rs index ce01b7c6e..80124274c 100644 --- a/payjoin/src/core/mod.rs +++ b/payjoin/src/core/mod.rs @@ -19,6 +19,10 @@ pub use into_url::{Error as IntoUrlError, IntoUrl}; pub(crate) mod url; pub use url::{ParseError as UrlParseError, Url}; #[cfg(feature = "v2")] +pub mod request_handler; +#[cfg(feature = "v2")] +pub(crate) mod selector; +#[cfg(feature = "v2")] pub mod time; pub mod uri; pub use uri::{PjParam, PjParseError, PjUri, Uri, UriParseError}; diff --git a/payjoin/src/core/receive/v2/mod.rs b/payjoin/src/core/receive/v2/mod.rs index bbb0bbc3c..896782c04 100644 --- a/payjoin/src/core/receive/v2/mod.rs +++ b/payjoin/src/core/receive/v2/mod.rs @@ -442,6 +442,16 @@ impl core::ops::DerefMut for Receiver { fn deref_mut(&mut self) -> &mut Self::Target { &mut self.state } } +impl Receiver { + /// The payjoin directory carrying this session. + /// + /// Pass this as the exclusion target to + /// [`crate::request_handler::RequestHandler::post_via_relay`]: a server + /// serving as both relay and directory rejects the resulting self-loop + /// request. + pub fn directory(&self) -> &Url { &self.session_context.directory } +} + /// The fallback transaction for cancelled or failed receiver session. /// /// See [`Receiver`] for further documentation. diff --git a/payjoin/src/core/request_handler.rs b/payjoin/src/core/request_handler.rs new file mode 100644 index 000000000..3aae92af7 --- /dev/null +++ b/payjoin/src/core/request_handler.rs @@ -0,0 +1,1080 @@ +//! Failover-aware OHTTP relay and payjoin directory management. +//! +//! [`RequestHandler`] owns the selection policy for OHTTP relays and payjoin +//! directories, tracks which endpoints have failed, and drives failover: +//! [`RequestHandler::post_via_relay`] retries a caller-supplied transport +//! ([`RelayTransport`], or [`FnRelayTransport`]) over another relay, and — +//! with the `io` feature — [`RequestHandler::fetch_ohttp_keys`] fetches +//! OHTTP keys with automatic failover, retrying relay failures over another +//! relay and moving directory failures on to another directory. +//! +//! A directory chosen for a session must not change: it is embedded in the +//! BIP21 URI at session creation and recovered from the session event log +//! on resume. [`RequestHandler`] selects a directory only when a new session is +//! created; resumption does not consult it. +//! +//! All selection goes through one internal `UrlSelector` path, so policy +//! stays uniform across integrators and topology-aware ordering (for example +//! AS-aware relay selection) can be introduced later in that one place rather +//! than forked per implementation. +//! +//! A relay equal to the directory it is reaching is never selected: a server +//! serving as both rejects the self-loop request. Exclusion is a filter on +//! candidates, so it composes with any future ordering policy. +//! +//! The `io` feature gates only the network-touching OHTTP key fetch; it is +//! opt-in. [`RequestHandler::post_via_relay`] takes the transport as a +//! [`RelayTransport`], so integrators that already have an HTTP client get +//! relay failover and selection without `io` — closure-based transports can +//! use [`FnRelayTransport`] — while those that enable it also get the OHTTP +//! key fetch with automatic failover. +use bitcoin::secp256k1::rand::Rng; + +use crate::selector::UrlSelector; +use crate::Url; + +/// Selects OHTTP relays and payjoin directories, excluding endpoints marked +/// failed. Failures are scoped to a single failover loop: both +/// [`RequestHandler::post_via_relay`] and +/// [`RequestHandler::fetch_ohttp_keys`] clear recorded relay failures on +/// entry, and the key fetch clears directory failures on entry, so a +/// transient outage that retires every endpoint does not persist to the next +/// call. +/// +/// Methods take `&self`: failure state and pinned configuration are behind +/// `Mutex`es held only for synchronous accesses and never across `.await`, so +/// a `RequestHandler` can be shared across tasks (for example via `Arc`) +/// without an external async lock and without serializing concurrent relay +/// posts. It is not `Clone`: a clone would fork the tracked failure state, +/// so sharing goes through `Arc` and configuration through the setters, which +/// every holder observes. +#[derive(Debug)] +pub struct RequestHandler { + relays: UrlSelector, + directories: UrlSelector, + ohttp_keys: std::sync::Mutex>, + #[cfg(feature = "_manual-tls")] + cert_der: std::sync::Mutex>>, +} + +impl RequestHandler { + /// Deduplicates `relays` and `directories` (preserving order) so uniform + /// selection isn't skewed by an endpoint listed more than once. + pub fn new(relays: Vec, directories: Vec) -> Self { + Self { + relays: UrlSelector::new(relays), + directories: UrlSelector::new(directories), + ohttp_keys: std::sync::Mutex::new(None), + #[cfg(feature = "_manual-tls")] + cert_der: std::sync::Mutex::new(None), + } + } + + /// Pin user-supplied OHTTP keys so [`RequestHandler::fetch_ohttp_keys`] returns + /// them without contacting a directory. + /// + /// Takes `&self`: the pinned keys are observed by every holder of this + /// handler rather than by a forked copy. + pub fn set_ohttp_keys(&self, ohttp_keys: crate::OhttpKeys) { + *self.lock_ohttp_keys() = Some(ohttp_keys); + } + + /// Lock the pinned OHTTP keys, recovering from a poisoned lock rather + /// than panicking: a panic in another thread should not take down + /// selection. + fn lock_ohttp_keys(&self) -> std::sync::MutexGuard<'_, Option> { + self.ohttp_keys.lock().unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// Lock the pinned certificate, recovering from a poisoned lock rather + /// than panicking. + #[cfg(feature = "_manual-tls")] + fn lock_cert_der(&self) -> std::sync::MutexGuard<'_, Option>> { + self.cert_der.lock().unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// Use a DER-encoded certificate for local HTTPS connections when + /// fetching OHTTP keys. + /// + /// Takes `&self`: the certificate is observed by every holder of this + /// handler rather than by a forked copy. + #[cfg(feature = "_manual-tls")] + #[cfg_attr(docsrs, doc(cfg(feature = "_manual-tls")))] + pub fn set_cert_der(&self, cert_der: Vec) { *self.lock_cert_der() = Some(cert_der); } + + /// Pick a relay, never one marked failed and never `exclude`. + /// + /// Pass the session directory so a directory is never reached through + /// itself: a server serving as both relay and directory rejects the + /// self-loop request. + pub(crate) fn select_relay_excluding( + &self, + rng: &mut R, + exclude: &Url, + ) -> Result { + self.relays.select_excluding(rng, exclude).ok_or(Error::NoRelaysAvailable) + } + + /// Pick a payjoin directory, never one marked failed. + pub(crate) fn select_directory(&self, rng: &mut R) -> Result { + self.directories.select(rng).ok_or(Error::NoDirectoriesAvailable) + } + + /// Record a relay transport failure so subsequent selections avoid it. + pub(crate) fn mark_relay_failed(&self, relay: &Url) { self.relays.mark_failed(relay); } + + /// Record a directory failure so subsequent selections avoid it. + #[cfg(test)] + pub(crate) fn mark_directory_failed(&self, directory: &Url) { + self.directories.mark_failed(directory); + } + + /// Clear all recorded relay failures so every configured relay is + /// selectable again. + /// + /// [`RequestHandler::post_via_relay`] and [`RequestHandler::fetch_ohttp_keys`] + /// call this at the start of each call, scoping relay failures to one + /// failover loop: an outage that retires every relay does not persist to + /// the next call. + pub(crate) fn clear_failed_relays(&self) { self.relays.clear_failed(); } + + /// Clear all recorded directory failures so every configured directory is + /// selectable again. + /// + /// [`RequestHandler::fetch_ohttp_keys`] calls this at the start of each + /// fetch, scoping directory failures to one session: a transient outage + /// that retires every directory does not persist to the next session. + pub(crate) fn clear_failed_directories(&self) { self.directories.clear_failed(); } + + /// Return the pinned OHTTP keys, or fetch them through `fetch`, with + /// automatic failover. + /// + /// The transport-generic form of [`RequestHandler::fetch_ohttp_keys`]: + /// `fetch` performs a single attempt per call against a `(relay, + /// directory)` pair, returning the keys or an error implementing + /// [`DirectoryFatal`] so this loop can attribute the failure to the relay + /// or the directory. Relay failures retry over another relay; a + /// directory-fatal failure retires the directory and moves on. + /// + /// Directory failures are scoped to this call: the failed-directory set is + /// cleared on entry, so a directory that failed a previous session is + /// tried again. Relay failures are scoped the same way: the failed-relay + /// set is cleared on entry and after each directory failure, so every + /// directory is attempted with a clean relay set. + /// + /// Returns the directory alongside the keys: a receiver pins the directory + /// in the BIP21 URI at session creation, so the directory is reported even + /// when the keys are pinned and no fetch happened. + #[cfg(feature = "v2")] + #[cfg_attr(docsrs, doc(cfg(feature = "v2")))] + pub async fn fetch_ohttp_keys_with( + &self, + rng: &mut R, + fetch: F, + ) -> Result<(Url, crate::OhttpKeys), Error> + where + R: Rng, + F: Fn(Url, Url) -> Fut, + Fut: std::future::Future>, + E: DirectoryFatal + std::fmt::Display, + { + self.clear_failed_directories(); + self.clear_failed_relays(); + loop { + let directory = self.select_directory(rng)?; + if let Some(ohttp_keys) = self.lock_ohttp_keys().clone() { + return Ok((directory, ohttp_keys)); + } + if let Ok(keys) = + fetch_keys_via_relay_with(&self.relays, &self.directories, &directory, rng, &fetch) + .await + { + return Ok((directory, keys)); + } + // The directory was marked failed (directly, or by exhausting + // every relay against it). Give the next directory a clean set + // of relays to fail over through. + self.clear_failed_relays(); + } + } + + /// Return the pinned OHTTP keys, or fetch them over the network with + /// automatic failover. + /// + /// Relay transport failures retry over another relay. A directory failure + /// (unexpected status or oversized key body) marks that directory failed, + /// resets the relay failure list, and retries over another directory. + /// Returns an error only once every configured endpoint has been marked + /// failed within this call. + /// + /// Directory failures are scoped to this call: the failed-directory set is + /// cleared on entry, so a directory that failed a previous session is + /// tried again rather than staying retired for the handler's lifetime. + #[cfg(feature = "io")] + #[cfg_attr(docsrs, doc(cfg(feature = "io")))] + pub async fn fetch_ohttp_keys( + &self, + rng: &mut R, + ) -> Result<(Url, crate::OhttpKeys), Error> { + #[cfg(feature = "_manual-tls")] + let cert_der = self.lock_cert_der().clone(); + #[cfg(not(feature = "_manual-tls"))] + let cert_der: Option> = None; + self.fetch_ohttp_keys_with(rng, move |relay, directory| { + let cert_der = cert_der.clone(); + async move { fetch_via_relay_impl(cert_der.as_deref(), &relay, &directory).await } + }) + .await + } +} + +/// Relay failover loop behind [`RequestHandler::fetch_ohttp_keys_with`], +/// factored out so tests can drive it with a synthetic `fetch` without touching +/// the network. +/// +/// On return the directory has been marked failed: either the directory itself +/// misbehaved, or every relay failed to reach it. +#[cfg(feature = "v2")] +async fn fetch_keys_via_relay_with( + relays: &UrlSelector, + directories: &UrlSelector, + directory: &Url, + rng: &mut R, + fetch: F, +) -> Result +where + R: Rng, + F: Fn(Url, Url) -> Fut, + Fut: std::future::Future>, + E: DirectoryFatal + std::fmt::Display, +{ + loop { + // A relay equal to the directory would loop back to itself and be + // rejected as a self-loop, so it is excluded up front. When every + // relay is either failed or equal to the directory, selection + // returns `None` and the directory is marked failed below: it + // cannot be reached through a relay at all. + let relay = match relays.select_excluding(rng, directory) { + Some(relay) => relay, + None => { + // A directory that exhausted every relay is marked failed so + // selection moves on instead of retrying it forever. + directories.mark_failed(directory); + return Err(DirectoryAttemptFailure::RelaysExhausted); + } + }; + match fetch(relay.clone(), directory.clone()).await { + Ok(keys) => return Ok(keys), + Err(e) if e.is_directory_fatal() => { + tracing::debug!("Directory {directory} failed via relay {relay}: {e}"); + directories.mark_failed(directory); + return Err(DirectoryAttemptFailure::Directory); + } + Err(e) => { + tracing::debug!("Relay {relay} failed: {e}"); + relays.mark_failed(&relay); + } + } + } +} + +#[cfg(feature = "io")] +async fn fetch_via_relay_impl( + cert_der: Option<&[u8]>, + relay: &Url, + directory: &Url, +) -> Result { + #[cfg(feature = "_manual-tls")] + if let Some(cert_der) = cert_der { + return crate::io::fetch_ohttp_keys_with_cert(relay.as_str(), directory.as_str(), cert_der) + .await; + } + #[cfg(not(feature = "_manual-tls"))] + let _ = cert_der; + crate::io::fetch_ohttp_keys(relay.as_str(), directory.as_str()).await +} + +/// Errors returned by [`RequestHandler`] relay and directory selection. +#[derive(Debug, PartialEq, Eq, Clone)] +#[non_exhaustive] +pub enum Error { + /// Every configured relay has been marked failed. + NoRelaysAvailable, + /// Every configured directory has been marked failed. + NoDirectoriesAvailable, +} + +impl std::fmt::Display for Error { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Error::NoRelaysAvailable => write!(f, "No valid relays available"), + Error::NoDirectoriesAvailable => write!(f, "No valid directories available"), + } + } +} + +impl std::error::Error for Error {} + +/// Whether an OHTTP keys fetch error should be attributed to the directory +/// rather than the relay transporting the request. +/// +/// The classification policy lives here so it is owned by the library: the +/// networked [`RequestHandler::fetch_ohttp_keys`] path implements it for its +/// transport error, and a caller supplying its own transport through +/// [`RequestHandler::fetch_ohttp_keys_with`] implements it over its own error +/// type instead of re-deciding the policy. +#[cfg(feature = "v2")] +#[cfg_attr(docsrs, doc(cfg(feature = "v2")))] +pub trait DirectoryFatal { + /// Returns `true` when the error is the directory's fault (for example a + /// non-2xx status), so retrying over another relay would not help. + fn is_directory_fatal(&self) -> bool; +} + +#[cfg(feature = "io")] +impl DirectoryFatal for crate::io::Error { + fn is_directory_fatal(&self) -> bool { + matches!( + self, + crate::io::Error::UnexpectedStatusCode(_) | crate::io::Error::OhttpKeysBodyTooLarge(_) + ) + } +} + +/// Whether a request-construction error was caused by session expiry. +/// +/// Implemented for the send and receive v2 `CreateRequestError` types so +/// [`RequestHandler::post_via_relay`] can hand expiry back to the caller (which owns +/// the typestate needed to react) instead of flattening it into this loop. +#[cfg(feature = "v2")] +#[cfg_attr(docsrs, doc(cfg(feature = "v2")))] +pub trait RequestExpiry { + /// Returns `true` when the request failed because the session expired. + fn expired(&self) -> bool; +} + +#[cfg(feature = "v2")] +impl RequestExpiry for crate::send::v2::CreateRequestError { + fn expired(&self) -> bool { self.is_expired() } +} + +#[cfg(feature = "v2")] +impl RequestExpiry for crate::receive::v2::CreateRequestError { + fn expired(&self) -> bool { self.is_expired() } +} + +/// Outcome of [`RequestHandler::post_via_relay`]. +#[cfg(feature = "v2")] +#[cfg_attr(docsrs, doc(cfg(feature = "v2")))] +pub enum RelayPost { + /// The request was posted; carries the transport response and the caller + /// context returned by the request builder. + Posted(P, T), + /// The request could not be built because the session had expired. + Expired, +} + +/// Errors from [`RequestHandler::post_via_relay`]. +#[cfg(feature = "v2")] +#[cfg_attr(docsrs, doc(cfg(feature = "v2")))] +#[derive(Debug)] +pub enum PostError { + /// Every configured relay has been marked failed or excluded. Carries the + /// most recent transport failure observed while trying them, when there was + /// one. + NoRelaysAvailable { + /// The most recent transport failure seen while exhausting the relays. + last_error: Option, + }, + /// The request could not be built for a reason other than expiry. + Build(E), +} + +#[cfg(feature = "v2")] +impl std::fmt::Display for PostError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + PostError::NoRelaysAvailable { .. } => write!(f, "No valid relays available"), + PostError::Build(e) => write!(f, "Could not build request: {e}"), + } + } +} + +#[cfg(feature = "v2")] +impl std::error::Error for PostError {} + +/// Builds and POSTs a request through a relay, implemented by the caller. +/// +/// [`RequestHandler::post_via_relay`] selects the relay, drives failover, and +/// tracks failures; a `RelayTransport` builds a fresh request for the selected +/// relay and performs the HTTP request, so callers keep their own HTTP +/// client, proxy, and TLS configuration while the loop owns the retry policy. +#[cfg(feature = "v2")] +#[cfg_attr(docsrs, doc(cfg(feature = "v2")))] +pub trait RelayTransport { + /// The request to POST to the relay. Generic so a caller whose request is + /// not [`crate::Request`] (for example an FFI binding with its own + /// request type) passes it through unchanged rather than reconstructing + /// one. + type Request; + /// Caller context produced by [`RelayTransport::build`] and threaded back + /// on success, so the caller can process the response (for example an + /// OHTTP context). + type Context; + /// The response produced by a successful POST. + type Response; + /// Why building a request failed. Must implement [`RequestExpiry`] so the + /// loop can report session expiry as [`RelayPost::Expired`] instead of a + /// build failure. + type BuildError: RequestExpiry; + /// Why a POST failed. Display-only: a transport error retires the relay + /// and the loop retries over another, so the loop never branches on its + /// structure. + type Error: std::fmt::Display; + + /// Build a fresh request (and its context) addressed to `relay`. + /// + /// Build a new request on every call: reusing or retransmitting + /// identical ciphertext breaks OHTTP privacy, since a relay could observe + /// that a request is being retried. + fn build(&mut self, relay: &Url) -> Result<(Self::Request, Self::Context), Self::BuildError>; + + /// POST a request produced by [`RelayTransport::build`]. + fn post( + &mut self, + request: Self::Request, + ) -> impl std::future::Future>; +} + +/// [`RelayTransport`] over a pair of closures, for callers that already have +/// a build closure and a post closure. +#[cfg(feature = "v2")] +#[cfg_attr(docsrs, doc(cfg(feature = "v2")))] +pub struct FnRelayTransport { + build: B, + post: P, +} + +#[cfg(feature = "v2")] +impl FnRelayTransport { + /// Adapt a `build` closure (relay URL to request plus context) and a + /// `post` closure (request to response) into a [`RelayTransport`]. + pub fn new(build: B, post: P) -> Self { Self { build, post } } +} + +#[cfg(feature = "v2")] +impl RelayTransport for FnRelayTransport +where + B: FnMut(&Url) -> Result<(Rq, T), E>, + E: RequestExpiry, + P: Fn(Rq) -> Fut, + Fut: std::future::Future>, + TransportError: std::fmt::Display, +{ + type Request = Rq; + type Context = T; + type Response = Resp; + type BuildError = E; + type Error = TransportError; + + fn build(&mut self, relay: &Url) -> Result<(Rq, T), E> { (self.build)(relay) } + + fn post( + &mut self, + request: Rq, + ) -> impl std::future::Future> { + (self.post)(request) + } +} + +#[cfg(feature = "v2")] +#[cfg_attr(docsrs, doc(cfg(feature = "v2")))] +impl RequestHandler { + /// Build and POST a request over a relay, failing over to another relay on + /// transport error. + /// + /// `exclude` names the target the request ultimately reaches — for + /// Payjoin v2 the session directory, available from the session types + /// (for example [`crate::receive::v2::Receiver::directory`]). A relay + /// equal to it is never selected: a server serving as both relay and + /// directory rejects the resulting self-loop request. + /// + /// `transport` builds a request per selected relay and performs the POST + /// (see [`RelayTransport`]); the caller context from `build` is threaded + /// back on success. When `build` reports [`RequestExpiry::expired`], the + /// loop stops and returns [`RelayPost::Expired`] so the caller can react + /// with its own typestate. A transport error marks the relay failed and + /// retries over another relay. + /// + /// Returns [`PostError::NoRelaysAvailable`] once every configured relay has + /// been marked failed or excluded, carrying the last transport failure. + /// + /// Relay failures are scoped to this call: the failed-relay set is cleared + /// on entry, so an outage that retires every relay in one call leaves the + /// next call free to try them again. + pub async fn post_via_relay( + &self, + rng: &mut R, + exclude: &Url, + mut transport: X, + ) -> Result, PostError> + where + R: Rng, + X: RelayTransport, + { + self.clear_failed_relays(); + let mut last_error = None; + loop { + let relay = match self.select_relay_excluding(rng, exclude) { + Ok(relay) => relay, + Err(_) => return Err(PostError::NoRelaysAvailable { last_error }), + }; + let (req, ctx) = match transport.build(&relay) { + Ok(r) => r, + Err(e) if e.expired() => return Ok(RelayPost::Expired), + Err(e) => return Err(PostError::Build(e)), + }; + match transport.post(req).await { + Ok(resp) => return Ok(RelayPost::Posted(resp, ctx)), + Err(e) => { + tracing::debug!("Request to relay {relay} failed: {e}"); + last_error = Some(e.to_string()); + self.mark_relay_failed(&relay); + } + } + } + } +} + +/// Why an attempt against one directory ended in failure. The underlying +/// error is logged where the failure is observed, before this is returned. +#[cfg(feature = "v2")] +enum DirectoryAttemptFailure { + /// Every relay failed to reach the directory. + RelaysExhausted, + /// The directory misbehaved; fetching over another relay would not help. + Directory, +} + +#[cfg(test)] +mod tests { + use bitcoin::secp256k1::rand::rngs::StdRng; + use bitcoin::secp256k1::rand::SeedableRng; + + use super::*; + + fn urls() -> Vec { + ["https://a.example", "https://b.example", "https://c.example"] + .iter() + .map(|s| Url::parse(s).unwrap()) + .collect() + } + + #[test] + fn select_directory_fails_when_all_directories_marked_failed() { + let mailroom = RequestHandler::new(urls(), urls()); + for directory in urls() { + mailroom.mark_directory_failed(&directory); + } + let mut rng = StdRng::seed_from_u64(2); + assert_eq!(mailroom.select_directory(&mut rng), Err(Error::NoDirectoriesAvailable)); + } + + #[test] + fn clear_failed_directories_restores_selection() { + let mailroom = RequestHandler::new(urls(), urls()); + for directory in urls() { + mailroom.mark_directory_failed(&directory); + } + let mut rng = StdRng::seed_from_u64(10); + assert!(mailroom.select_directory(&mut rng).is_err()); + mailroom.clear_failed_directories(); + assert!(mailroom.select_directory(&mut rng).is_ok()); + } + + #[test] + fn error_displays_endpoint_kind() { + assert_eq!(Error::NoRelaysAvailable.to_string(), "No valid relays available"); + assert_eq!(Error::NoDirectoriesAvailable.to_string(), "No valid directories available"); + } + #[cfg(feature = "io")] + #[tokio::test] + async fn pinned_ohttp_keys_short_circuit_fetch() { + let keys = crate::OhttpKeys::decode(&payjoin_test_utils::ohttp_key_config_bytes()).unwrap(); + let directories = urls(); + let mailroom = RequestHandler::new(Vec::new(), directories.clone()); + mailroom.set_ohttp_keys(keys); + let mut rng = StdRng::seed_from_u64(5); + // No relays configured: only the pinned keys can satisfy the fetch. + // The selected directory is still reported for BIP21 pinning. + let (directory, _) = mailroom.fetch_ohttp_keys(&mut rng).await.unwrap(); + assert!(directories.contains(&directory)); + } + + #[cfg(feature = "io")] + #[tokio::test] + async fn fetch_fails_without_directories_even_with_pinned_keys() { + let keys = crate::OhttpKeys::decode(&payjoin_test_utils::ohttp_key_config_bytes()).unwrap(); + let mailroom = RequestHandler::new(Vec::new(), Vec::new()); + mailroom.set_ohttp_keys(keys); + let mut rng = StdRng::seed_from_u64(8); + assert!(matches!( + mailroom.fetch_ohttp_keys(&mut rng).await, + Err(Error::NoDirectoriesAvailable) + )); + } + + #[cfg(feature = "io")] + #[tokio::test] + async fn fetch_resets_directory_failures() { + let keys = crate::OhttpKeys::decode(&payjoin_test_utils::ohttp_key_config_bytes()).unwrap(); + let directories = urls(); + let mailroom = RequestHandler::new(Vec::new(), directories.clone()); + mailroom.set_ohttp_keys(keys); + // Retire every directory, as a previous session's outage might have. + for directory in &directories { + mailroom.mark_directory_failed(directory); + } + let mut rng = StdRng::seed_from_u64(9); + // A fresh fetch starts from a clean directory set rather than + // reporting every directory retired for the handler's lifetime. + let (directory, _) = mailroom.fetch_ohttp_keys(&mut rng).await.unwrap(); + assert!(directories.contains(&directory)); + } + + // The pinned certificate is stored behind `&self`, so a later fetch uses + // it and an overwriting call replaces it rather than stacking or + // forking per-caller state. + #[cfg(feature = "_manual-tls")] + #[test] + fn set_cert_der_pins_certificate_for_every_holder() { + let handler = RequestHandler::new(urls(), Vec::new()); + assert!(handler.lock_cert_der().is_none()); + handler.set_cert_der(b"first".to_vec()); + assert_eq!(handler.lock_cert_der().as_deref(), Some(b"first".as_slice())); + handler.set_cert_der(b"second".to_vec()); + assert_eq!(handler.lock_cert_der().as_deref(), Some(b"second".as_slice())); + } + + #[cfg(feature = "io")] + #[test] + fn unexpected_status_is_directory_fatal() { + let err = crate::io::Error::UnexpectedStatusCode(http::StatusCode::NOT_FOUND); + assert!(err.is_directory_fatal()); + } + + #[cfg(feature = "io")] + #[test] + fn oversized_keys_body_is_directory_fatal() { + let err = crate::io::Error::OhttpKeysBodyTooLarge(65_603); + assert!(err.is_directory_fatal()); + } + + #[cfg(feature = "io")] + #[test] + fn transport_error_is_not_directory_fatal() { + // A URL parse error surfaces as an internal (transport-class) error. + let parse_err = Url::parse("invalid url").unwrap_err(); + let err = crate::io::Error::from(parse_err); + assert!(!err.is_directory_fatal()); + } + + #[cfg(feature = "io")] + #[tokio::test] + async fn directory_fatal_error_marks_directory_not_relay() { + let relays = UrlSelector::new(urls()); + let directories = UrlSelector::new(urls()); + let directory = urls()[0].clone(); + let mut rng = StdRng::seed_from_u64(20); + let calls = std::cell::Cell::new(0u32); + let result = fetch_keys_via_relay_with( + &relays, + &directories, + &directory, + &mut rng, + |_relay, _directory| { + calls.set(calls.get() + 1); + let n = calls.get(); + async move { + assert!(n <= 3, "a directory-fatal error must not be retried per relay"); + Err::(crate::io::Error::UnexpectedStatusCode( + http::StatusCode::NOT_FOUND, + )) + } + }, + ) + .await; + assert!(matches!(result, Err(DirectoryAttemptFailure::Directory))); + // The directory is retired immediately; no relay is blamed for it. + for _ in 0..50 { + assert_ne!(directories.select(&mut rng), Some(directory.clone())); + } + assert!(relays.select(&mut rng).is_some()); + } + + #[cfg(feature = "io")] + #[tokio::test] + async fn transport_error_marks_relay_not_directory() { + let relays = UrlSelector::new(urls()); + let directories = UrlSelector::new(urls()); + let directory = urls()[0].clone(); + let mut rng = StdRng::seed_from_u64(21); + let calls = std::cell::Cell::new(0u32); + let result = fetch_keys_via_relay_with( + &relays, + &directories, + &directory, + &mut rng, + |_relay, _directory| { + calls.set(calls.get() + 1); + let n = calls.get(); + async move { + // The relay equal to the directory is excluded up front, + // so only the other two are attempted, once each. + assert!(n <= 2, "failover must retire a relay on every transport error"); + Err::(crate::io::Error::from( + Url::parse("invalid url").unwrap_err(), + )) + } + }, + ) + .await; + assert!(matches!(result, Err(DirectoryAttemptFailure::RelaysExhausted))); + // Every relay other than the excluded one is retired; the directory + // itself is left selectable. The excluded relay was never attempted, + // so it remains selectable on its own — just not against this + // directory. + assert!(directories.select(&mut rng).is_some()); + assert_eq!(relays.select(&mut rng), Some(directory.clone())); + assert_eq!(relays.select_excluding(&mut rng, &directory), None); + } + + #[derive(Debug)] + struct TestExpiry(bool); + + impl RequestExpiry for TestExpiry { + fn expired(&self) -> bool { self.0 } + } + + impl std::fmt::Display for TestExpiry { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "test expiry: {}", self.0) + } + } + + fn req() -> crate::Request { + crate::Request { + url: "https://relay.example".to_string(), + content_type: "message/ohttp-req", + body: Vec::new(), + } + } + + /// A URL absent from `urls()`, so exclusion does not shrink the + /// selectable pool in tests that only exercise posting. + fn unconfigured() -> Url { Url::parse("https://unconfigured.example").unwrap() } + + #[cfg(feature = "v2")] + #[tokio::test] + async fn post_via_relay_returns_response_and_context() { + let mailroom = RequestHandler::new(urls(), Vec::new()); + let mut rng = StdRng::seed_from_u64(11); + let result = mailroom + .post_via_relay( + &mut rng, + &unconfigured(), + FnRelayTransport::new( + |_relay: &Url| Ok::<_, TestExpiry>((req(), 42u8)), + |_req| async { Ok::<_, TestExpiry>("response") }, + ), + ) + .await; + match result { + Ok(RelayPost::Posted(resp, ctx)) => { + assert_eq!(resp, "response"); + assert_eq!(ctx, 42); + } + _ => panic!("expected a posted response"), + } + } + + #[cfg(feature = "v2")] + #[tokio::test] + async fn post_via_relay_reports_expiry() { + let mailroom = RequestHandler::new(urls(), Vec::new()); + let mut rng = StdRng::seed_from_u64(12); + let result = mailroom + .post_via_relay( + &mut rng, + &unconfigured(), + FnRelayTransport::new( + |_relay: &Url| Err::<(crate::Request, u8), _>(TestExpiry(true)), + |_req| async { Ok::<_, TestExpiry>("response") }, + ), + ) + .await; + assert!(matches!(result, Ok(RelayPost::Expired))); + } + + #[cfg(feature = "v2")] + #[tokio::test] + async fn post_via_relay_surfaces_build_error() { + let mailroom = RequestHandler::new(urls(), Vec::new()); + let mut rng = StdRng::seed_from_u64(13); + let result = mailroom + .post_via_relay( + &mut rng, + &unconfigured(), + FnRelayTransport::new( + |_relay: &Url| Err::<(crate::Request, u8), _>(TestExpiry(false)), + |_req| async { Ok::<_, TestExpiry>("response") }, + ), + ) + .await; + assert!(matches!(result, Err(PostError::Build(TestExpiry(false))))); + } + + #[cfg(feature = "v2")] + #[tokio::test] + async fn post_via_relay_fails_over_then_succeeds() { + let mailroom = RequestHandler::new(urls(), Vec::new()); + let mut rng = StdRng::seed_from_u64(14); + let attempts = std::cell::Cell::new(0u32); + let post = |_req: crate::Request| { + let attempt = attempts.get(); + attempts.set(attempt + 1); + async move { + if attempt == 0 { + Err("transport down") + } else { + Ok("response") + } + } + }; + let result = mailroom + .post_via_relay( + &mut rng, + &unconfigured(), + FnRelayTransport::new(|_relay: &Url| Ok::<_, TestExpiry>((req(), ())), post), + ) + .await; + assert!(matches!(result, Ok(RelayPost::Posted("response", ())))); + assert_eq!(attempts.get(), 2); + } + + #[cfg(feature = "v2")] + #[tokio::test] + async fn post_via_relay_exhausts_relays() { + let mailroom = RequestHandler::new(urls(), Vec::new()); + let mut rng = StdRng::seed_from_u64(15); + let attempts = std::cell::Cell::new(0u32); + // Bail out if a no-op `mark_relay_failed` lets the failover loop spin + // forever: failing the test beats hanging the suite. + let post = |_req: crate::Request| { + attempts.set(attempts.get() + 1); + assert!(attempts.get() <= 4, "failover must stop once every relay is marked failed"); + async { Err::<&str, &str>("transport down") } + }; + let result = mailroom + .post_via_relay( + &mut rng, + &unconfigured(), + FnRelayTransport::new(|_relay: &Url| Ok::<_, TestExpiry>((req(), ())), post), + ) + .await; + assert_eq!(attempts.get(), 3); + match result { + Err(PostError::NoRelaysAvailable { last_error }) => + assert_eq!(last_error.as_deref(), Some("transport down")), + _ => panic!("expected relays to be exhausted"), + } + } + + // Relay failures are scoped to one `post_via_relay` call: an outage that + // retires every relay does not persist to the next call. + #[cfg(feature = "v2")] + #[tokio::test] + async fn post_via_relay_recovers_relays_on_next_call() { + let mailroom = RequestHandler::new(urls(), Vec::new()); + let mut rng = StdRng::seed_from_u64(22); + let attempts = std::cell::Cell::new(0u32); + // Fail every relay attempt, then succeed from the next call on. + let post = |_req: crate::Request| { + let attempt = attempts.get(); + attempts.set(attempt + 1); + async move { + if attempt < 3 { + Err("outage") + } else { + Ok("response") + } + } + }; + let result = mailroom + .post_via_relay( + &mut rng, + &unconfigured(), + FnRelayTransport::new(|_relay: &Url| Ok::<_, TestExpiry>((req(), ())), post), + ) + .await; + // The first call exhausted all three relays... + assert!(matches!(result, Err(PostError::NoRelaysAvailable { .. }))); + assert_eq!(attempts.get(), 3); + // ...and the second call starts from a clean relay set rather than + // failing without touching the network. + let result = mailroom + .post_via_relay( + &mut rng, + &unconfigured(), + FnRelayTransport::new(|_relay: &Url| Ok::<_, TestExpiry>((req(), ())), post), + ) + .await; + assert!(matches!(result, Ok(RelayPost::Posted("response", ())))); + assert_eq!(attempts.get(), 4); + } + + #[test] + fn select_relay_excluding_skips_excluded_relay() { + let mailroom = RequestHandler::new(urls(), urls()); + let excluded = urls()[0].clone(); + // With the excluded relay also marked failed the outcome must not + // change: exclusion and failure tracking are independent filters. + mailroom.mark_relay_failed(&urls()[1]); + let mut rng = StdRng::seed_from_u64(16); + for _ in 0..50 { + assert_ne!(mailroom.select_relay_excluding(&mut rng, &excluded), Ok(excluded.clone())); + } + } + + #[test] + fn select_relay_excluding_fails_when_only_excluded_remains() { + let mailroom = RequestHandler::new(vec![urls()[0].clone()], Vec::new()); + let mut rng = StdRng::seed_from_u64(17); + assert_eq!( + mailroom.select_relay_excluding(&mut rng, &urls()[0]), + Err(Error::NoRelaysAvailable) + ); + } + + #[cfg(feature = "v2")] + #[tokio::test] + async fn post_via_relay_never_selects_excluded_relay() { + let mailroom = RequestHandler::new(urls(), Vec::new()); + let excluded = urls()[0].clone(); + let mut rng = StdRng::seed_from_u64(18); + let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let seen_clone = seen.clone(); + let attempts = std::cell::Cell::new(0u32); + // Bail out if a no-op `mark_relay_failed` lets the failover loop spin + // forever: failing the test beats hanging the suite. + let post = |_req: crate::Request| { + attempts.set(attempts.get() + 1); + assert!(attempts.get() <= 4, "failover must stop once every relay is marked failed"); + async { Err::<&str, &str>("transport down") } + }; + let result = mailroom + .post_via_relay( + &mut rng, + &excluded, + FnRelayTransport::new( + |relay: &Url| { + seen_clone.lock().unwrap().push(relay.to_string()); + Ok::<_, TestExpiry>((req(), ())) + }, + post, + ), + ) + .await; + assert!(matches!(result, Err(PostError::NoRelaysAvailable { .. }))); + let seen = seen.lock().unwrap(); + assert!(!seen.is_empty()); + assert!(seen.iter().all(|r| r != excluded.as_str()), "excluded relay must never be built"); + } + + // Failure state is internally synchronized, so concurrent posts over a + // shared `&RequestHandler` run at the same time rather than serializing on + // an external lock held across the await. + #[cfg(feature = "v2")] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn post_via_relay_does_not_serialize_concurrent_posts() { + use std::sync::Arc; + + use tokio::sync::Barrier; + + let mailroom = Arc::new(RequestHandler::new(urls(), Vec::new())); + let barrier = Arc::new(Barrier::new(2)); + let mut handles = Vec::new(); + for seed in [31u64, 32u64] { + let mailroom = mailroom.clone(); + let barrier = barrier.clone(); + handles.push(tokio::spawn(async move { + let mut rng = StdRng::seed_from_u64(seed); + mailroom + .post_via_relay( + &mut rng, + &unconfigured(), + FnRelayTransport::new( + |_relay: &Url| Ok::<_, TestExpiry>((req(), ())), + |_req| { + let barrier = barrier.clone(); + async move { + // Both posts must be in flight at once; + // an external lock held across the await + // would make the first wait on a barrier + // the second can never reach. + barrier.wait().await; + Ok::<_, &str>("response") + } + }, + ), + ) + .await + })); + } + for handle in handles { + let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle) + .await + .expect("concurrent posts must not serialize") + .unwrap(); + assert!(matches!(result, Ok(RelayPost::Posted("response", ())))); + } + } + + #[cfg(feature = "io")] + #[tokio::test] + async fn fetch_never_uses_relay_equal_to_directory() { + let relays = UrlSelector::new(urls()); + let directories = UrlSelector::new(urls()); + let directory = urls()[0].clone(); + let mut rng = StdRng::seed_from_u64(19); + let seen = std::cell::RefCell::new(Vec::new()); + let result = fetch_keys_via_relay_with( + &relays, + &directories, + &directory, + &mut rng, + |relay, target| { + // This covers a mutant as a no-op `mark_failed` lets the failover + // loop spin forever: failing the test beats hanging the suite. + assert!( + seen.borrow().len() < urls().len() - 1, + "failover must stop once every relay is marked failed" + ); + seen.borrow_mut().push((relay, target)); + async { + Err::(crate::io::Error::from( + Url::parse("invalid url").unwrap_err(), + )) + } + }, + ) + .await; + assert!(matches!(result, Err(DirectoryAttemptFailure::RelaysExhausted))); + let seen = seen.borrow(); + assert!(!seen.is_empty()); + assert!( + seen.iter().all(|(relay, target)| relay != target), + "a relay equal to the directory must never be attempted" + ); + } +} diff --git a/payjoin/src/core/selector.rs b/payjoin/src/core/selector.rs new file mode 100644 index 000000000..636541642 --- /dev/null +++ b/payjoin/src/core/selector.rs @@ -0,0 +1,207 @@ +use std::sync::{Mutex, PoisonError}; + +use bitcoin::secp256k1::rand::seq::IteratorRandom; +use bitcoin::secp256k1::rand::{self}; + +use crate::Url; + +/// Picks a URL, excluding any marked failed, so clients share one +/// selection policy instead of each diverging. +/// +/// Failure state is behind a `Mutex` so selection and marking take `&self` and +/// can be shared across tasks without an external lock. The lock is only held +/// for the duration of a synchronous access, never across an `.await`, so the +/// selector can back an async failover loop without serializing it. +/// +/// Kept as the single selection path so a future topology-aware ordering (for +/// example AS-aware relay selection) lands here once, without forking into a +/// separate API that integrators would have to opt into individually. +#[derive(Debug)] +pub struct UrlSelector { + urls: Vec, + failed: Mutex>, +} + +impl UrlSelector { + /// Deduplicates `urls` (preserving order) so uniform selection isn't + /// skewed by a URL listed more than once. + pub fn new(urls: Vec) -> Self { + let mut deduped: Vec = Vec::new(); + for url in urls { + if !deduped.contains(&url) { + deduped.push(url); + } + } + Self { urls: deduped, failed: Mutex::new(Vec::new()) } + } + + /// Lock the failed list, recovering from a poisoned lock rather than + /// panicking: a panic in another thread should not take down selection. + fn failed(&self) -> std::sync::MutexGuard<'_, Vec> { + self.failed.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// Pick a URL, never one marked failed, or `None` when none remain. + pub fn select(&self, rng: &mut R) -> Option { + let failed = self.failed(); + self.urls.iter().filter(|u| !failed.contains(u)).choose(rng).cloned() + } + + /// Pick a URL, never one marked failed and never `exclude`, or `None` + /// when none remain. + /// + /// A relay equal to the selection target would loop back to itself: a + /// server serving as both relay and directory rejects such a request as + /// a self-loop. Callers pass the target (for example the session + /// directory) so that candidate is skipped rather than tried and + /// rejected over the network. URL equality is a heuristic: the same + /// server under a different URL still slips through, and the server's + /// self-loop protection remains authoritative. + pub fn select_excluding(&self, rng: &mut R, exclude: &Url) -> Option { + let failed = self.failed(); + self.urls.iter().filter(|u| !failed.contains(u) && *u != exclude).choose(rng).cloned() + } + + /// Record a transport failure so `select` avoids the URL. + pub fn mark_failed(&self, url: &Url) { + let mut failed = self.failed(); + if !failed.contains(url) { + failed.push(url.clone()); + } + } + + /// Clear all recorded failures so every configured URL is selectable + /// again. + pub fn clear_failed(&self) { self.failed().clear(); } +} + +#[cfg(test)] +mod tests { + use bitcoin::secp256k1::rand::rngs::StdRng; + use bitcoin::secp256k1::rand::SeedableRng; + + use super::*; + + fn urls() -> Vec { + ["https://a.example", "https://b.example", "https://c.example"] + .iter() + .map(|s| Url::parse(s).unwrap()) + .collect() + } + + #[test] + fn select_returns_a_configured_url() { + let selector = UrlSelector::new(urls()); + let mut rng = StdRng::seed_from_u64(1); + let picked = selector.select(&mut rng).expect("a url"); + assert!(urls().contains(&picked)); + } + + #[test] + fn select_never_returns_a_failed_relay() { + let selector = UrlSelector::new(urls()); + 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_never_returns_a_failed_directory() { + let selector = UrlSelector::new(urls()); + let mut rng = StdRng::seed_from_u64(7); + let failed = Url::parse("https://b.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 selector = UrlSelector::new(urls()); + for u in urls() { + selector.mark_failed(&u); + } + let mut rng = StdRng::seed_from_u64(3); + assert_eq!(selector.select(&mut rng), None); + } + + #[test] + fn clear_failed_restores_all_urls() { + let selector = UrlSelector::new(urls()); + for u in urls() { + selector.mark_failed(&u); + } + 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_urls_preserving_order() { + let a = Url::parse("https://a.example").unwrap(); + let b = Url::parse("https://b.example").unwrap(); + let selector = UrlSelector::new(vec![a.clone(), a.clone(), b.clone()]); + assert_eq!(selector.urls, vec![a, b]); + } + + #[test] + fn select_is_none_when_empty() { + let selector = UrlSelector::new(Vec::new()); + let mut rng = StdRng::seed_from_u64(4); + assert_eq!(selector.select(&mut rng), None); + } + + // Selection is uniform across all configured URLs. + #[test] + fn select_excluding_never_returns_the_excluded_url() { + let selector = UrlSelector::new(urls()); + let excluded = urls()[1].clone(); + let mut rng = StdRng::seed_from_u64(8); + for _ in 0..50 { + assert_ne!(selector.select_excluding(&mut rng, &excluded), Some(excluded.clone())); + } + } + + #[test] + fn select_excluding_is_none_when_only_the_excluded_url_remains() { + let selector = UrlSelector::new(urls()); + let excluded = urls()[0].clone(); + let mut rng = StdRng::seed_from_u64(9); + for url in urls().iter().skip(1) { + selector.mark_failed(url); + } + assert_eq!(selector.select_excluding(&mut rng, &excluded), None); + } + + #[test] + fn select_excluding_still_skips_failed_urls() { + let selector = UrlSelector::new(urls()); + let excluded = urls()[0].clone(); + let failed = urls()[1].clone(); + selector.mark_failed(&failed); + let mut rng = StdRng::seed_from_u64(10); + for _ in 0..50 { + let picked = selector.select_excluding(&mut rng, &excluded).unwrap(); + assert_ne!(picked, excluded); + assert_ne!(picked, failed); + } + } + + #[test] + fn select_is_uniform_across_urls() { + let selector = UrlSelector::new(urls()); + let mut seen = std::collections::BTreeSet::new(); + let mut rng = StdRng::seed_from_u64(5); + for _ in 0..200 { + if let Some(u) = selector.select(&mut rng) { + seen.insert(u.to_string()); + } + } + assert_eq!(seen.len(), urls().len(), "random must reach every url"); + } +} diff --git a/payjoin/src/core/send/v2/mod.rs b/payjoin/src/core/send/v2/mod.rs index 2be4b0f58..2642fa55e 100644 --- a/payjoin/src/core/send/v2/mod.rs +++ b/payjoin/src/core/send/v2/mod.rs @@ -256,6 +256,14 @@ impl core::ops::DerefMut for Sender { impl Sender { /// The endpoint in the Payjoin URI pub fn endpoint(&self) -> String { self.session_context.pj_param.endpoint().to_string() } + + /// The payjoin directory carrying this session. + /// + /// Pass this as the exclusion target to + /// [`crate::request_handler::RequestHandler::post_via_relay`]: a server + /// serving as both relay and directory rejects the resulting self-loop + /// request. + pub fn directory(&self) -> &Url { self.session_context.pj_param.directory() } } impl Sender { diff --git a/payjoin/src/core/uri/v2.rs b/payjoin/src/core/uri/v2.rs index 052cc59a8..1e2a89a75 100644 --- a/payjoin/src/core/uri/v2.rs +++ b/payjoin/src/core/uri/v2.rs @@ -136,6 +136,13 @@ impl PjParam { /// previously seen, as it should not be reused across different sessions. pub fn receiver_pubkey(&self) -> &HpkePublicKey { &self.receiver_pubkey } + /// The payjoin directory server carrying this session's mailboxes. + /// + /// Useful for excluding the directory itself from relay selection: a + /// server serving as both relay and directory rejects the resulting + /// self-loop request. + pub fn directory(&self) -> &Url { &self.directory } + pub(crate) fn ohttp_keys(&self) -> &OhttpKeys { &self.ohttp_keys } pub(crate) fn expiration(&self) -> Time { self.expiration }