Poll the directory on a Poisson schedule

Author: bc1cindy

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

@@ -30,11 +30,16 @@ use crate::db::v2::{ReceiverPersister, SenderPersister, SessionId};
 use crate::db::Database;
 
 mod ohttp;
+mod schedule;
 
 const W_ID: usize = 12;
 const W_ROLE: usize = 25;
 const W_DONE: usize = 15;
 const W_STATUS: usize = 15;
+// 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.
+const POLL_MEAN: std::time::Duration = std::time::Duration::from_secs(5);
+const POLL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
 
 #[derive(Clone)]
 pub(crate) struct App {
@@ -704,27 +709,44 @@ impl App {
         sender: Sender<PollingForProposal>,
         persister: &SenderPersister,
     ) -> Result<()> {
-        let mut session = sender.clone();
-        // Long poll until we get a response
+        let session = sender;
+        let mut schedule = schedule::Poisson::new(POLL_MEAN);
+        let mut polls = tokio::task::JoinSet::new();
+        let next = tokio::time::sleep(schedule.next_gap());
+        tokio::pin!(next);
         loop {
-            let (response, ctx) =
-                self.post_via_relay(|relay| session.create_poll_request(relay)).await?;
-            let res = session.process_response(&response.bytes().await?, ctx).save(persister);
-            match res {
-                Ok(OptionalTransitionOutcome::Progress(psbt)) => {
-                    println!("Proposal received. Processing...");
-                    self.process_pj_response(psbt)?;
-                    return Ok(());
-                }
-                Ok(OptionalTransitionOutcome::Stasis(current_state)) => {
-                    println!("No response yet.");
-                    session = current_state;
-                    continue;
+            tokio::select! {
+                Some(joined) = polls.join_next(), if !polls.is_empty() => {
+                    let (body, ctx): (Vec<u8>, _) = match joined {
+                        Ok(Ok(v)) => v,
+                        _ => continue,
+                    };
+                    match session.clone().process_response(&body, ctx).save(persister) {
+                        Ok(OptionalTransitionOutcome::Progress(psbt)) => {
+                            println!("Proposal received. Processing...");
+                            self.process_pj_response(psbt)?;
+                            return Ok(());
+                        }
+                        Ok(OptionalTransitionOutcome::Stasis(_)) => {
+                            println!("No response yet.");
+                        }
+                        Err(re) => {
+                            println!("{re}");
+                            tracing::debug!("{re:?}");
+                            return Err(anyhow!("Response error").context(re));
+                        }
+                    }
                 }
-                Err(re) => {
-                    println!("{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.relay_manager.choose_relay()?;
+                    let (req, ctx) = session.create_poll_request(relay.as_str())?;
+                    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))
+                    });
                 }
             }
         }
@@ -735,24 +757,37 @@ impl App {
         session: Receiver<Initialized>,
         persister: &ReceiverPersister,
     ) -> Result<Receiver<UncheckedOriginalPayload>> {
-        let mut session = session;
+        let mut schedule = schedule::Poisson::new(POLL_MEAN);
+        let mut polls = tokio::task::JoinSet::new();
+        let next = tokio::time::sleep(schedule.next_gap());
+        tokio::pin!(next);
         loop {
-            println!("Polling receive request...");
-            let (ohttp_response, context) =
-                self.post_via_relay(|relay| session.create_poll_request(relay)).await?;
-            let state_transition = session
-                .process_response(ohttp_response.bytes().await?.to_vec().as_slice(), context)
-                .save(persister);
-            match state_transition {
-                Ok(OptionalTransitionOutcome::Progress(next_state)) => {
-                    println!("Got a request from the sender. Responding with a Payjoin proposal.");
-                    return Ok(next_state);
+            tokio::select! {
+                Some(joined) = polls.join_next(), if !polls.is_empty() => {
+                    let (body, ctx): (Vec<u8>, _) = match joined {
+                        Ok(Ok(v)) => v,
+                        _ => continue,
+                    };
+                    match session.clone().process_response(&body, ctx).save(persister) {
+                        Ok(OptionalTransitionOutcome::Progress(next_state)) => {
+                            println!("Got a request from the sender. Responding with a Payjoin proposal.");
+                            return Ok(next_state);
+                        }
+                        Ok(OptionalTransitionOutcome::Stasis(_)) => {}
+                        Err(e) => return Err(e.into()),
+                    }
                 }
-                Ok(OptionalTransitionOutcome::Stasis(current_state)) => {
-                    session = current_state;
-                    continue;
+                () = &mut next => {
+                    next.as_mut().reset(tokio::time::Instant::now() + schedule.next_gap());
+                    let relay = self.relay_manager.choose_relay()?;
+                    let (req, ctx) = session.create_poll_request(relay.as_str())?;
+                    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))
+                    });
                 }
-                Err(e) => return Err(e.into()),
             }
         }
     }

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

@@ -0,0 +1,58 @@
+use std::collections::hash_map::RandomState;
+use std::hash::{BuildHasher, Hasher};
+use std::time::Duration;
+
+/// Inter-poll gaps come from an independent Exp(1/mean) clock, so the
+/// interval never depends on the previous response latency.
+pub(crate) struct Poisson {
+    state: u64,
+    mean: Duration,
+}
+
+impl Poisson {
+    pub(crate) fn new(mean: Duration) -> Self {
+        // Seed from OS entropy without adding a crate: RandomState's hasher is randomly
+        // keyed; hashing a fixed value mixes those keys into a u64.
+        let mut h = RandomState::new().build_hasher();
+        h.write_u64(0);
+        Self { state: h.finish(), mean }
+    }
+
+    pub(crate) fn next_gap(&mut self) -> Duration {
+        // SplitMix64
+        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;
+        let u = ((z >> 11) as f64 + 1.0) / ((1u64 << 53) as f64 + 1.0); // uniform in (0,1)
+        Duration::from_secs_f64(-self.mean.as_secs_f64() * u.ln())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use super::*;
+
+    fn from_seed(seed: u64, mean: Duration) -> Poisson { Poisson { state: seed, mean } }
+
+    #[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");
+    }
+}

payjoin-cli/tests/e2e.rs

@@ -421,7 +421,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")),
@@ -436,7 +436,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")),
@@ -466,7 +466,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| {