diff --git a/apps/remote/src/client.rs b/apps/remote/src/client.rs index c932a99..2f7c746 100644 --- a/apps/remote/src/client.rs +++ b/apps/remote/src/client.rs @@ -6,17 +6,14 @@ use std::io::Read; use std::path::Path; -use std::time::Duration; use collapse_core::compression::compress_tar_dir; use collapse_core::Algorithm; use crate::protocol::{self, Progress}; +use crate::waiting::{self, Poller}; use crate::RemoteError; -/// How often the job status is polled while the server compresses. -const POLL_INTERVAL: Duration = Duration::from_millis(200); - /// Compress a file or a whole directory on a remote server and return the /// archive bytes. /// @@ -131,20 +128,33 @@ fn create_job( parse_json(response) } -/// Poll `GET /jobs/{id}` until the job is ready (Ok) or gives up (Err). -fn wait_for_completion(base: &str, job_id: &str) -> Result<(), RemoteError> { - loop { - let response = ureq::get(&format!("{base}/jobs/{job_id}")) - .call() - .map_err(|e| remote_error(base, e))?; +/// Ask `GET /jobs/{id}` once. The loop that decides when to ask again lives in +/// [`crate::waiting`]; this is only the half that needs a socket. +struct HttpPoller<'a> { + base: &'a str, + job_id: &'a str, +} - match protocol::progress_of(&parse_json(response)?)? { - Progress::Ready => return Ok(()), - Progress::Waiting => std::thread::sleep(POLL_INTERVAL), - } +impl Poller for HttpPoller<'_> { + fn poll(&self) -> Result { + let response = ureq::get(&format!("{}/jobs/{}", self.base, self.job_id)) + .call() + .map_err(|e| remote_error(self.base, e))?; + protocol::progress_of(&parse_json(response)?) } } +/// Poll `GET /jobs/{id}` until the job is ready (Ok) or gives up (Err). +/// +/// The wait starts at [`waiting::FIRST_POLL_DELAY`] and doubles to +/// [`waiting::MAX_POLL_DELAY`]. It used to be that ceiling from the first wait +/// onwards, so a job the server had already finished still cost the caller +/// 200 ms of sleeping (issue #48). +fn wait_for_completion(base: &str, job_id: &str) -> Result<(), RemoteError> { + let poller = HttpPoller { base, job_id }; + waiting::wait_for(&waiting::RealSleeper, &poller).map(|_| ()) +} + /// `GET /jobs/{id}/download`: the archive bytes. fn download(base: &str, job_id: &str) -> Result, RemoteError> { let response = ureq::get(&format!("{base}/jobs/{job_id}/download")) diff --git a/apps/remote/src/lib.rs b/apps/remote/src/lib.rs index 18ed51e..ec69e1b 100644 --- a/apps/remote/src/lib.rs +++ b/apps/remote/src/lib.rs @@ -8,7 +8,9 @@ //! //! The crate is split so the decisions can be tested without a server: //! [`protocol`] holds the pure ones (URL building, reading the server's JSON, -//! what to do with each job status) and the HTTP plumbing stays private. +//! what to do with each job status), [`waiting`] holds the poll loop and the +//! two traits that let a test drive it without spending the time it is +//! deciding how to spend, and the HTTP plumbing stays private. //! //! It exists as its own crate because more than one front-end needs it: the //! CLI's `--server` flag today, the desktop app next. Duplicating the exchange @@ -18,6 +20,7 @@ mod client; mod error; pub mod protocol; +pub mod waiting; pub use client::{check_health, compress_path}; pub use error::RemoteError; diff --git a/apps/remote/src/waiting.rs b/apps/remote/src/waiting.rs new file mode 100644 index 0000000..71d39d5 --- /dev/null +++ b/apps/remote/src/waiting.rs @@ -0,0 +1,120 @@ +//! The poll loop: how long to wait between asking, and the seam that lets it +//! be tested without spending the time it is deciding how to spend. +//! +//! Kept apart from [`crate::protocol`] on purpose. That module answers "what +//! does this JSON mean"; this one answers "how often should I ask". They used +//! to be one file and the split is what makes the second question answerable +//! by a test in microseconds. +//! +//! The inversion is two one-method traits, [`Sleeper`] and [`Poller`]. Between +//! them a test can express the thing that actually matters and that no stub +//! server can express deterministically: **a job that takes exactly N +//! milliseconds**. The fake sleeper accumulates virtual time instead of +//! spending real time, and the fake poller reports the job finished once that +//! accumulation passes N. +//! +//! [`wait_for`] returns [`Waited`], its own account of what it did. That is +//! deliberate: a test asserting on the returned count is asserting on the +//! subject, while a test reaching into the fake to count calls is asserting on +//! the fake. The second kind passes when the fake is wrong. + +use std::time::Duration; + +use crate::protocol::Progress; +use crate::RemoteError; + +/// How long to wait before the **first** re-poll of a job the server has not +/// finished yet. +/// +/// Short on purpose. Nearly every archive a person compresses is done in less +/// time than a person notices, and the old schedule waited a flat 200 ms +/// before asking a second time, so a five byte file took ~235 ms end to end +/// with almost none of that spent compressing (issue #48). +/// +/// Not zero, and not one millisecond: the point is to stop making a finished +/// job wait, not to spin on a server that is genuinely busy. +pub const FIRST_POLL_DELAY: Duration = Duration::from_millis(10); + +/// The ceiling the wait grows to, and the interval a long job settles into. +/// +/// Deliberately the old fixed interval, so nothing about a job that takes +/// minutes changes: it reaches this after five polls and stays here. +/// +/// It is also the bound on how much *worse* the backoff can be than the old +/// schedule. A job that finishes just after the ramp is polled again a whole +/// ceiling later, where the flat schedule might have caught it sooner. That +/// band is narrow and bounded, and `the_backoff_is_never_worse_by_more_than_ +/// one_ceiling` pins it. +pub const MAX_POLL_DELAY: Duration = Duration::from_millis(200); + +/// The wait before the next poll, given the wait before the last one. +/// +/// Doubles until it reaches [`MAX_POLL_DELAY`], so the schedule is +/// 10, 20, 40, 80, 160, 200, 200, ... It reaches the ceiling in 310 ms and +/// costs a job of any real length about three extra requests over its whole +/// life, against saving ~190 ms on every job that was already done. +pub fn next_delay(previous: Duration) -> Duration { + previous.saturating_mul(2).min(MAX_POLL_DELAY) +} + +/// The passage of time, injected rather than called directly. +/// +/// One method, and it is the only thing the loop does that a test cannot +/// afford to let happen for real. +pub trait Sleeper { + fn sleep(&self, delay: Duration); +} + +/// One question to the server: how is the job doing? +/// +/// The address and the job id are the implementation's business, so this takes +/// no arguments: a test's fake has no server to address. +pub trait Poller { + fn poll(&self) -> Result; +} + +/// Real time. +pub struct RealSleeper; + +impl Sleeper for RealSleeper { + fn sleep(&self, delay: Duration) { + std::thread::sleep(delay); + } +} + +/// What [`wait_for`] did, so a caller (or a test) can account for it without +/// inspecting the collaborators it was handed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Waited { + /// How many times the server was asked, including the one that settled it. + pub polls: u32, + /// How long was spent sleeping between those questions. Not wall clock: + /// the requests themselves are not counted, because this is the part the + /// schedule controls. + pub slept: Duration, +} + +/// Poll until the job settles, waiting a little longer each time. +/// +/// **Unbounded, still.** A server that answers `compressing` forever keeps this +/// running forever (issue #71). A backoff changes how often it asks, not +/// whether it ever stops. The seam here makes a deadline a few lines, and a +/// deliberately separate change. +pub fn wait_for(sleeper: &dyn Sleeper, poller: &dyn Poller) -> Result { + let mut delay = FIRST_POLL_DELAY; + let mut account = Waited { + polls: 0, + slept: Duration::ZERO, + }; + loop { + account.polls += 1; + match poller.poll()? { + Progress::Ready => return Ok(account), + Progress::Waiting => { + sleeper.sleep(delay); + account.slept += delay; + delay = next_delay(delay); + } + } + } +} diff --git a/apps/remote/tests/client.rs b/apps/remote/tests/client.rs index b31b67c..77db9c1 100644 --- a/apps/remote/tests/client.rs +++ b/apps/remote/tests/client.rs @@ -518,3 +518,78 @@ fn a_missing_source_fails_before_any_request() { .expect_err("the source does not exist"); assert!(matches!(error, RemoteError::Io(_)), "got {error:?}"); } + +// ------------------------------------------------------------- the backoff -- + +/// A job still compressing, so the client keeps polling. +const COMPRESSING_JOB: &str = r#"{"job_id":"stub","name":"notes.txt","archive_name":"notes.txt.zip","algorithm":"zip","level":3,"envelope":"none","status":"compressing","error_message":null}"#; + +/// Issue #48: the wait between polls used to be a flat 200 ms from the first +/// one, so a job the server had already finished still cost the caller that +/// much of `wait_for_completion` sleeping. +/// +/// The stub reports `compressing` three times and then `completed`, so the +/// client sleeps three times. Under the old schedule that was 600 ms; under +/// the backoff it is 10 + 20 + 40 = 70 ms. The assertion has a wide margin on +/// purpose, since a loaded CI runner is not a stopwatch, but 400 ms is still +/// far below what a fixed 200 ms interval could achieve here, so a revert +/// fails this rather than merely slowing it down. +#[test] +fn a_job_that_finishes_quickly_is_not_made_to_wait_out_the_ceiling() { + let polls = Arc::new(Mutex::new(0usize)); + let counter = Arc::clone(&polls); + let archive = vec![b'Z'; 32]; + let served = archive.clone(); + + let server = raw_server(move |path, _body, out| { + if path.ends_with("/download") { + respond_with(out, served.len(), &served, "application/zip"); + return; + } + if path.starts_with("/jobs/") { + let mut seen = counter.lock().unwrap(); + *seen += 1; + // The first three say "still working"; the fourth settles it. + let body = if *seen <= 3 { + COMPRESSING_JOB + } else { + FINISHED_JOB + }; + respond_with(out, body.len(), body.as_bytes(), "application/json"); + return; + } + // POST /compress + respond_with( + out, + FINISHED_JOB.len(), + FINISHED_JOB.as_bytes(), + "application/json", + ); + }); + + let dir = tempfile::tempdir().unwrap(); + let source = dir.path().join("notes.txt"); + std::fs::write(&source, b"tiny").unwrap(); + + let started = std::time::Instant::now(); + let delivered = + compress_path(&server, &source, Algorithm::Zip, 3).expect("the stub finishes the job"); + let elapsed = started.elapsed(); + + assert_eq!(delivered, archive); + // It really did poll four times, so the timing below is measuring the + // schedule and not a stub that answered "done" straight away. + assert!( + *polls.lock().unwrap() >= 4, + "the stub was not polled as expected: {:?}", + polls.lock().unwrap() + ); + assert!( + elapsed >= std::time::Duration::from_millis(70), + "it cannot have slept the schedule in {elapsed:?}" + ); + assert!( + elapsed < std::time::Duration::from_millis(400), + "three waits took {elapsed:?}; the fixed 200 ms interval is back" + ); +} diff --git a/apps/remote/tests/waiting.rs b/apps/remote/tests/waiting.rs new file mode 100644 index 0000000..a12c121 --- /dev/null +++ b/apps/remote/tests/waiting.rs @@ -0,0 +1,293 @@ +//! The poll loop, driven through its injected collaborators. +//! +//! Every case here runs in microseconds and spends no real time, which is the +//! whole reason [`Sleeper`] and [`Poller`] exist. The fake sleeper accumulates +//! virtual time instead of spending it, and the fake job reads that +//! accumulation to answer "am I done yet". Between them they express the thing +//! a stub HTTP server cannot express deterministically: a job that takes +//! **exactly** N milliseconds. +//! +//! Assertions are on the [`Waited`] the loop returns, not on counters inside +//! the fakes. A test that reaches into a fake is testing the fake, and passes +//! when the fake is wrong. + +use std::cell::{Cell, RefCell}; +use std::time::Duration; + +use collapse_remote::protocol::Progress; +use collapse_remote::waiting::{ + next_delay, wait_for, Poller, Sleeper, Waited, FIRST_POLL_DELAY, MAX_POLL_DELAY, +}; +use collapse_remote::RemoteError; + +const MS: fn(u64) -> Duration = Duration::from_millis; + +// ------------------------------------------------------------------ fakes -- + +/// A sleeper that does not sleep: it records the wait and moves virtual time +/// forward by it. +#[derive(Default)] +struct VirtualClock { + elapsed: Cell, + waits: RefCell>, +} + +impl VirtualClock { + fn elapsed(&self) -> Duration { + self.elapsed.get() + } + fn waits(&self) -> Vec { + self.waits.borrow().clone() + } +} + +impl Sleeper for VirtualClock { + fn sleep(&self, delay: Duration) { + self.elapsed.set(self.elapsed.get() + delay); + self.waits.borrow_mut().push(delay); + } +} + +/// A job that finishes once the caller has waited `takes`. +/// +/// This is the mock the issue is really about: compression that takes a known +/// amount of time, with none of it actually elapsing. +struct JobTaking<'a> { + clock: &'a VirtualClock, + takes: Duration, +} + +impl Poller for JobTaking<'_> { + fn poll(&self) -> Result { + Ok(if self.clock.elapsed() >= self.takes { + Progress::Ready + } else { + Progress::Waiting + }) + } +} + +/// A server that answers, then breaks. +struct FailsAfter { + remaining: Cell, +} + +impl Poller for FailsAfter { + fn poll(&self) -> Result { + if self.remaining.get() == 0 { + return Err(RemoteError::Malformed( + "the server stopped making sense".into(), + )); + } + self.remaining.set(self.remaining.get() - 1); + Ok(Progress::Waiting) + } +} + +/// Run a job of a known length and report what the loop did. +fn run(takes: Duration) -> (Waited, Vec) { + let clock = VirtualClock::default(); + let job = JobTaking { + clock: &clock, + takes, + }; + let account = wait_for(&clock, &job).expect("the job finishes"); + (account, clock.waits()) +} + +/// What the old flat schedule would have spent on the same job: poll, and if +/// it is not ready, sleep a whole ceiling. +fn under_the_old_schedule(takes: Duration) -> Duration { + let mut slept = Duration::ZERO; + while slept < takes { + slept += MAX_POLL_DELAY; + } + slept +} + +// ------------------------------------------------------------ issue #48 -- + +/// The case the issue is about, stated exactly. +/// +/// A job that finishes almost immediately, but not before the client's first +/// question. Under the old schedule the caller then slept a full 200 ms; it now +/// sleeps 10 and asks again. Same two polls either way: the difference is +/// entirely the wait, which is why the fix is a schedule and not a protocol +/// change. +#[test] +fn a_job_that_finishes_just_after_the_first_question_waits_ten_milliseconds() { + let (account, waits) = run(MS(1)); + + assert_eq!(account.polls, 2); + assert_eq!(account.slept, MS(10)); + assert_eq!(waits, vec![MS(10)]); + assert_eq!( + under_the_old_schedule(MS(1)), + MS(200), + "the old schedule this replaces" + ); +} + +/// A job already finished when the first question arrives never sleeps at all. +/// True before and after; pinned so a future schedule cannot introduce a wait +/// before the first poll. +#[test] +fn a_job_already_finished_is_never_slept_on() { + let (account, waits) = run(Duration::ZERO); + assert_eq!( + account, + Waited { + polls: 1, + slept: Duration::ZERO + } + ); + assert!(waits.is_empty(), "it waited for a finished job: {waits:?}"); +} + +// -------------------------------------------------------------- the ramp -- + +/// The exact sequence, which is the part a reader of the constant cannot see. +#[test] +fn the_wait_doubles_from_ten_to_the_ceiling_and_then_holds() { + let (_, waits) = run(MS(2_000)); + + assert_eq!( + &waits[..8], + &[ + MS(10), + MS(20), + MS(40), + MS(80), + MS(160), + MS(200), + MS(200), + MS(200) + ], + "got {waits:?}" + ); + assert!( + waits.iter().all(|w| *w <= MAX_POLL_DELAY), + "past the ceiling: {waits:?}" + ); + assert!( + waits.windows(2).all(|p| p[1] >= p[0]), + "not monotonic: {waits:?}" + ); +} + +/// A job of any real length pays almost nothing for the ramp: it reaches the +/// ceiling in 310 ms and five polls, and after that behaves exactly as before. +#[test] +fn a_long_job_costs_only_the_ramp() { + let (account, waits) = run(MS(10_000)); + + let ramp: Duration = waits.iter().take_while(|w| **w < MAX_POLL_DELAY).sum(); + assert_eq!(ramp, MS(310), "the ramp changed length"); + + // Against the old schedule over the same job: a handful of extra requests + // on a job that ran for ten seconds. + let old_polls = + (under_the_old_schedule(MS(10_000)).as_millis() / MAX_POLL_DELAY.as_millis()) as u32 + 1; + assert!( + account.polls <= old_polls + 4, + "{} polls against the old {old_polls}", + account.polls + ); +} + +// ------------------------------------------------- honest about the cost -- + +/// The backoff is **not** uniformly faster, and this pins how much slower it +/// can be. +/// +/// A job that finishes just after the ramp is asked again a whole ceiling +/// later, where the flat schedule might have caught it sooner: at 199 ms the +/// old schedule finished at 200 and this one finishes at 310. The band is +/// narrow and bounded by one ceiling, and that bound is the promise worth +/// keeping. +#[test] +fn the_backoff_is_never_worse_by_more_than_one_ceiling() { + let mut worst = Duration::ZERO; + let mut worst_at = Duration::ZERO; + for ms in (0..2_000).step_by(7) { + let takes = MS(ms); + let new = run(takes).0.slept; + let old = under_the_old_schedule(takes); + if new > old && new - old > worst { + worst = new - old; + worst_at = takes; + } + } + assert!( + worst < MAX_POLL_DELAY, + "a job taking {worst_at:?} is {worst:?} slower, past the one-ceiling bound" + ); + assert!( + worst > Duration::ZERO, + "if nothing is ever slower this test has stopped measuring anything" + ); +} + +/// And what it buys: every job that finishes within the first four waits, which +/// is the common case, is strictly faster than the flat schedule was. +#[test] +fn everything_that_finishes_inside_the_ramp_is_faster_than_before() { + for ms in 1..=150 { + let takes = MS(ms); + let new = run(takes).0.slept; + let old = under_the_old_schedule(takes); + assert!( + new < old, + "a job taking {takes:?} waited {new:?}, no better than the old {old:?}" + ); + } +} + +// ------------------------------------------------------------ the account -- + +/// The count includes the poll that settled it, so it is the number of requests +/// actually issued and not the number of waits. +#[test] +fn the_account_counts_the_question_that_settled_it() { + let (account, waits) = run(MS(1)); + assert_eq!(account.polls as usize, waits.len() + 1); +} + +// ------------------------------------------------------------- giving up -- + +/// A poller that fails stops the loop rather than retrying forever. +#[test] +fn an_error_from_the_server_ends_the_wait() { + let clock = VirtualClock::default(); + let poller = FailsAfter { + remaining: Cell::new(3), + }; + let outcome = wait_for(&clock, &poller); + assert!(outcome.is_err(), "got {outcome:?}"); + // It did back off between the three answers it did get. + assert_eq!(clock.waits(), vec![MS(10), MS(20), MS(40)]); +} + +// ---------------------------------------------------- the schedule itself -- + +/// `next_delay` is public, so it must be total: no overflow panic on a nonsense +/// input. +#[test] +fn the_schedule_saturates_instead_of_overflowing() { + assert_eq!(next_delay(Duration::MAX), MAX_POLL_DELAY); + assert_eq!(next_delay(Duration::ZERO), Duration::ZERO); + assert_eq!(next_delay(FIRST_POLL_DELAY), MS(20)); +} + +/// The first wait has to stay in the band that makes the whole change worth +/// having: short enough that a finished job is not waiting, long enough that a +/// busy server is not hammered. +#[test] +fn the_first_wait_stays_in_its_band() { + assert!(FIRST_POLL_DELAY >= MS(5), "too close to a spin"); + assert!( + FIRST_POLL_DELAY <= MS(25), + "a finished job would still be waiting" + ); + assert!(FIRST_POLL_DELAY < MAX_POLL_DELAY); +}