diff --git a/payjoin-cli/src/app/v2/mod.rs b/payjoin-cli/src/app/v2/mod.rs index 73247b7e8..1fdf2b866 100644 --- a/payjoin-cli/src/app/v2/mod.rs +++ b/payjoin-cli/src/app/v2/mod.rs @@ -12,6 +12,7 @@ use payjoin::receive::v2::{ ReceiverBuilder, SessionOutcome as ReceiverSessionOutcome, UncheckedOriginalPayload, WantsFeeRange, WantsInputs, WantsOutputs, }; +use payjoin::schedule::PollSchedule; use payjoin::send::v2::{ replay_event_log as replay_sender_event_log, PendingFallback as SenderPendingFallback, PollingForProposal, SendSession, Sender, SenderBuilder, SessionOutcome as SenderSessionOutcome, @@ -34,6 +35,7 @@ mod ohttp; const W_ID: usize = 36; const W_ROLE: usize = 15; const W_STATUS: usize = 15; +const POLL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); /// Delay before retrying a transiently failed state transition, so a /// misbehaving directory or relay is not hammered in a tight loop. @@ -891,34 +893,54 @@ 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 session = sender; + let mut schedule = PollSchedule::new(); + let mut polls = tokio::task::JoinSet::new(); + let next = tokio::time::sleep(schedule.next_gap()); + tokio::pin!(next); + loop { + tokio::select! { + Some(joined) = polls.join_next(), if !polls.is_empty() => { + let (body, ctx): (Vec, _) = match joined { + Ok(Ok(v)) => v, + _ => continue, + }; + match session.clone().process_response(&body, ctx).save(persister) { + Ok(OptionalTransitionOutcome::Progress(psbt)) => { + persister.print("Proposal received. Processing..."); + return Ok(SendSession::Closed(SenderSessionOutcome::Success(psbt))); + } + Ok(OptionalTransitionOutcome::Stasis(_)) => { + persister.print("No response yet."); + } + Err(e) if e.is_transient() => { + tracing::debug!("Transient error polling for proposal, retrying: {e:?}"); + } + Err(re) => { + persister.print(&re); + tracing::debug!("{re:?}"); + return Err(anyhow!("Response error").context(re)); + } + } + } + () = &mut next => { + next.as_mut().reset(tokio::time::Instant::now() + schedule.next_gap()); + let relay = self.mailroom_manager.choose_relay()?; + let (req, ctx) = match session.create_poll_request(relay.as_str()) { + Ok(r) => r, + Err(e) if e.expired() => { + self.cancel_sender_session(persister.session_id(), true)?; + return Ok(SendSession::Closed(SenderSessionOutcome::Aborted)); + } + Err(e) => return Err(e.into()), + }; + let app = self.clone(); + polls.spawn(async move { + let resp = + tokio::time::timeout(POLL_TIMEOUT, app.post_request(req)).await??; + Ok::<_, anyhow::Error>((resp.bytes().await?.to_vec(), ctx)) + }); } - }; - let res = sender.clone().process_response(&response.bytes().await?, ctx).save(persister); - match res { - Ok(OptionalTransitionOutcome::Progress(psbt)) => { - persister.print("Proposal received. Processing..."); - Ok(SendSession::Closed(SenderSessionOutcome::Success(psbt))) - } - Ok(OptionalTransitionOutcome::Stasis(current_state)) => { - persister.print("No response yet."); - Ok(SendSession::PollingForProposal(current_state)) - } - Err(e) if e.is_transient() => { - tracing::debug!("Transient error polling for proposal, retrying: {e:?}"); - let sender = e.transient_state().expect("transient error carries current state"); - tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; - Ok(SendSession::PollingForProposal(sender)) - } - Err(re) => { - persister.print(&re); - tracing::debug!("{re:?}"); - Err(anyhow!("Response error").context(re)) } } } @@ -972,39 +994,58 @@ impl App { } } - /// Poll the directory once for the sender's original proposal. + /// Poll the directory on a Poisson schedule for the sender's original + /// proposal. async fn read_from_directory( &self, session: Receiver, 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 mut schedule = PollSchedule::new(); + let mut polls = tokio::task::JoinSet::new(); + let next = tokio::time::sleep(schedule.next_gap()); + tokio::pin!(next); + loop { + tokio::select! { + Some(joined) = polls.join_next(), if !polls.is_empty() => { + let (body, ctx): (Vec, _) = match joined { + Ok(Ok(v)) => v, + _ => continue, + }; + match session.clone().process_response(&body, ctx).save(persister) { + Ok(OptionalTransitionOutcome::Progress(next_state)) => { + persister.print( + "Got a request from the sender. Responding with a Payjoin proposal.", + ); + return Ok(ReceiveSession::UncheckedOriginalPayload(next_state)); + } + Ok(OptionalTransitionOutcome::Stasis(_)) => {} + Err(e) if e.is_transient() => { + tracing::debug!("Transient error polling for request, retrying: {e:?}"); + } + Err(e) => return Err(e.into()), + } + } + () = &mut next => { + next.as_mut().reset(tokio::time::Instant::now() + schedule.next_gap()); + let relay = self.mailroom_manager.choose_relay()?; + let (req, ctx) = match session.create_poll_request(relay.as_str()) { + Ok(r) => r, + Err(e) if e.expired() => { + self.cancel_receiver_session(persister.session_id(), true)?; + return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); + } + Err(e) => return Err(e.into()), + }; + let app = self.clone(); + polls.spawn(async move { + let resp = + tokio::time::timeout(POLL_TIMEOUT, app.post_request(req)).await??; + Ok::<_, anyhow::Error>((resp.bytes().await?.to_vec(), ctx)) + }); } - }; - let state_transition = session - .process_response(ohttp_response.bytes().await?.to_vec().as_slice(), context) - .save(persister); - match state_transition { - Ok(OptionalTransitionOutcome::Progress(next_state)) => { - persister - .print("Got a request from the sender. Responding with a Payjoin proposal."); - Ok(ReceiveSession::UncheckedOriginalPayload(next_state)) - } - Ok(OptionalTransitionOutcome::Stasis(current_state)) => - Ok(ReceiveSession::Initialized(current_state)), - Err(e) if e.is_transient() => { - tracing::debug!("Transient error polling for request, retrying: {e:?}"); - let session = e.transient_state().expect("transient error carries current state"); - tokio::time::sleep(TRANSIENT_RETRY_DELAY).await; - Ok(ReceiveSession::Initialized(session)) } - Err(e) => Err(e.into()), } } diff --git a/payjoin-cli/tests/e2e.rs b/payjoin-cli/tests/e2e.rs index 9cebbb9b8..dcdb880b2 100644 --- a/payjoin-cli/tests/e2e.rs +++ b/payjoin-cli/tests/e2e.rs @@ -459,7 +459,7 @@ mod e2e { async fn respond_with_payjoin(mut cli_receive_resumer: Child) -> Result<()> { let mut stdout = cli_receive_resumer.stdout.take().expect("Failed to take stdout of child process"); - let timeout = tokio::time::Duration::from_secs(10); + let timeout = tokio::time::Duration::from_secs(45); let res = tokio::time::timeout( timeout, wait_for_stdout_match(&mut stdout, |line| line.contains("Response successful")), @@ -474,7 +474,7 @@ mod e2e { async fn check_payjoin_sent(mut cli_send_resumer: Child) -> Result<()> { let mut stdout = cli_send_resumer.stdout.take().expect("Failed to take stdout of child process"); - let timeout = tokio::time::Duration::from_secs(10); + let timeout = tokio::time::Duration::from_secs(45); let res = tokio::time::timeout( timeout, wait_for_stdout_match(&mut stdout, |line| line.contains("Payjoin sent")), @@ -504,7 +504,7 @@ mod e2e { async fn check_resume_completed(mut cli_resumer: Child) -> Result<()> { let mut stdout = cli_resumer.stdout.take().expect("Failed to take stdout of child process"); - let timeout = tokio::time::Duration::from_secs(10); + let timeout = tokio::time::Duration::from_secs(45); let res = tokio::time::timeout( timeout, wait_for_stdout_match(&mut stdout, |line| line.ends_with("Session completed.")), diff --git a/payjoin-mailroom/src/db/files.rs b/payjoin-mailroom/src/db/files.rs index 1f6a77ece..e962f8906 100644 --- a/payjoin-mailroom/src/db/files.rs +++ b/payjoin-mailroom/src/db/files.rs @@ -238,6 +238,15 @@ impl DbTrait for FilesDb { Ok(guard.post_v2(id, payload).await?) } + async fn peek_v2_payload( + &self, + id: &ShortId, + ) -> Result>>, DbError> { + let mut guard = self.mailboxes.lock().await; + Ok(guard.read(id).await?) + } + + // Unused by GET after the non-blocking switch; v2 waitmap removal is a follow-up. async fn wait_for_v2_payload( &self, id: &ShortId, @@ -1070,4 +1079,37 @@ mod tests { Ok(()) } + + #[tokio::test(start_paused = true)] + async fn peek_returns_immediately_on_empty_mailbox() { + let dir = tempfile::tempdir().unwrap(); + let db = FilesDb::init( + Duration::from_secs(30), + dir.path().to_owned(), + Duration::from_secs(60 * 60 * 24 * 7), + ) + .await + .unwrap(); + let id = ShortId([0u8; 8]); + let start = tokio::time::Instant::now(); + let got = db.peek_v2_payload(&id).await.expect("peek"); + assert!(got.is_none()); + assert_eq!(start.elapsed(), Duration::ZERO, "peek must not block"); + } + + #[tokio::test] + async fn peek_returns_present_payload() { + let dir = tempfile::tempdir().unwrap(); + let db = FilesDb::init( + Duration::from_millis(10), + dir.path().to_owned(), + Duration::from_secs(60 * 60 * 24 * 7), + ) + .await + .unwrap(); + let id = ShortId([0u8; 8]); + db.post_v2_payload(&id, b"hi".to_vec()).await.unwrap().unwrap(); + let got = db.peek_v2_payload(&id).await.expect("peek").expect("present"); + assert_eq!(&got[..], b"hi"); + } } diff --git a/payjoin-mailroom/src/db/mod.rs b/payjoin-mailroom/src/db/mod.rs index 210bac4e2..73a878536 100644 --- a/payjoin-mailroom/src/db/mod.rs +++ b/payjoin-mailroom/src/db/mod.rs @@ -72,6 +72,12 @@ pub trait Db: Clone + Send + Sync + 'static { mailbox_id: &ShortId, ) -> impl Future>, Error>> + Send; + /// Read a stored v2 payload if present, without waiting. + fn peek_v2_payload( + &self, + mailbox_id: &ShortId, + ) -> impl Future>>, Error>> + Send; + /// Write a v1 response payload. fn post_v1_response( &self, @@ -91,6 +97,7 @@ pub trait Db: Clone + Send + Sync + 'static { pub enum DbRequest { PostV2Payload { mailbox_id: ShortId, payload: Vec }, WaitForV2Payload { mailbox_id: ShortId }, + PeekV2Payload { mailbox_id: ShortId }, PostV1Response { mailbox_id: ShortId, payload: Vec }, PostV1RequestAndWaitForResponse { mailbox_id: ShortId, payload: Vec }, } @@ -99,6 +106,7 @@ pub enum DbRequest { pub enum DbResponse { PostV2Payload(Option<()>), WaitForV2Payload(Arc>), + PeekV2Payload(Option>>), PostV1Response(()), PostV1RequestAndWaitForResponse(Arc>), } @@ -134,6 +142,8 @@ impl Service for FilesDbService { Ok(DbResponse::PostV2Payload(db.post_v2_payload(&mailbox_id, payload).await?)), DbRequest::WaitForV2Payload { mailbox_id } => Ok(DbResponse::WaitForV2Payload(db.wait_for_v2_payload(&mailbox_id).await?)), + DbRequest::PeekV2Payload { mailbox_id } => + Ok(DbResponse::PeekV2Payload(db.peek_v2_payload(&mailbox_id).await?)), DbRequest::PostV1Response { mailbox_id, payload } => { db.post_v1_response(&mailbox_id, payload).await?; Ok(DbResponse::PostV1Response(())) @@ -199,6 +209,21 @@ impl Db for DbServiceAdapter { } } + async fn peek_v2_payload( + &self, + mailbox_id: &ShortId, + ) -> Result>>, Error> { + let response = self + .inner + .clone() + .oneshot(DbRequest::PeekV2Payload { mailbox_id: *mailbox_id }) + .await?; + match response { + DbResponse::PeekV2Payload(result) => Ok(result), + _ => Err(Self::invalid_response("peek_v2_payload")), + } + } + async fn post_v1_response( &self, mailbox_id: &ShortId, @@ -272,6 +297,14 @@ impl Db for MetricsDb { self.inner.wait_for_v2_payload(mailbox_id).await } + async fn peek_v2_payload( + &self, + mailbox_id: &ShortId, + ) -> Result>>, Error> { + self.metrics.record_short_id(mailbox_id); + self.inner.peek_v2_payload(mailbox_id).await + } + async fn post_v1_response( &self, mailbox_id: &ShortId, diff --git a/payjoin-mailroom/src/directory.rs b/payjoin-mailroom/src/directory.rs index 038cc4070..9a4750209 100644 --- a/payjoin-mailroom/src/directory.rs +++ b/payjoin-mailroom/src/directory.rs @@ -284,8 +284,16 @@ impl Service { async fn get_mailbox(&self, id: &str) -> Result, HandlerError> { let id = ShortId::from_str(id)?; - let timeout_response = Response::builder().status(StatusCode::ACCEPTED).body(empty())?; - handle_peek(self.db.wait_for_v2_payload(&id).await, timeout_response) + let empty_response = Response::builder().status(StatusCode::ACCEPTED).body(empty())?; + match self.db.peek_v2_payload(&id).await { + Ok(Some(payload)) => Ok(Response::new(full((*payload).clone()))), + Ok(None) => Ok(empty_response), + Err(DbError::Operational(err)) => { + error!("Storage error: {err}"); + Err(HandlerError::InternalServerError(anyhow::Error::msg("Internal server error"))) + } + Err(_) => Ok(empty_response), + } } /// Screen a V1 PSBT body against the address blocklist. @@ -872,6 +880,25 @@ mod tests { } } + #[tokio::test(start_paused = true)] + async fn get_mailbox_returns_immediately_when_empty() { + let svc = test_service(None).await; + let id = valid_short_id_path(); + let start = tokio::time::Instant::now(); + let res = svc.get_mailbox(&id).await.expect("get_mailbox"); + assert_eq!(res.status(), StatusCode::ACCEPTED); + assert_eq!(start.elapsed(), Duration::ZERO, "GET must not block"); + } + + #[tokio::test] + async fn get_mailbox_returns_payload_when_present() { + let svc = test_service(None).await; + let id = valid_short_id_path(); + svc.post_mailbox(&id, Body::from(b"hi".to_vec())).await.expect("post"); + let res = svc.get_mailbox(&id).await.expect("get_mailbox"); + assert_eq!(res.status(), StatusCode::OK); + } + #[tokio::test] async fn post_mailbox_records_short_id_cardinality() { use opentelemetry_sdk::metrics::{ diff --git a/payjoin/src/core/mod.rs b/payjoin/src/core/mod.rs index ff1fdb2af..70005e748 100644 --- a/payjoin/src/core/mod.rs +++ b/payjoin/src/core/mod.rs @@ -19,6 +19,8 @@ pub use into_url::{Error as IntoUrlError, IntoUrl}; pub(crate) mod url; pub use url::{ParseError as UrlParseError, Url}; #[cfg(feature = "v2")] +pub mod schedule; +#[cfg(feature = "v2")] pub mod time; pub mod uri; pub use uri::{PjParam, PjParseError, PjUri, Uri, UriExt}; diff --git a/payjoin/src/core/schedule.rs b/payjoin/src/core/schedule.rs new file mode 100644 index 000000000..cd3f9e7d9 --- /dev/null +++ b/payjoin/src/core/schedule.rs @@ -0,0 +1,80 @@ +use std::collections::hash_map::RandomState; +use std::hash::{BuildHasher, Hasher}; +use std::time::Duration; + +/// Mean gap of the Poisson poll schedule. The directory learns only this rate, +/// so it must stay uniform across clients, not a per-user knob. +pub const POLL_MEAN: Duration = Duration::from_secs(5); + +/// Samples inter-poll gaps from an Exp(1/mean) distribution. Emit polls on +/// this clock independently of responses (reset on fire, before awaiting the +/// poll) so the observed interval is the gap, not gap + round-trip. +#[derive(Debug)] +pub struct PollSchedule { + state: u64, + mean: Duration, +} + +impl PollSchedule { + /// Create a schedule seeded from OS entropy at the standard [`POLL_MEAN`] rate. + pub fn new() -> Self { + // Seed from OS entropy via the standard library's randomly-keyed hasher: + // hashing a fixed value mixes those random keys into a u64. + let mut h = RandomState::new().build_hasher(); + h.write_u64(0); + Self { state: h.finish(), mean: POLL_MEAN } + } + + /// Sample the next inter-poll gap. + pub fn next_gap(&mut self) -> Duration { + Duration::from_secs_f64(-self.mean.as_secs_f64() * self.next_uniform().ln()) + } + + /// Draw the next uniform sample in (0, 1) from the SplitMix64 state. + fn next_uniform(&mut self) -> f64 { + // SplitMix64 (reference constants: golden-ratio increment + two mixers) + self.state = self.state.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.state; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^= z >> 31; + ((z >> 11) as f64 + 1.0) / (1u64 << 53) as f64 + } +} + +impl Default for PollSchedule { + fn default() -> Self { Self::new() } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn from_seed(seed: u64, mean: Duration) -> PollSchedule { PollSchedule { state: seed, mean } } + + #[test] + fn uniform_sequence_is_pinned() { + let mut s = from_seed(1, Duration::from_secs(5)); + assert_eq!(s.next_uniform(), 0.566561575172281); + assert_eq!(s.next_uniform(), 0.7457817572627012); + assert_eq!(s.next_uniform(), 0.9710027535867963); + } + + #[test] + fn same_seed_is_deterministic() { + let mut a = from_seed(1, Duration::from_secs(5)); + let mut b = from_seed(1, Duration::from_secs(5)); + for _ in 0..100 { + assert_eq!(a.next_gap(), b.next_gap()); + } + } + + #[test] + fn mean_is_near_lambda_inverse() { + let mut s = from_seed(440, Duration::from_secs(5)); + let n = 20_000; + let total: f64 = (0..n).map(|_| s.next_gap().as_secs_f64()).sum(); + let mean = total / n as f64; + assert!((mean - 5.0).abs() < 0.2, "mean gap {mean} not ~5s"); + } +}