Back off from a short first wait instead of a flat 200 ms

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<Progress, RemoteError> {
+        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<Vec<u8>, RemoteError> {
     let response = ureq::get(&format!("{base}/jobs/{job_id}/download"))

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;

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<Progress, RemoteError>;
+}
+
+/// 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<Waited, RemoteError> {
+    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);
+            }
+        }
+    }
+}

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"
+    );
+}