From 15b24e68580ec8bec54fdcfbf456b39b44cfd9df Mon Sep 17 00:00:00 2001 From: Benalleng Date: Thu, 15 Oct 2026 11:52:35 -0400 Subject: [PATCH 1/4] Add `payjoin::RequestHandler` to allow for core library mailroom management This adds the mailroom management logic directly into the core library allowing implementers to no longer rely on mimicing the reference implementation as closely as possible and instead simply applying the API methods directly from the payjoin library. This feature is gated behind the `io` feature as this will inevitably pull in the `reqwest` dep something that we want to avoid where possible. --- payjoin/CHANGELOG.md | 5 + payjoin/src/core/mod.rs | 4 + payjoin/src/core/request_handler.rs | 356 ++++++++++++++++++++++++++++ payjoin/src/core/selector.rs | 207 ++++++++++++++++ 4 files changed, 572 insertions(+) create mode 100644 payjoin/src/core/request_handler.rs create mode 100644 payjoin/src/core/selector.rs diff --git a/payjoin/CHANGELOG.md b/payjoin/CHANGELOG.md index 055cbfffe..2f19316c0 100644 --- a/payjoin/CHANGELOG.md +++ b/payjoin/CHANGELOG.md @@ -2,6 +2,11 @@ ## Unreleased +### URL Selection + +- Add `payjoin::selector::UrlSelector`, reusable OHTTP relay and payjoin directory selection primitives (uniform-random over non-failed URLs; callers mark failures to fail over), so integrators share one selection policy instead of each diverging +- Add `payjoin::request_handler::RequestHandler`, which owns relay and directory selection (uniform-random over non-failed URLs, via an internal `UrlSelector` so integrators share one selection policy instead of each diverging), tracks failed endpoints, and — with the `io` feature — fetches OHTTP keys with automatic failover (relay failures retry over another relay, directory failures move on to another directory), absorbing the failover loops integrators previously hand-rolled. Selection and failure tracking stay internal to the handler and in-memory per `RequestHandler`. The `io` feature is opt-in: integrators with their own transport drive the same failover without it, while those who enable it get key fetching and failover for free. Routing all selection through one internal path leaves room for topology-aware ordering (for example AS-aware relay selection) to be added later without forking the API + ## 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/request_handler.rs b/payjoin/src/core/request_handler.rs new file mode 100644 index 000000000..2e3b8528c --- /dev/null +++ b/payjoin/src/core/request_handler.rs @@ -0,0 +1,356 @@ +//! 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 — with the `io` +//! feature — fetches OHTTP keys with automatic failover: relay failures +//! retry over another relay, directory failures move 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. +use bitcoin::secp256k1::rand::Rng; + +use crate::selector::UrlSelector; +use crate::Url; + +/// Selects OHTTP relays and payjoin directories, excluding endpoints marked +/// failed for the lifetime of this `RequestHandler`. +/// +/// 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. 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 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 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 fn mark_relay_failed(&self, relay: &Url) { self.relays.mark_failed(relay); } + + /// Record a directory failure so subsequent selections avoid it. + pub fn mark_directory_failed(&self, directory: &Url) { + self.directories.mark_failed(directory); + } + + /// Clear all recorded relay failures so every configured relay is + /// selectable again. + pub fn clear_failed_relays(&self) { self.relays.clear_failed(); } + + /// Return the pinned OHTTP keys, or fetch them 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. + /// + /// 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 = "io")] + #[cfg_attr(docsrs, doc(cfg(feature = "io")))] + pub async fn fetch_ohttp_keys( + &self, + rng: &mut R, + ) -> Result<(Url, crate::OhttpKeys), Error> { + 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) = self.fetch_keys_via_relay(&directory, rng).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(); + } + } + + /// Fetch keys from `directory`, failing over relays on transport errors. + /// + /// On return the directory has been marked failed: either the directory + /// itself misbehaved, or every relay failed to reach it. + #[cfg(feature = "io")] + #[cfg_attr(docsrs, doc(cfg(feature = "io")))] + async fn fetch_keys_via_relay( + &self, + directory: &Url, + rng: &mut R, + ) -> Result { + loop { + let relay = match self.relays.select(rng) { + Some(relay) => relay, + None => { + // A directory that exhausted every relay is marked + // failed so selection moves on instead of retrying it + // forever. + self.directories.mark_failed(directory); + return Err(DirectoryAttemptFailure::RelaysExhausted); + } + }; + let result = self.fetch_via_relay(&relay, directory).await; + match result { + Ok(keys) => return Ok(keys), + Err(e) if is_directory_fatal(&e) => { + tracing::debug!("Directory {directory} failed via relay {relay}: {e}"); + self.directories.mark_failed(directory); + return Err(DirectoryAttemptFailure::Directory); + } + Err(e) => { + tracing::debug!("Relay {relay} failed: {e}"); + self.mark_relay_failed(&relay); + } + } + } + } + + #[cfg(feature = "io")] + #[cfg_attr(docsrs, doc(cfg(feature = "io")))] + async fn fetch_via_relay( + &self, + relay: &Url, + directory: &Url, + ) -> Result { + #[cfg(feature = "_manual-tls")] + let cert_der = self.lock_cert_der().clone(); + #[cfg(feature = "_manual-tls")] + if let Some(cert_der) = cert_der.as_ref() { + return crate::io::fetch_ohttp_keys_with_cert( + relay.as_str(), + directory.as_str(), + cert_der, + ) + .await; + } + 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 {} + +/// 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 = "io")] +enum DirectoryAttemptFailure { + /// Every relay failed to reach the directory. + RelaysExhausted, + /// The directory misbehaved; fetching over another relay would not help. + Directory, +} + +/// Whether an OHTTP keys fetch error should be attributed to the directory +/// rather than the relay transporting the request. +#[cfg(feature = "io")] +pub(crate) fn is_directory_fatal(err: &crate::io::Error) -> bool { + matches!( + err, + crate::io::Error::UnexpectedStatusCode(_) | crate::io::Error::OhttpKeysBodyTooLarge(_) + ) +} + +#[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 error_displays_endpoint_kind() { + assert_eq!(Error::NoRelaysAvailable.to_string(), "No valid relays available"); + assert_eq!(Error::NoDirectoriesAvailable.to_string(), "No valid directories available"); + } + + // 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")] + #[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")] + #[test] + fn unexpected_status_is_directory_fatal() { + let err = crate::io::Error::UnexpectedStatusCode(http::StatusCode::NOT_FOUND); + assert!(is_directory_fatal(&err)); + } + + #[cfg(feature = "io")] + #[test] + fn oversized_keys_body_is_directory_fatal() { + let err = crate::io::Error::OhttpKeysBodyTooLarge(65_603); + assert!(is_directory_fatal(&err)); + } + + #[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!(!is_directory_fatal(&err)); + } + + #[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) + ); + } +} 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"); + } +} From 0a3a5855f51140015dfae5232726a83c2822ead3 Mon Sep 17 00:00:00 2001 From: Benalleng Date: Wed, 14 Oct 2026 13:45:43 -0400 Subject: [PATCH 2/4] Move post_via_relay loop into core `payjoin::RequestHandler` --- payjoin/CHANGELOG.md | 25 +- payjoin/src/core/receive/v2/mod.rs | 10 + payjoin/src/core/request_handler.rs | 934 ++++++++++++++++++++++++---- payjoin/src/core/send/v2/mod.rs | 8 + payjoin/src/core/uri/v2.rs | 7 + 5 files changed, 877 insertions(+), 107 deletions(-) diff --git a/payjoin/CHANGELOG.md b/payjoin/CHANGELOG.md index 2f19316c0..4ef38943e 100644 --- a/payjoin/CHANGELOG.md +++ b/payjoin/CHANGELOG.md @@ -4,8 +4,29 @@ ### URL Selection -- Add `payjoin::selector::UrlSelector`, reusable OHTTP relay and payjoin directory selection primitives (uniform-random over non-failed URLs; callers mark failures to fail over), so integrators share one selection policy instead of each diverging -- Add `payjoin::request_handler::RequestHandler`, which owns relay and directory selection (uniform-random over non-failed URLs, via an internal `UrlSelector` so integrators share one selection policy instead of each diverging), tracks failed endpoints, and — with the `io` feature — fetches OHTTP keys with automatic failover (relay failures retry over another relay, directory failures move on to another directory), absorbing the failover loops integrators previously hand-rolled. Selection and failure tracking stay internal to the handler and in-memory per `RequestHandler`. The `io` feature is opt-in: integrators with their own transport drive the same failover without it, while those who enable it get key fetching and failover for free. Routing all selection through one internal path leaves room for topology-aware ordering (for example AS-aware relay selection) to be added later without forking the API +- 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 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 index 2e3b8528c..3aae92af7 100644 --- a/payjoin/src/core/request_handler.rs +++ b/payjoin/src/core/request_handler.rs @@ -1,29 +1,53 @@ //! 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 — with the `io` -//! feature — fetches OHTTP keys with automatic failover: relay failures -//! retry over another relay, directory failures move on to another -//! directory. +//! 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 for the lifetime of this `RequestHandler`. +/// 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. 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. +/// `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, @@ -83,50 +107,88 @@ impl RequestHandler { /// 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 fn select_relay_excluding(&self, rng: &mut R, exclude: &Url) -> Result { + 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 fn select_directory(&self, rng: &mut R) -> Result { + 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 fn mark_relay_failed(&self, relay: &Url) { self.relays.mark_failed(relay); } + pub(crate) fn mark_relay_failed(&self, relay: &Url) { self.relays.mark_failed(relay); } /// Record a directory failure so subsequent selections avoid it. - pub fn mark_directory_failed(&self, directory: &Url) { + #[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. - pub fn clear_failed_relays(&self) { self.relays.clear_failed(); } + /// + /// [`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(); } - /// Return the pinned OHTTP keys, or fetch them with automatic failover. + /// Clear all recorded directory failures so every configured directory is + /// selectable again. /// - /// 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. + /// [`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. /// - /// 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 = "io")] - #[cfg_attr(docsrs, doc(cfg(feature = "io")))] - pub async fn fetch_ohttp_keys( + /// 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, - ) -> Result<(Url, crate::OhttpKeys), Error> { + 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) = self.fetch_keys_via_relay(&directory, rng).await { + 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 @@ -136,64 +198,100 @@ impl RequestHandler { } } - /// Fetch keys from `directory`, failing over relays on transport errors. + /// 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. /// - /// On return the directory has been marked failed: either the directory - /// itself misbehaved, or every relay failed to reach it. + /// 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")))] - async fn fetch_keys_via_relay( + pub async fn fetch_ohttp_keys( &self, - directory: &Url, rng: &mut R, - ) -> Result { - loop { - let relay = match self.relays.select(rng) { - Some(relay) => relay, - None => { - // A directory that exhausted every relay is marked - // failed so selection moves on instead of retrying it - // forever. - self.directories.mark_failed(directory); - return Err(DirectoryAttemptFailure::RelaysExhausted); - } - }; - let result = self.fetch_via_relay(&relay, directory).await; - match result { - Ok(keys) => return Ok(keys), - Err(e) if is_directory_fatal(&e) => { - tracing::debug!("Directory {directory} failed via relay {relay}: {e}"); - self.directories.mark_failed(directory); - return Err(DirectoryAttemptFailure::Directory); - } - Err(e) => { - tracing::debug!("Relay {relay} failed: {e}"); - self.mark_relay_failed(&relay); - } + ) -> 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")] - #[cfg_attr(docsrs, doc(cfg(feature = "io")))] - async fn fetch_via_relay( - &self, - relay: &Url, - directory: &Url, - ) -> Result { - #[cfg(feature = "_manual-tls")] - let cert_der = self.lock_cert_der().clone(); - #[cfg(feature = "_manual-tls")] - if let Some(cert_der) = cert_der.as_ref() { - return crate::io::fetch_ohttp_keys_with_cert( - relay.as_str(), - directory.as_str(), - cert_der, - ) +#[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; - } - crate::io::fetch_ohttp_keys(relay.as_str(), directory.as_str()).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. @@ -217,9 +315,240 @@ impl std::fmt::Display for Error { 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 = "io")] +#[cfg(feature = "v2")] enum DirectoryAttemptFailure { /// Every relay failed to reach the directory. RelaysExhausted, @@ -227,16 +556,6 @@ enum DirectoryAttemptFailure { Directory, } -/// Whether an OHTTP keys fetch error should be attributed to the directory -/// rather than the relay transporting the request. -#[cfg(feature = "io")] -pub(crate) fn is_directory_fatal(err: &crate::io::Error) -> bool { - matches!( - err, - crate::io::Error::UnexpectedStatusCode(_) | crate::io::Error::OhttpKeysBodyTooLarge(_) - ) -} - #[cfg(test)] mod tests { use bitcoin::secp256k1::rand::rngs::StdRng; @@ -262,25 +581,22 @@ mod tests { } #[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"); + 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()); } - // 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())); + 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() { @@ -308,18 +624,50 @@ mod tests { )); } + #[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!(is_directory_fatal(&err)); + 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!(is_directory_fatal(&err)); + assert!(err.is_directory_fatal()); } #[cfg(feature = "io")] @@ -328,7 +676,261 @@ mod tests { // 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!(!is_directory_fatal(&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] @@ -353,4 +955,126 @@ mod tests { 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/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 } From 940619ee805e8b08d07d6ae676592f3ce215240c Mon Sep 17 00:00:00 2001 From: Benalleng Date: Thu, 15 Oct 2026 11:54:37 -0400 Subject: [PATCH 3/4] Replace the MailroomManager with the payjoin::RequestHandler in payjoin-cli Instead of relying on direct implementation logic for core payjoin success logic we have moved the mailroom manager logic into the payjoin library to ensure that implementers have the least footguns possible. --- payjoin-cli/CHANGELOG.md | 3 + payjoin-cli/src/app/v2/mod.rs | 170 ++++++++++++++----------------- payjoin-cli/src/app/v2/ohttp.rs | 173 +++++++++----------------------- 3 files changed, 122 insertions(+), 224 deletions(-) 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, -} From 44430a18f26c34a05bea4ccb8be0c0aa320a4430 Mon Sep 17 00:00:00 2001 From: Benalleng Date: Wed, 14 Oct 2026 15:26:42 -0400 Subject: [PATCH 4/4] Add RequestHandler for payjoin-ffi --- payjoin-ffi/CHANGELOG.md | 20 + payjoin-ffi/src/lib.rs | 2 + payjoin-ffi/src/ohttp.rs | 9 + payjoin-ffi/src/request_handler.rs | 638 +++++++++++++++++++++++++++++ payjoin-ffi/src/uri/mod.rs | 2 +- 5 files changed, 670 insertions(+), 1 deletion(-) create mode 100644 payjoin-ffi/src/request_handler.rs 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]