diff --git a/Cargo.lock b/Cargo.lock index 161f7a4..429f81b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -309,7 +309,7 @@ checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" [[package]] name = "collapse-cli" -version = "0.8.0" +version = "0.9.0" dependencies = [ "axum", "clap", @@ -324,7 +324,7 @@ dependencies = [ [[package]] name = "collapse-core" -version = "0.8.0" +version = "0.9.0" dependencies = [ "same-file", "serde", @@ -338,7 +338,7 @@ dependencies = [ [[package]] name = "collapse-remote" -version = "0.8.0" +version = "0.9.0" dependencies = [ "axum", "collapse-core", @@ -353,7 +353,7 @@ dependencies = [ [[package]] name = "collapse-server-backend" -version = "0.8.0" +version = "0.9.0" dependencies = [ "axum", "clap", diff --git a/README.md b/README.md index 0b5f28d..81b82d1 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 (604 Rust tests + 113 Vitest cases) +make test/rust # only the Rust tests that need no Node toolchain (489) ``` `make test` includes the desktop app's own Rust suite, which compiles Tauri, so diff --git a/apps/cli/Cargo.toml b/apps/cli/Cargo.toml index 991ed17..c16f12c 100644 --- a/apps/cli/Cargo.toml +++ b/apps/cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "collapse-cli" -version = "0.8.0" +version = "0.9.0" edition = "2021" license = "GPL-3.0-only" diff --git a/apps/core/Cargo.toml b/apps/core/Cargo.toml index 6463d9d..5f90cba 100644 --- a/apps/core/Cargo.toml +++ b/apps/core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "collapse-core" -version = "0.8.0" +version = "0.9.0" edition = "2021" license = "GPL-3.0-only" diff --git a/apps/core/src/compression.rs b/apps/core/src/compression.rs index 8d829a1..13da057 100644 --- a/apps/core/src/compression.rs +++ b/apps/core/src/compression.rs @@ -398,6 +398,49 @@ fn algorithm_of(archive: &Path) -> Result { .ok_or_else(|| CompressionError::Failed(format!("Unknown archive extension: .{ext}"))) } +/// Say that the archive could not be read, in words, keeping the parser's +/// detail but not its debris. +/// +/// This mattered less while a failed listing was advisory. Now that it stops +/// the extraction, the message is the whole of what the user gets, and the tar +/// crate embeds the bytes it choked on, so a damaged header arrived as a wall +/// of replacement characters ("numeric field did not have utf-8 text: ???? +/// when getting cksum for ????...", with every ? an unprintable byte). +/// +/// Narrower than issue #66, which is about callers being able to tell causes +/// apart. This is only about what is fit to show a person. +fn unreadable_archive(error: CompressionError) -> CompressionError { + let detail = match &error { + // Unwrap rather than nest, or the message repeats its own prefix: + // "Compression failed: ... : Compression failed: ...". + CompressionError::Failed(message) => legible(message), + CompressionError::Io(io) => legible(&io.to_string()), + other => legible(&other.to_string()), + }; + CompressionError::Failed(format!( + "this archive could not be read, so nothing was extracted: {detail}" + )) +} + +/// Strip what a person cannot read out of a dependency's message, and cap it. +fn legible(detail: &str) -> String { + let cleaned: String = detail + .chars() + .map(|c| { + if c.is_control() || c == char::REPLACEMENT_CHARACTER { + ' ' + } else { + c + } + }) + .collect(); + let collapsed = cleaned.split_whitespace().collect::>().join(" "); + match collapsed.char_indices().nth(140) { + Some((cut, _)) => format!("{}...", &collapsed[..cut]), + None => collapsed, + } +} + /// The entry names an archive holds, without extracting anything. fn list_entries(archive: &Path, algorithm: Algorithm) -> Result, CompressionError> { match algorithm { @@ -446,8 +489,9 @@ pub fn extract(archive: &Path, output_dir: &Path) -> Result, Compres /// cannot write. /// /// Naming is settled over the whole listing before the first byte is written, -/// so the two answers nothing can recover from (a character with no -/// replacement, and two entries that would land on one name) leave the output +/// and a listing that cannot be read stops it there (issue #89), so the answers +/// nothing can recover from (a character with no replacement, two entries that +/// would land on one name, an archive too damaged to read) leave the output /// directory as they found it. pub fn extract_with( archive: &Path, @@ -469,10 +513,31 @@ pub fn extract_with( /// Work out what every entry will be called, from the listing. /// -/// A listing that cannot be read is deliberately **not** an error here: the -/// extractor is about to open the same archive and fail on it in its own -/// vocabulary, which is the message this layer would otherwise replace with a -/// worse one. Only a readable listing produces a plan. +/// **A listing that cannot be read stops the extraction here**, before anything +/// is written. +/// +/// It used to be swallowed, on the reasoning that the extractor was about to +/// open the same archive and fail on it in its own vocabulary, and that this +/// layer would only replace that message with a worse one. The reasoning was +/// right about the message and wrong about the timing (issue #89): the +/// extractor fails **while streaming**, so by the time it notices it has +/// already written every entry before the fault, and written them with no plan +/// at all, which means no rewriting, no refusal and no collision check. An +/// archive holding `notes.txt:hidden` was refused outright when its listing was +/// intact and written as an invisible NTFS stream when it was not: the harm of +/// issue #63, performed without the user ever being asked. One bad 512 byte +/// header appended to a tar was enough to arrange it. +/// +/// The message turned out to be the weaker half of the argument anyway. Both +/// passes go through the same parser, so in practice they say close to the same +/// thing; the reproduction had them differing only in where they were cut off. +/// +/// What this does cost is partial recovery: a truncated tar used to hand back +/// whatever preceded the damage and now hands back nothing. That was a +/// deliberate call. It is still reachable on purpose through the backends +/// themselves ([`self::tar::extract_tar`] and friends), which take no options +/// and so never come through here; `recovering_from_a_damaged_archive_is_still_ +/// possible_through_the_backend` pins that. /// /// It costs one listing per extraction, paid even when nothing needs renaming, /// because the only way to know that is to read the names. For zip and 7z that @@ -483,8 +548,6 @@ fn plan_for( algorithm: Algorithm, options: &ExtractOptions, ) -> Result { - let Ok(names) = list_entries(archive, algorithm) else { - return Ok(NamePlan::identity()); - }; + let names = list_entries(archive, algorithm).map_err(unreadable_archive)?; Ok(plan_names(&names, options.rules(), options.replacements())?) } diff --git a/apps/core/tests/names.rs b/apps/core/tests/names.rs index 7bfe740..48a9c1f 100644 --- a/apps/core/tests/names.rs +++ b/apps/core/tests/names.rs @@ -19,7 +19,9 @@ use collapse_core::compression::{ extract_7z, extract_tar, extract_zip, CharacterFault, NameError, NameProblem, NameReport, NameRules, Substitutions, }; -use collapse_core::{extract, extract_with, unwritable_names_with, ExtractOptions}; +use collapse_core::{ + extract, extract_with, unwritable_names_with, CompressionError, ExtractOptions, +}; use sevenz_rust2::{SevenZArchiveEntry, SevenZWriter}; use tar::{Builder, EntryType, Header}; use zip::write::SimpleFileOptions; @@ -810,10 +812,11 @@ fn the_backends_called_directly_still_write_the_archive_s_own_names() { } #[test] -fn an_archive_that_cannot_be_listed_still_fails_in_the_extractor_s_words() { - // The listing pass is advisory on purpose. A corrupt archive must produce - // the message extraction has always produced, not a second-hand one from a - // pass that only exists to plan names. +fn an_archive_that_cannot_be_listed_still_fails_in_the_parser_s_words() { + // The listing pass is no longer advisory (issue #89): it stops the + // extraction. The message is unaffected, because both passes go through the + // same parser, which is what made the old "keep going so the extractor can + // phrase it better" argument weaker than it looked. let dir = tempfile::TempDir::new().unwrap(); let archive = dir.path().join("truncated.zip"); fs::write(&archive, b"PK\x03\x04 and then nothing").unwrap(); @@ -973,3 +976,160 @@ fn a_unix_name_holding_a_backslash_survives_the_round_trip() { ); } } + +// ------------------------------------------ a listing that cannot be read --- + +/// A tar whose entries are sound and whose end-of-archive marker is not. +/// +/// This shape is the whole of issue #89: `list_tar_entries` walks every header +/// before extraction starts, while extraction writes as it walks, so the +/// listing dies on the damage after the extractor would already have written +/// everything before it. +fn tar_with_a_broken_tail(dir: &Path, entries: &[(&str, &[u8])]) -> PathBuf { + let mut builder = Builder::new(Vec::new()); + for (name, content) in entries { + let mut header = Header::new_gnu(); + header.set_size(content.len() as u64); + header.set_mode(0o644); + header.set_entry_type(EntryType::Regular); + let raw = name.as_bytes(); + header.as_old_mut().name[..raw.len()].copy_from_slice(raw); + header.set_cksum(); + builder.append(&header, *content).unwrap(); + } + let mut bytes = builder.into_inner().unwrap(); + // Replace the two trailing zero blocks with something that is not a header. + let tail = bytes.len() - 1024; + for byte in bytes[tail..].iter_mut() { + *byte = 0xAA; + } + let archive = dir.join("broken-tail.tar"); + fs::write(&archive, &bytes).unwrap(); + archive +} + +/// Issue #89. An unreadable listing used to turn the entire naming layer off +/// and let every entry before the damage be written under its raw name. +/// +/// On Windows `notes.txt:hidden` is not a file name: it is the `hidden` +/// alternate data stream of `notes.txt`, so the write succeeds, the bytes land +/// where no listing shows them, and the user was told there was nothing to +/// answer for. That is issue #63's harm performed without consent, and one bad +/// 512 byte header was enough to arrange it. +#[test] +fn a_damaged_archive_writes_nothing_rather_than_writing_raw_names() { + let dir = tempfile::TempDir::new().unwrap(); + let archive = tar_with_a_broken_tail( + dir.path(), + &[("notes.txt:hidden", b"payload"), ("second.txt", b"more")], + ); + let out = dir.path().join("out"); + + let options = ExtractOptions::new().with_rules(NameRules::windows()); + let err = extract_with(&archive, &out, &options).unwrap_err(); + + // The parser's own words, not a second-hand summary. + assert!(err.to_string().starts_with("Compression failed:"), "{err}"); + // And nothing on disk. Before the fix both entries were here, the first as + // an NTFS stream on a Windows host. + let written: Vec<_> = fs::read_dir(&out) + .map(|entries| { + entries + .filter_map(Result::ok) + .map(|e| e.file_name()) + .collect() + }) + .unwrap_or_default(); + assert!( + written.is_empty(), + "a damaged archive wrote {written:?} before failing" + ); +} + +/// The same archive with an intact tail is refused too, but for the reason the +/// user can act on. The pair is the point: an archive must not become *more* +/// permissive by being damaged. +#[test] +fn the_same_names_are_refused_whether_or_not_the_archive_is_damaged() { + let dir = tempfile::TempDir::new().unwrap(); + let entries: &[(&str, &[u8])] = &[("notes.txt:hidden", b"payload"), ("second.txt", b"more")]; + let options = ExtractOptions::new().with_rules(NameRules::windows()); + + let intact = archive_with(dir.path(), "tar", entries); + let out = dir.path().join("intact"); + let err = extract_with(&intact, &out, &options).unwrap_err(); + assert!( + matches!(err, CompressionError::Name(_)), + "an intact archive must name the character to answer for: {err}" + ); + assert!(!out.exists() || fs::read_dir(&out).unwrap().count() == 0); + + let damaged = tar_with_a_broken_tail(dir.path(), entries); + let out = dir.path().join("damaged"); + assert!(extract_with(&damaged, &out, &options).is_err()); + assert!(!out.exists() || fs::read_dir(&out).unwrap().count() == 0); +} + +/// Recovering what a damaged archive still holds is deliberately not gone, it +/// is just no longer what `extract` does by default. +/// +/// The backends take no options, so they never go through the planning pass. +/// That is the escape hatch, and it is worth knowing it exists: without this +/// test the capability would look like it had been deleted. +#[test] +fn recovering_from_a_damaged_archive_is_still_possible_through_the_backend() { + let dir = tempfile::TempDir::new().unwrap(); + let archive = + tar_with_a_broken_tail(dir.path(), &[("first.txt", b"one"), ("second.txt", b"two")]); + let out = dir.path().join("salvage"); + + // It still fails, on the damage, but only after handing back what preceded + // it. `extract` writes nothing at all for the same archive. + let _ = extract_tar(&archive, &out); + let salvaged: Vec<_> = fs::read_dir(&out) + .map(|entries| { + entries + .filter_map(Result::ok) + .map(|e| e.file_name()) + .collect() + }) + .unwrap_or_default(); + assert_eq!(salvaged.len(), 2, "the backend salvaged {salvaged:?}"); +} + +/// A refusal is now the whole of what the user gets for a damaged archive, so +/// the message has to be fit to read. +/// +/// The tar crate embeds the bytes it choked on, so this one used to arrive as +/// "numeric field did not have utf-8 text: when +/// getting cksum for ". That was survivable while the listing +/// pass was advisory and the files came out anyway; it is not now. +#[test] +fn a_damaged_archive_says_so_in_words_a_person_can_read() { + let dir = tempfile::TempDir::new().unwrap(); + let archive = tar_with_a_broken_tail(dir.path(), &[("a.txt", b"one")]); + + let err = extract(&archive, &dir.path().join("out")).unwrap_err(); + let message = err.to_string(); + + assert!( + message.contains("could not be read, so nothing was extracted"), + "it must lead with what happened: {message}" + ); + assert!( + !message.contains(char::REPLACEMENT_CHARACTER), + "the dependency's debris reached the user: {message}" + ); + assert!( + !message.chars().any(char::is_control), + "control characters reached the user: {message}" + ); + // The parser's own detail is kept, just cleaned up. + assert!(message.contains("cksum"), "the detail was lost: {message}"); + // And it stays short enough to read in a terminal. + assert!( + message.chars().count() < 240, + "{} chars", + message.chars().count() + ); +} diff --git a/apps/desktop/package-lock.json b/apps/desktop/package-lock.json index 55bee7b..e9b136f 100644 --- a/apps/desktop/package-lock.json +++ b/apps/desktop/package-lock.json @@ -1,12 +1,12 @@ { "name": "collapse-desktop", - "version": "0.8.0", + "version": "0.9.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "collapse-desktop", - "version": "0.8.0", + "version": "0.9.0", "dependencies": { "@tauri-apps/api": "^2", "@tauri-apps/plugin-dialog": "^2", diff --git a/apps/desktop/package.json b/apps/desktop/package.json index 0deb5d8..f8de4d8 100644 --- a/apps/desktop/package.json +++ b/apps/desktop/package.json @@ -1,7 +1,7 @@ { "name": "collapse-desktop", "private": true, - "version": "0.8.0", + "version": "0.9.0", "type": "module", "scripts": { "dev": "vite", diff --git a/apps/desktop/src-tauri/Cargo.lock b/apps/desktop/src-tauri/Cargo.lock index 273e5c1..a0b29d7 100644 --- a/apps/desktop/src-tauri/Cargo.lock +++ b/apps/desktop/src-tauri/Cargo.lock @@ -516,7 +516,7 @@ checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" [[package]] name = "collapse-core" -version = "0.8.0" +version = "0.9.0" dependencies = [ "same-file", "serde", @@ -528,7 +528,7 @@ dependencies = [ [[package]] name = "collapse-desktop" -version = "0.8.0" +version = "0.9.0" dependencies = [ "axum", "collapse-core", @@ -546,7 +546,7 @@ dependencies = [ [[package]] name = "collapse-remote" -version = "0.8.0" +version = "0.9.0" dependencies = [ "collapse-core", "serde_json", @@ -557,7 +557,7 @@ dependencies = [ [[package]] name = "collapse-server-backend" -version = "0.8.0" +version = "0.9.0" dependencies = [ "axum", "clap", diff --git a/apps/desktop/src-tauri/Cargo.toml b/apps/desktop/src-tauri/Cargo.toml index 2a17caf..35b6cab 100644 --- a/apps/desktop/src-tauri/Cargo.toml +++ b/apps/desktop/src-tauri/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "collapse-desktop" -version = "0.8.0" +version = "0.9.0" edition = "2021" description = "Collapse — a small, fast file compressor" authors = ["cervantic"] diff --git a/apps/desktop/src-tauri/tauri.conf.json b/apps/desktop/src-tauri/tauri.conf.json index d264ccb..a7c48b2 100644 --- a/apps/desktop/src-tauri/tauri.conf.json +++ b/apps/desktop/src-tauri/tauri.conf.json @@ -1,7 +1,7 @@ { "$schema": "https://schema.tauri.app/config/2", "productName": "Collapse", - "version": "0.8.0", + "version": "0.9.0", "identifier": "com.cervantic.collapse", "build": { "frontendDist": "../dist", diff --git a/apps/desktop/src-tauri/tests/commands.rs b/apps/desktop/src-tauri/tests/commands.rs index 217dcda..a423f78 100644 --- a/apps/desktop/src-tauri/tests/commands.rs +++ b/apps/desktop/src-tauri/tests/commands.rs @@ -1540,11 +1540,21 @@ fn a_truncated_archive_is_reported_legibly_instead_of_panicking() { ); } // The short read reaches core as its dependency's `Io` variant, so - // core unwraps it to `CompressionError::Io` (issue #66). It used to - // read `Compression failed: Io(Error { kind: UnexpectedEof, - // message: "failed to fill whole buffer" }, "")`. + // core unwraps it (issue #66). It used to read `Compression failed: + // Io(Error { kind: UnexpectedEof, message: "failed to fill whole + // buffer" }, "")`, and then plain `IO error: failed to fill whole + // buffer` once that was mapped properly. + // + // It now names the consequence first: a listing that cannot be read + // stops the extraction rather than letting it write raw names + // (issue #89), and that refusal is the whole of what the user gets, + // so it has to say what happened before it says why. "7z" => { - assert_eq!(err, "IO error: failed to fill whole buffer"); + assert_eq!( + err, + "Compression failed: this archive could not be read, so nothing was \ + extracted: failed to fill whole buffer" + ); let leftovers: Vec = fs::read_dir(&out_dir) .map(|entries| entries.map(|e| e.unwrap().path()).collect()) .unwrap_or_default(); diff --git a/apps/remote/Cargo.toml b/apps/remote/Cargo.toml index 157072d..62a0813 100644 --- a/apps/remote/Cargo.toml +++ b/apps/remote/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "collapse-remote" -version = "0.8.0" +version = "0.9.0" edition = "2021" license = "GPL-3.0-only" description = "Client for a remote Collapse compression server" diff --git a/apps/remote/src/client.rs b/apps/remote/src/client.rs index c932a99..6da144d 100644 --- a/apps/remote/src/client.rs +++ b/apps/remote/src/client.rs @@ -12,10 +12,63 @@ 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); +/// 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. @@ -38,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 @@ -58,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 @@ -78,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, @@ -86,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) } @@ -104,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, @@ -121,35 +207,69 @@ 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) } -/// 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> { + agent: &'a ureq::Agent, + timeouts: Timeouts, + 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 = self + .agent + .get(&format!("{}/jobs/{}", self.base, self.job_id)) + .call() + .map_err(|e| remote_error(self.base, self.timeouts, 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( + 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) @@ -165,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(); @@ -174,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 18ed51e..0a84e92 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,10 @@ mod client; 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/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..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. /// @@ -518,3 +521,271 @@ 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" + ); +} + +// ------------------------------------------- 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/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); +} diff --git a/apps/server-backend/Cargo.toml b/apps/server-backend/Cargo.toml index cf1acfb..0a6cafd 100644 --- a/apps/server-backend/Cargo.toml +++ b/apps/server-backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "collapse-server-backend" -version = "0.8.0" +version = "0.9.0" edition = "2021" license = "GPL-3.0-only" diff --git a/apps/server-frontend/package-lock.json b/apps/server-frontend/package-lock.json index bfd3d6e..add4f50 100644 --- a/apps/server-frontend/package-lock.json +++ b/apps/server-frontend/package-lock.json @@ -1,12 +1,12 @@ { "name": "collapse-server-frontend", - "version": "0.8.0", + "version": "0.9.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "collapse-server-frontend", - "version": "0.8.0", + "version": "0.9.0", "dependencies": { "vue": "^3.5.30" }, diff --git a/apps/server-frontend/package.json b/apps/server-frontend/package.json index a4df019..c9fbf1b 100644 --- a/apps/server-frontend/package.json +++ b/apps/server-frontend/package.json @@ -1,7 +1,7 @@ { "name": "collapse-server-frontend", "private": true, - "version": "0.8.0", + "version": "0.9.0", "type": "module", "scripts": { "dev": "vite", 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`. diff --git a/docs/threat_model.md b/docs/threat_model.md index e4d009d..230e645 100644 --- a/docs/threat_model.md +++ b/docs/threat_model.md @@ -149,11 +149,25 @@ tar has used it since v7), so the split is now on `/` on every machine, and a backslash is judged as what it is: an ordinary character, legal on Unix, refused by Windows. +An archive whose **listing cannot be read** stops extraction before anything is +written. It used to be waved through: the pre-flight pass gave up, extraction +carried on with no plan, and every entry before the damage was written under its +raw name, unrewritten and unrefused. One bad 512 byte header appended to a tar +was enough, because tar lists all its headers up front and writes as it walks. +On Windows that turned `notes.txt:hidden` into an invisible NTFS stream without +the user ever seeing the question (issue #89). + +Recovering what a damaged archive still holds is not gone, it is simply no +longer the default: the backends (`extract_tar` and friends) take no options and +never come through the planning pass. + **Covered by** `apps/core/tests/names.rs`, in particular `an_entry_splits_the_same_way_on_every_host`, `the_report_sees_a_colon_in_the_first_component_too`, `only_windows_refuses_a_backslash_inside_a_component`, and -`a_unix_name_holding_a_backslash_survives_the_round_trip`. +`a_unix_name_holding_a_backslash_survives_the_round_trip`, +`a_damaged_archive_writes_nothing_rather_than_writing_raw_names` and +`the_same_names_are_refused_whether_or_not_the_archive_is_damaged`. ## Compression measures