From 9c40dafa83b1d231e6839c6ed7f12613e4ac7252 Mon Sep 17 00:00:00 2001 From: Javier Parada Date: Wed, 26 Aug 2026 15:00:21 +0200 Subject: [PATCH 1/2] remote: bound the server's answers, never the job `wait_for_completion` had no deadline, no maximum number of polls and no per-request timeout, so a server that accepted the connection and then said nothing kept the client waiting for as long as the process lived. Measured against v0.7.0 in the issue: still polling after 60 s, nothing printed, nothing written, killed by the harness. The limit added here is on the **server's responsiveness**, and deliberately not on the job. Compression is allowed to take as long as it takes; what is not allowed is silence on an open socket. There is no total deadline on the poll loop, and `a_long_job_is_never_cut_short_while_the_server_keeps_answering` fails if one is ever added: the mistake is easy to make and would cut off exactly the users who need the server most. `Timeouts` carries connect (10 s), read and write (30 s each) on one shared agent. The values are generous on purpose. Hitting one has to be unambiguous evidence that something is wrong rather than evidence that we were impatient, because the only thing the user gets is the message. Two properties were verified against ureq rather than assumed, and both matter: - `timeout_read` and `timeout_write` are **per socket operation**, not per response. 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. A test pins it. - ureq separates "no socket could be opened" (`ConnectionFailed` for both a refused connection and a connect timeout, `Dns` for a name that will not resolve) from "the socket was open and the exchange broke" (`Io`). That separation becomes `RemoteError::Unresponsive`, kept apart from `Unreachable` because the two want different things from whoever reads them: a wrong address or a server that is down, against a server that is up and stuck. Its message says the job may still be running on the far side and that the archive was not downloaded, which is the true and useful thing to say. `Timeouts` is injectable through `compress_path_with` and `check_health_with`, following `collapse_core::extract_with`: the defaults suit every front-end, and a suite cannot wait 30 seconds to prove what happens after 30 seconds of silence. Both existing entry points keep their signatures, so the CLI and the desktop are untouched. Four tests, each checked against a broken implementation: dropping the timeouts makes the silent-server case take 5 s and fail, reporting a quiet socket as merely unreachable fails the classification case, and a total deadline on the loop fails the long-job case. Closes #71 --- README.md | 4 +- apps/remote/src/client.rs | 157 ++++++++++++++++++++++++++++---- apps/remote/src/error.rs | 25 ++++++ apps/remote/src/lib.rs | 5 +- apps/remote/tests/client.rs | 174 +++++++++++++++++++++++++++++++++++- docs/architecture.md | 40 ++++++++- 6 files changed, 383 insertions(+), 22 deletions(-) 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..c20b7ed 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,172 @@ 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. +fn quick_timeouts() -> Timeouts { + Timeouts { + connect: Duration::from_millis(500), + read: Duration::from_millis(200), + write: Duration::from_millis(500), + } +} + +/// 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, quick_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, quick_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 <= 8 { + 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, quick_timeouts()) + .expect("a slow job that keeps answering must not be aborted"); + let elapsed = started.elapsed(); + + assert_eq!(delivered, archive); + // 10 + 20 + 40 + 80 + 160 + 200 + 200 + 200 = 910 ms of waiting, against a + // 200 ms read timeout. If the limit ever starts bounding the job instead of + // the answers, this cannot pass. + assert!( + elapsed > quick_timeouts().read * 3, + "the job only lasted {elapsed:?}; it is not outliving the timeout" + ); + assert!(*polls.lock().unwrap() >= 9); +} + +/// 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 = 8; + 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(); + // Comfortably inside the read timeout each time, and well past it + // in total. + std::thread::sleep(Duration::from_millis(60)); + } + 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, quick_timeouts()) + .expect("a steady trickle is not a hang"); + + assert_eq!(delivered.len(), total); + assert!( + started.elapsed() > quick_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`. From bec92fc4927666a4457bc28c490898592998cacc Mon Sep 17 00:00:00 2001 From: Javier Parada Date: Wed, 26 Aug 2026 15:10:00 +0200 Subject: [PATCH 2/2] tests: give the timing assertions a margin a loaded runner cannot eat `a_body_delivered_in_slow_pieces_is_not_mistaken_for_silence` failed on the macOS leg. Not the code: the test. It trickled a body in 60 ms pieces against a 200 ms read timeout, and `thread::sleep` promises a floor rather than a ceiling, so a shared runner turning one 60 ms gap into 250 ms was enough. A margin that looked generous on a quiet laptop was 3x. The same mistake was latent in two more cases, which passed only because they got lucky: they used one short timeout for everything, including the cases where the stub has to answer *inside* it. So there are two helpers now, and the split is the point. `firing_timeouts` keeps the short read for the cases where the timeout firing is the thing being proved. `patient_timeouts` gives a second, at least a 25x margin over every nominal gap, for the cases where the point is that nothing fires. The long-job case grows from eight `compressing` answers to twelve, so it still outlives the timeout it is measured against: 1710 ms of waiting against 1000 ms. The trickle case sends 30 pieces 40 ms apart, ~1200 ms in total, which is past the read timeout, so a per-response limit would still fire where a per-read one does not. The suite goes from 0.95 s to 1.77 s. Worth it: six consecutive local runs are identical, and all three mutations are still caught (no timeouts, a quiet socket misreported as unreachable, and a total deadline on the loop). This is real-socket behaviour, so unlike the poll schedule it cannot be moved onto a virtual clock. Margin is the only defence available. --- apps/remote/tests/client.rs | 58 ++++++++++++++++++++++++++----------- 1 file changed, 41 insertions(+), 17 deletions(-) diff --git a/apps/remote/tests/client.rs b/apps/remote/tests/client.rs index c20b7ed..80fc290 100644 --- a/apps/remote/tests/client.rs +++ b/apps/remote/tests/client.rs @@ -602,7 +602,15 @@ fn a_job_that_finishes_quickly_is_not_made_to_wait_out_the_ceiling() { /// 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. -fn quick_timeouts() -> Timeouts { +/// +/// **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), @@ -610,6 +618,16 @@ fn quick_timeouts() -> Timeouts { } } +/// 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. @@ -628,7 +646,7 @@ fn a_server_that_goes_quiet_is_given_up_on_rather_than_waited_on_forever() { std::fs::write(&source, b"upload me").unwrap(); let started = std::time::Instant::now(); - let error = compress_path_with(&server, &source, Algorithm::Zip, 3, quick_timeouts()) + 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(); @@ -653,7 +671,7 @@ fn a_server_that_goes_quiet_is_given_up_on_rather_than_waited_on_forever() { #[test] fn an_unreachable_address_is_not_reported_as_an_unresponsive_server() { let error = - check_health_with(UNREACHABLE, quick_timeouts()).expect_err("nothing is listening there"); + check_health_with(UNREACHABLE, firing_timeouts()).expect_err("nothing is listening there"); assert!( matches!(error, RemoteError::Unreachable { .. }), "got {error:?}" @@ -682,7 +700,7 @@ fn a_long_job_is_never_cut_short_while_the_server_keeps_answering() { let body = if path.starts_with("/jobs/") { let mut seen = counter.lock().unwrap(); *seen += 1; - if *seen <= 8 { + if *seen <= 12 { COMPRESSING_JOB } else { FINISHED_JOB @@ -698,19 +716,22 @@ fn a_long_job_is_never_cut_short_while_the_server_keeps_answering() { 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, quick_timeouts()) + 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); - // 10 + 20 + 40 + 80 + 160 + 200 + 200 + 200 = 910 ms of waiting, against a - // 200 ms read timeout. If the limit ever starts bounding the job instead of - // the answers, this cannot pass. + // 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 > quick_timeouts().read * 3, - "the job only lasted {elapsed:?}; it is not outliving the timeout" + elapsed > patient_timeouts().read, + "the job lasted {elapsed:?}, which does not outlive the read timeout, \ + so this proves nothing" ); - assert!(*polls.lock().unwrap() >= 9); + assert!(*polls.lock().unwrap() >= 13); } /// A response that arrives slowly but steadily is not a hang. @@ -722,7 +743,7 @@ fn a_long_job_is_never_cut_short_while_the_server_keeps_answering() { fn a_body_delivered_in_slow_pieces_is_not_mistaken_for_silence() { use std::io::Write; - let chunks = 8; + let chunks = 30; let chunk = vec![b'A'; 64]; let total = chunk.len() * chunks; let served = chunk.clone(); @@ -744,9 +765,10 @@ fn a_body_delivered_in_slow_pieces_is_not_mistaken_for_silence() { for _ in 0..chunks { let _ = out.write_all(&served); let _ = out.flush(); - // Comfortably inside the read timeout each time, and well past it - // in total. - std::thread::sleep(Duration::from_millis(60)); + // 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); }); @@ -756,12 +778,14 @@ fn a_body_delivered_in_slow_pieces_is_not_mistaken_for_silence() { std::fs::write(&source, b"x").unwrap(); let started = std::time::Instant::now(); - let delivered = compress_path_with(&server, &source, Algorithm::Zip, 3, quick_timeouts()) + 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() > quick_timeouts().read, + started.elapsed() > patient_timeouts().read, "the download did not outlast the read timeout, so this proves nothing" ); }