diff --git a/README.md b/README.md index 0b5f28d..5ad1d0d 100644 --- a/README.md +++ b/README.md @@ -188,8 +188,8 @@ Requires **Rust 1.88+** (2021 edition). ```bash make build # build the Rust crates -make test # run every suite (585 Rust tests + 113 Vitest cases) -make test/rust # only the Rust tests that need no Node toolchain (470) +make test # run every suite (600 Rust tests + 113 Vitest cases) +make test/rust # only the Rust tests that need no Node toolchain (485) ``` `make test` includes the desktop app's own Rust suite, which compiles Tauri, so diff --git a/apps/remote/src/client.rs b/apps/remote/src/client.rs index 2f7c746..6da144d 100644 --- a/apps/remote/src/client.rs +++ b/apps/remote/src/client.rs @@ -6,6 +6,7 @@ use std::io::Read; use std::path::Path; +use std::time::Duration; use collapse_core::compression::compress_tar_dir; use collapse_core::Algorithm; @@ -14,6 +15,61 @@ use crate::protocol::{self, Progress}; use crate::waiting::{self, Poller}; use crate::RemoteError; +/// How long to wait on each stage of **one exchange** with the server. +/// +/// These bound the server's *responsiveness*, never the job. That distinction +/// is the whole design: compression can legitimately run for minutes or hours, +/// and nothing here shortens it. +/// +/// * a status poll is a tiny request the server answers at once, so silence on +/// a live socket for [`Timeouts::read`] means the far side is gone, not busy; +/// * `read` and `write` are **per socket operation**, not per response, which +/// was verified rather than assumed: a response dribbled out over 2.7 s in +/// fast chunks passes a 1 s read timeout untouched. A large upload or +/// download that keeps moving therefore never trips one. +/// +/// The values are deliberately generous. Hitting one has to be unambiguous +/// evidence that something is wrong, not evidence that we were impatient. +#[derive(Debug, Clone, Copy)] +pub struct Timeouts { + /// Getting a socket open at all. + pub connect: Duration, + /// Waiting on any single read from an open socket. + pub read: Duration, + /// Waiting on any single write to an open socket. + pub write: Duration, +} + +/// Long enough that a loaded server or a slow link is not mistaken for a dead +/// one. +pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +/// Silence this long, on a socket that is open, is not a busy server. +pub const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(30); +/// A peer that has stopped draining what we send is in the same state. +pub const DEFAULT_WRITE_TIMEOUT: Duration = Duration::from_secs(30); + +impl Default for Timeouts { + fn default() -> Self { + Self { + connect: DEFAULT_CONNECT_TIMEOUT, + read: DEFAULT_READ_TIMEOUT, + write: DEFAULT_WRITE_TIMEOUT, + } + } +} + +impl Timeouts { + /// One agent for the whole exchange, so every request carries the same + /// limits. Building it per call would also throw away the connection pool. + fn agent(&self) -> ureq::Agent { + ureq::AgentBuilder::new() + .timeout_connect(self.connect) + .timeout_read(self.read) + .timeout_write(self.write) + .build() + } +} + /// Compress a file or a whole directory on a remote server and return the /// archive bytes. /// @@ -35,6 +91,21 @@ pub fn compress_path( source: &Path, algorithm: Algorithm, level: u32, +) -> Result, RemoteError> { + compress_path_with(server, source, algorithm, level, Timeouts::default()) +} + +/// [`compress_path`] with the exchange limits spelled out. +/// +/// Exists for the same reason `collapse_core::extract_with` does: the default +/// is right for every front-end, and a test cannot afford to wait 30 seconds +/// to prove what happens after 30 seconds of silence. +pub fn compress_path_with( + server: &str, + source: &Path, + algorithm: Algorithm, + level: u32, + timeouts: Timeouts, ) -> Result, RemoteError> { // First, ahead of even looking at the source: a blank address is the // thing that is wrong, and packing a whole directory into a tar envelope @@ -55,7 +126,10 @@ pub fn compress_path( (std::fs::read(source)?, "none") }; - upload_and_collect(base, &name, algorithm, level, envelope, &data) + let agent = timeouts.agent(); + upload_and_collect( + &agent, timeouts, base, &name, algorithm, level, envelope, &data, + ) } /// Pack a directory into a tar, on disk, and hand back its bytes. The archive @@ -75,7 +149,10 @@ fn pack_directory(source: &Path, name: &str) -> Result, RemoteError> { /// `base` is already normalized by [`protocol::base_url`], which is also what /// rejects an unusable address, so every caller has to go through it first. +#[allow(clippy::too_many_arguments)] fn upload_and_collect( + agent: &ureq::Agent, + timeouts: Timeouts, base: &str, name: &str, algorithm: Algorithm, @@ -83,15 +160,17 @@ fn upload_and_collect( envelope: &str, data: &[u8], ) -> Result, RemoteError> { - let job = create_job(base, name, algorithm, level, envelope, data)?; + let job = create_job( + agent, timeouts, base, name, algorithm, level, envelope, data, + )?; let job_id = protocol::job_id_of(&job)?.to_string(); - wait_for_completion(base, &job_id)?; - let archive = download(base, &job_id)?; + wait_for_completion(agent, timeouts, base, &job_id)?; + let archive = download(agent, timeouts, base, &job_id)?; // Best-effort cleanup: the archive is already downloaded, so a failed // delete should not fail the operation. - let _ = ureq::delete(&format!("{base}/jobs/{job_id}")).call(); + let _ = agent.delete(&format!("{base}/jobs/{job_id}")).call(); Ok(archive) } @@ -101,16 +180,26 @@ fn upload_and_collect( /// Used before adding a server to a UI's list, so a typo shows up there /// instead of at the end of an upload. pub fn check_health(server: &str) -> Result<(), RemoteError> { + check_health_with(server, Timeouts::default()) +} + +/// [`check_health`] with the exchange limits spelled out. +pub fn check_health_with(server: &str, timeouts: Timeouts) -> Result<(), RemoteError> { let base = protocol::base_url(server)?; - let response = ureq::get(&format!("{base}/health")) + let response = timeouts + .agent() + .get(&format!("{base}/health")) .call() - .map_err(|e| remote_error(base, e))?; + .map_err(|e| remote_error(base, timeouts, e))?; protocol::healthy(&parse_json(response)?) } /// `POST /compress`: send the bytes, get the queued job back (202). +#[allow(clippy::too_many_arguments)] fn create_job( + agent: &ureq::Agent, + timeouts: Timeouts, base: &str, name: &str, algorithm: Algorithm, @@ -118,28 +207,33 @@ fn create_job( envelope: &str, data: &[u8], ) -> Result { - let response = ureq::post(&format!("{base}/compress")) + let response = agent + .post(&format!("{base}/compress")) .query("name", name) .query("algorithm", algorithm.extension()) .query("level", &level.to_string()) .query("envelope", envelope) .send_bytes(data) - .map_err(|e| remote_error(base, e))?; + .map_err(|e| remote_error(base, timeouts, e))?; parse_json(response) } /// 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> { + agent: &'a ureq::Agent, + timeouts: Timeouts, base: &'a str, job_id: &'a str, } impl Poller for HttpPoller<'_> { fn poll(&self) -> Result { - let response = ureq::get(&format!("{}/jobs/{}", self.base, self.job_id)) + let response = self + .agent + .get(&format!("{}/jobs/{}", self.base, self.job_id)) .call() - .map_err(|e| remote_error(self.base, e))?; + .map_err(|e| remote_error(self.base, self.timeouts, e))?; protocol::progress_of(&parse_json(response)?) } } @@ -150,16 +244,32 @@ impl Poller for HttpPoller<'_> { /// [`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 }; +fn wait_for_completion( + agent: &ureq::Agent, + timeouts: Timeouts, + base: &str, + job_id: &str, +) -> Result<(), RemoteError> { + let poller = HttpPoller { + agent, + timeouts, + 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")) +fn download( + agent: &ureq::Agent, + timeouts: Timeouts, + base: &str, + job_id: &str, +) -> Result, RemoteError> { + let response = agent + .get(&format!("{base}/jobs/{job_id}/download")) .call() - .map_err(|e| remote_error(base, e))?; + .map_err(|e| remote_error(base, timeouts, e))?; let mut archive = Vec::new(); response.into_reader().read_to_end(&mut archive)?; Ok(archive) @@ -175,7 +285,7 @@ fn parse_json(response: ureq::Response) -> Result RemoteError { +fn remote_error(server: &str, timeouts: Timeouts, err: ureq::Error) -> RemoteError { match err { ureq::Error::Status(status, response) => { let body = response.into_string().unwrap_or_default(); @@ -184,6 +294,19 @@ fn remote_error(server: &str, err: ureq::Error) -> RemoteError { message: protocol::rejection_message(status, &body), } } + // "I could not get a socket open" and "I had one and the far side went + // quiet" are different diagnoses and want different sentences. ureq + // separates them for us: a refused connection and a connect timeout are + // both `ConnectionFailed`, an unresolvable name is `Dns`, and anything + // that goes wrong on an established socket, a read timeout included, is + // `Io`. Checked against the real crate rather than assumed. + ureq::Error::Transport(transport) if transport.kind() == ureq::ErrorKind::Io => { + RemoteError::Unresponsive { + server: server.to_string(), + after: timeouts.read, + reason: transport.to_string(), + } + } other => RemoteError::Unreachable { server: server.to_string(), reason: other.to_string(), diff --git a/apps/remote/src/error.rs b/apps/remote/src/error.rs index dc0a902..ecf6a63 100644 --- a/apps/remote/src/error.rs +++ b/apps/remote/src/error.rs @@ -1,3 +1,5 @@ +use std::time::Duration; + use thiserror::Error; /// What can go wrong talking to a remote Collapse server. @@ -19,6 +21,29 @@ pub enum RemoteError { #[error("cannot reach the server at {server}: {reason}")] Unreachable { server: String, reason: String }, + /// A socket was open and the far side stopped answering on it. + /// + /// Kept apart from [`Self::Unreachable`] because the two call for different + /// things from whoever reads it: an address that cannot be reached is + /// usually wrong or the server is down, while a server that accepted the + /// connection and then went silent is running and stuck. The job may well + /// still exist on the far side. + /// + /// The limit this reports is on the **server's answers**, never on the + /// compression: a job is free to run for hours as long as each poll is + /// answered (see `Timeouts`). + #[error( + "the server at {server} accepted the connection and then stopped answering \ + (nothing for {} seconds). The job may still be running there; the archive was not \ + downloaded. Underlying error: {reason}", + after.as_secs() + )] + Unresponsive { + server: String, + after: Duration, + reason: String, + }, + /// The server answered with a 4xx/5xx. `message` is already rendered for /// a human (it prefers the server's JSON `detail` field). #[error("{message}")] diff --git a/apps/remote/src/lib.rs b/apps/remote/src/lib.rs index ec69e1b..0a84e92 100644 --- a/apps/remote/src/lib.rs +++ b/apps/remote/src/lib.rs @@ -22,5 +22,8 @@ mod error; pub mod protocol; pub mod waiting; -pub use client::{check_health, compress_path}; +pub use client::{ + check_health, check_health_with, compress_path, compress_path_with, Timeouts, + DEFAULT_CONNECT_TIMEOUT, DEFAULT_READ_TIMEOUT, DEFAULT_WRITE_TIMEOUT, +}; pub use error::RemoteError; diff --git a/apps/remote/tests/client.rs b/apps/remote/tests/client.rs index 77db9c1..80fc290 100644 --- a/apps/remote/tests/client.rs +++ b/apps/remote/tests/client.rs @@ -9,9 +9,12 @@ use std::path::Path; use std::sync::{Arc, Mutex}; +use std::time::Duration; use collapse_core::Algorithm; -use collapse_remote::{check_health, compress_path, RemoteError}; +use collapse_remote::{ + check_health, check_health_with, compress_path, compress_path_with, RemoteError, Timeouts, +}; /// Serve something on an ephemeral port for the rest of the test process. /// @@ -593,3 +596,196 @@ fn a_job_that_finishes_quickly_is_not_made_to_wait_out_the_ceiling() { "three waits took {elapsed:?}; the fixed 200 ms interval is back" ); } + +// ------------------------------------------- a server that stops answering -- + +/// Test-sized limits. The real ones are tens of seconds, deliberately, so that +/// hitting one is unambiguous evidence of a hang rather than of impatience +/// (see `Timeouts`); a suite cannot wait that long to prove it. +/// +/// **Two helpers, because the cases pull in opposite directions**, and getting +/// that wrong is what made the first version of these tests flaky. Where the +/// timeout is the thing being proved, it has to be short. Where the stub has to +/// answer *inside* it, it has to be long enough to survive a loaded CI runner +/// stalling a thread: `thread::sleep` promises a floor, not a ceiling, and a +/// shared runner will happily turn 60 ms into 250 ms. A margin that looked +/// generous on a quiet laptop (3x) was not, and macOS CI caught it. +fn firing_timeouts() -> Timeouts { + Timeouts { + connect: Duration::from_millis(500), + read: Duration::from_millis(200), + write: Duration::from_millis(500), + } +} + +/// For the cases where the stub answers promptly and the point is that nothing +/// fires. The margin over each nominal gap is at least 20x. +fn patient_timeouts() -> Timeouts { + Timeouts { + connect: Duration::from_secs(2), + read: Duration::from_millis(1_000), + write: Duration::from_secs(2), + } +} + +/// Issue #71: the client used to wait on a silent server for as long as the +/// process lived. Measured against v0.7.0 it was still polling after 60 s, +/// having printed nothing and written nothing, and had to be killed. +/// +/// A socket that opens and then goes quiet is now bounded. +#[test] +fn a_server_that_goes_quiet_is_given_up_on_rather_than_waited_on_forever() { + let server = raw_server(|_path, _body, _out| { + // Accept, read the request, answer nothing, and hold the socket open. + // Dropping it would close the connection, which is a different failure. + std::thread::sleep(Duration::from_secs(5)); + }); + + let dir = tempfile::tempdir().unwrap(); + let source = dir.path().join("notes.txt"); + std::fs::write(&source, b"upload me").unwrap(); + + let started = std::time::Instant::now(); + let error = compress_path_with(&server, &source, Algorithm::Zip, 3, firing_timeouts()) + .expect_err("a server that never answers must not be waited on"); + let elapsed = started.elapsed(); + + assert!( + matches!(error, RemoteError::Unresponsive { .. }), + "got {error:?}" + ); + assert!( + elapsed < Duration::from_secs(2), + "it gave up, but only after {elapsed:?}" + ); + + // The message has to be unmistakable: it is the only thing the user gets. + let message = error.to_string(); + assert!(message.contains("stopped answering"), "{message}"); + assert!(message.contains("may still be running"), "{message}"); +} + +/// Going quiet and never being reachable are different diagnoses, and the +/// error says which. One usually means a wrong address or a server that is +/// down; the other means a server that is up and stuck. +#[test] +fn an_unreachable_address_is_not_reported_as_an_unresponsive_server() { + let error = + check_health_with(UNREACHABLE, firing_timeouts()).expect_err("nothing is listening there"); + assert!( + matches!(error, RemoteError::Unreachable { .. }), + "got {error:?}" + ); +} + +/// **The limit is on the server's answers, not on the job.** +/// +/// This is the property that makes the whole change safe, and the one most +/// easily broken by a later "simplification" that puts a deadline on the loop +/// instead. The stub answers every poll at once, but reports `compressing` +/// eight times, so the client legitimately waits far longer than any single +/// timeout allows. Compression is allowed to take as long as it takes. +#[test] +fn a_long_job_is_never_cut_short_while_the_server_keeps_answering() { + let polls = Arc::new(Mutex::new(0usize)); + let counter = Arc::clone(&polls); + let archive = vec![b'Z'; 64]; + 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; + } + let body = if path.starts_with("/jobs/") { + let mut seen = counter.lock().unwrap(); + *seen += 1; + if *seen <= 12 { + COMPRESSING_JOB + } else { + FINISHED_JOB + } + } else { + FINISHED_JOB + }; + respond_with(out, body.len(), body.as_bytes(), "application/json"); + }); + + let dir = tempfile::tempdir().unwrap(); + let source = dir.path().join("notes.txt"); + std::fs::write(&source, b"a job that takes a while").unwrap(); + + let started = std::time::Instant::now(); + let delivered = compress_path_with(&server, &source, Algorithm::Zip, 3, patient_timeouts()) + .expect("a slow job that keeps answering must not be aborted"); + let elapsed = started.elapsed(); + + assert_eq!(delivered, archive); + // Twelve `compressing` answers is 10 + 20 + 40 + 80 + 160 + 200 * 7 = + // 1710 ms of waiting, comfortably past the 1000 ms read timeout. If the + // limit ever starts bounding the job instead of the answers, this cannot + // pass. Sleeps only ever overrun, so the comparison is safe in the one + // direction a loaded runner can move it. + assert!( + elapsed > patient_timeouts().read, + "the job lasted {elapsed:?}, which does not outlive the read timeout, \ + so this proves nothing" + ); + assert!(*polls.lock().unwrap() >= 13); +} + +/// A response that arrives slowly but steadily is not a hang. +/// +/// `read` is a per socket operation limit, not a per response one. That is what +/// lets a 500 MB archive come back over a slow link without tripping anything, +/// and it was verified against ureq rather than assumed. +#[test] +fn a_body_delivered_in_slow_pieces_is_not_mistaken_for_silence() { + use std::io::Write; + + let chunks = 30; + let chunk = vec![b'A'; 64]; + let total = chunk.len() * chunks; + let served = chunk.clone(); + + let server = raw_server(move |path, _body, out| { + if !path.ends_with("/download") { + respond_with( + out, + FINISHED_JOB.len(), + FINISHED_JOB.as_bytes(), + "application/json", + ); + return; + } + let head = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/zip\r\nContent-Length: {total}\r\nConnection: close\r\n\r\n" + ); + let _ = out.write_all(head.as_bytes()); + for _ in 0..chunks { + let _ = out.write_all(&served); + let _ = out.flush(); + // 40 ms nominal against a 1000 ms read timeout: a 25x margin, so + // only a stall of most of a second breaks it. The first version of + // this used 60 ms against 200 ms and macOS CI stalled past it. + std::thread::sleep(Duration::from_millis(40)); + } + let _ = out.shutdown(std::net::Shutdown::Write); + }); + + let dir = tempfile::tempdir().unwrap(); + let source = dir.path().join("notes.txt"); + std::fs::write(&source, b"x").unwrap(); + + let started = std::time::Instant::now(); + let delivered = compress_path_with(&server, &source, Algorithm::Zip, 3, patient_timeouts()) + .expect("a steady trickle is not a hang"); + + assert_eq!(delivered.len(), total); + // 30 chunks 40 ms apart is ~1200 ms, past the 1000 ms read timeout, so a + // per-response limit would have fired and a per-read one does not. + assert!( + started.elapsed() > patient_timeouts().read, + "the download did not outlast the read timeout, so this proves nothing" + ); +} diff --git a/docs/architecture.md b/docs/architecture.md index 9f3b059..e96ed2e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -218,7 +218,45 @@ crosses that boundary fine. The split inside mirrors the one the rest of the project uses for testability: `protocol.rs` is **public and pure** (URL normalization, reading the server's JSON, and `progress_of`, which decides whether a status means keep polling, -download, or give up), while the HTTP plumbing in `client.rs` stays private. +download, or give up), `waiting.rs` is the poll loop behind two one-method +traits, while the HTTP plumbing in `client.rs` stays private. + +### Waiting, and the two limits that are not the same limit + +The client asks `GET /jobs/{id}` until the job settles. Two separate decisions +live there and conflating them is the mistake to avoid. + +**How often to ask** is the backoff in `waiting.rs`: 10 ms, doubling to a +200 ms ceiling. It used to be a flat 200 ms from the first wait, so a job the +server had already finished still cost the caller 200 ms of sleeping (a five +byte file measured 236 ms against 44 ms now). A long job is unaffected: it +reaches the ceiling after 310 ms and stays there. It is not uniformly faster, +and that is pinned rather than glossed over: a job finishing just after the +ramp is asked again a whole ceiling later, worst case 110 ms, bounded by one +`MAX_POLL_DELAY`. + +**When to give up** is `Timeouts`, and it bounds the *server's answers*, never +the job. Compression is free to run for hours; what is not allowed is silence +on an open socket. `connect` is 10 s, `read` and `write` 30 s, and they are +deliberately generous so that hitting one is unambiguous evidence of a hang +rather than of impatience. `read` and `write` are **per socket operation**, not +per response, which was verified against ureq rather than assumed: a body +dribbled out over 2.7 s in fast chunks passes a 1 s read timeout untouched, so +a large upload or download that keeps moving never trips one. There is +deliberately **no total deadline on the job**, and a test asserts a job cannot +be cut short while the server keeps answering. + +That distinction is also what `RemoteError` now spells out. `Unreachable` means +no socket could be opened (refused, connect timeout, DNS); `Unresponsive` means +one was open and the far side went quiet, which is a server that is up and +stuck rather than a wrong address, and its message says the job may still be +running there. ureq separates the two for us (`ConnectionFailed`/`Dns` against +`Io`), checked against the real crate. + +`Timeouts` is injectable through `compress_path_with` and `check_health_with`, +following `collapse_core::extract_with`: the defaults are right for every +front-end, and a suite cannot wait 30 seconds to prove what happens after 30 +seconds of silence. Errors are a `RemoteError` of its own, so the crate does not depend on any front-end's error type; the CLI absorbs it into `CliError`.