From 052f1e5887ffaa88fda05906a848687091e222b1 Mon Sep 17 00:00:00 2001 From: Roman Inflianskas Date: Mon, 3 Aug 2026 11:27:25 +0000 Subject: [PATCH 1/2] fix(download): never leave partial files behind MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Interrupted downloads and failed atomic writes left large partial files on disk that nothing ever removed. Because each temp name embeds a timestamp, every retry leaked a distinct orphan — so on a full disk each attempt consumed more of the little space that was left. Stream downloads to a sibling .part file and rename into place only once the body is complete, removing the partial on every failure path. This also stops a truncated transfer from landing at the destination, where the lemonade engine would accept it as a valid cache entry forever. Clean up the temp file on the failure paths of both write_file_atomically copies, and preserve full file names so sdk.tar.gz no longer becomes sdk.tar.tmp-. The SDK tarball is now streamed rather than read fully into memory, which removes a multi-gigabyte allocation on the install path. Refs #158 Signed-off-by: Roman Inflianskas --- apps/rocm/src/therock.rs | 120 ++++++++++++++++++++++++++---- apps/rocmd/src/lib.rs | 65 +++++++++++++++-- crates/rocm-core/src/lib.rs | 141 +++++++++++++++++++++++++++++++++++- 3 files changed, 302 insertions(+), 24 deletions(-) diff --git a/apps/rocm/src/therock.rs b/apps/rocm/src/therock.rs index 204a15ae..2606f6f3 100644 --- a/apps/rocm/src/therock.rs +++ b/apps/rocm/src/therock.rs @@ -10,8 +10,8 @@ use rocm_core::{ normalize_runtime_path_for_storage, normalize_runtime_path_text_for_host, normalize_runtime_path_text_for_storage, normalize_therock_family, runtime_is_windows, runtime_os_name, runtime_path_for_windows_child, runtime_path_list_split, - runtime_python_executable_in_env, unix_time_millis, uv_command_env, uv_pip_install_base, - uv_venv_args, verify_rsa_pkcs1_sha256_signature, + runtime_python_executable_in_env, stream_to_path_atomically, unix_time_millis, uv_command_env, + uv_pip_install_base, uv_venv_args, verify_rsa_pkcs1_sha256_signature, }; #[cfg(test)] use rocm_core::{generate_rsa_signing_keypair, sign_rsa_pkcs1_sha256_signature}; @@ -2075,16 +2075,26 @@ fn http_header_value(headers: &str, name: &str) -> Option { value } +/// Download `url` to `destination`, streaming the body straight to disk. +/// +/// SDK tarballs are multi-gigabyte, so the body is never held in memory: it is +/// streamed to a sibling `.part` file and renamed into place once complete. fn download_file(url: &str, destination: &Path) -> Result<()> { let parent = destination .parent() .context("download destination has no parent directory")?; fs::create_dir_all(parent)?; - let response = http_get(url, &[], None)?; - if response.status != 200 { - bail!("HTTP {} while fetching {url}", response.status); - } - write_file_atomically(destination, &response.body) + let agent = ureq::AgentBuilder::new() + .timeout(Duration::from_mins(10)) + .build(); + let response = match agent.get(url).set("User-Agent", "rocm-cli").call() { + Ok(response) => response, + Err(ureq::Error::Status(status, _)) => bail!("HTTP {status} while fetching {url}"), + Err(error) => bail!("HTTP request failed for {url}: {error}"), + }; + let mut reader = response.into_reader(); + stream_to_path_atomically(&mut reader, destination) + .with_context(|| format!("failed to download {url}")) } fn http_get( @@ -2184,20 +2194,50 @@ fn windows_child_path(path: &Path) -> String { runtime_path_for_windows_child(path) } +/// A unique temp path next to `path`, preserving the full file name so a +/// multi-extension artifact keeps its extensions (`sdk.tar.gz` becomes +/// `sdk.tar.gz.tmp-`, where `with_extension` would drop `.gz`). +fn temp_sibling_path(path: &Path) -> Result { + let parent = path.parent().context("file path has no parent directory")?; + let file_name = path + .file_name() + .context("file path has no file name")? + .to_string_lossy() + .into_owned(); + Ok(parent.join(format!( + "{file_name}.tmp-{}-{}", + std::process::id(), + unix_time_millis() + ))) +} + fn write_file_atomically(path: &Path, bytes: &[u8]) -> Result<()> { let parent = path.parent().context("file path has no parent directory")?; fs::create_dir_all(parent)?; - let tmp = path.with_extension(format!("tmp-{}", unix_time_millis())); - { + let tmp = temp_sibling_path(path)?; + // Clean up the temp file on every failure path. It carries a unique + // timestamped name, so leaving it behind would accumulate a fresh orphan + // per attempt — and when the failure is a full disk, those orphans are what + // keep it full. + let write_result = (|| -> Result<()> { let mut file = fs::File::create(&tmp) .with_context(|| format!("failed to create {}", tmp.display()))?; file.write_all(bytes) .with_context(|| format!("failed to write {}", tmp.display()))?; - } - fs::rename(&tmp, path).or_else(|_| { - let _ = fs::remove_file(path); - fs::rename(&tmp, path) - })?; + Ok(()) + })(); + if let Err(error) = write_result { + let _ = fs::remove_file(&tmp); + return Err(error); + } + fs::rename(&tmp, path) + .or_else(|_| { + let _ = fs::remove_file(path); + fs::rename(&tmp, path) + }) + .inspect_err(|_| { + let _ = fs::remove_file(&tmp); + })?; Ok(()) } @@ -4333,4 +4373,56 @@ echo Python 3.12.10 "invalid calendar dates should not be displayed" ); } + + /// Regression: a failed write must not leave a `.tmp-*` scratch file + /// behind. The name is unique per attempt, so an orphan per retry used to + /// accumulate — and when the failure is a full disk, those orphans are + /// exactly what keeps it full. + /// + /// Ignored by default: it fills `/dev/shm` to provoke ENOSPC, which is + /// shared with anything else on the host. Run with + /// `cargo test -p rocm -- --ignored write_file_atomically_cleans_up`. + #[test] + #[ignore = "fills /dev/shm to provoke ENOSPC; not safe to run concurrently"] + fn write_file_atomically_cleans_up_temp_on_write_failure() { + let shm = Path::new("/dev/shm"); + if !shm.is_dir() { + eprintln!("skipping: /dev/shm unavailable"); + return; + } + let dir = shm.join(format!("rocm-enospc-{}", std::process::id())); + let _ = fs::remove_dir_all(&dir); + fs::create_dir_all(&dir).unwrap(); + let dest = dir.join("artifact.tar.gz"); + // Larger than the tmpfs, so the write is guaranteed to hit ENOSPC. + let payload = vec![0u8; 256 * 1024 * 1024]; + + for _ in 0..2 { + write_file_atomically(&dest, &payload) + .expect_err("writing past the end of the filesystem should fail"); + let leftovers: Vec = fs::read_dir(&dir) + .unwrap() + .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) + .collect(); + assert!( + leftovers.is_empty(), + "failed write left files behind: {leftovers:?}" + ); + } + assert!(!dest.exists(), "destination must not exist after failure"); + let _ = fs::remove_dir_all(&dir); + } + + /// The temp name keeps every extension, so a cleanup glob over a cache + /// directory can still tell what a leftover was going to be. + #[test] + fn temp_sibling_path_preserves_multi_dot_file_names() { + let temp = temp_sibling_path(Path::new("/tmp/cache/sdk.tar.gz")).unwrap(); + let name = temp.file_name().unwrap().to_string_lossy().into_owned(); + assert!( + name.starts_with("sdk.tar.gz.tmp-"), + "expected the full name to be preserved, got {name}" + ); + assert_eq!(temp.parent().unwrap(), Path::new("/tmp/cache")); + } } diff --git a/apps/rocmd/src/lib.rs b/apps/rocmd/src/lib.rs index a448cb76..732c5645 100644 --- a/apps/rocmd/src/lib.rs +++ b/apps/rocmd/src/lib.rs @@ -1064,12 +1064,31 @@ fn sha256_hex(bytes: &[u8]) -> String { fn write_file_atomically(path: &Path, bytes: &[u8]) -> Result<()> { let parent = path.parent().context("file path has no parent directory")?; fs::create_dir_all(parent).with_context(|| format!("failed to create {}", parent.display()))?; - let tmp = path.with_extension(format!("tmp-{}", unix_time_millis())); - fs::write(&tmp, bytes).with_context(|| format!("failed to write {}", tmp.display()))?; - fs::rename(&tmp, path).or_else(|_| { - let _ = fs::remove_file(path); - fs::rename(&tmp, path) - })?; + // Preserve the full file name so multi-extension paths keep their + // extensions, and remove the temp file on every failure path so repeated + // attempts cannot accumulate orphans. + let file_name = path + .file_name() + .context("file path has no file name")? + .to_string_lossy() + .into_owned(); + let tmp = parent.join(format!( + "{file_name}.tmp-{}-{}", + std::process::id(), + unix_time_millis() + )); + if let Err(error) = fs::write(&tmp, bytes) { + let _ = fs::remove_file(&tmp); + return Err(error).with_context(|| format!("failed to write {}", tmp.display())); + } + fs::rename(&tmp, path) + .or_else(|_| { + let _ = fs::remove_file(path); + fs::rename(&tmp, path) + }) + .inspect_err(|_| { + let _ = fs::remove_file(&tmp); + })?; Ok(()) } @@ -5006,6 +5025,40 @@ mod tests { use rocm_core::ModelRecipeArtifactSourcePolicyRecord; use std::path::PathBuf; + /// Regression: a failed write must not leave a `.tmp-*` scratch file + /// behind. Mirrors the test in `apps/rocm/src/therock.rs`. + /// + /// Ignored by default: it fills `/dev/shm` to provoke ENOSPC, which is + /// shared with anything else on the host. + #[test] + #[ignore = "fills /dev/shm to provoke ENOSPC; not safe to run concurrently"] + fn write_file_atomically_cleans_up_temp_on_write_failure() { + let shm = std::path::Path::new("/dev/shm"); + if !shm.is_dir() { + eprintln!("skipping: /dev/shm unavailable"); + return; + } + let dir = shm.join(format!("rocmd-enospc-{}", std::process::id())); + let _ = fs::remove_dir_all(&dir); + fs::create_dir_all(&dir).unwrap(); + let dest = dir.join("manifest.json"); + let payload = vec![b'x'; 256 * 1024 * 1024]; + + write_file_atomically(&dest, &payload) + .expect_err("writing past the end of the filesystem should fail"); + let leftovers: Vec = fs::read_dir(&dir) + .unwrap() + .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) + .collect(); + assert!( + leftovers.is_empty(), + "failed write left files behind: {leftovers:?}" + ); + + assert!(!dest.exists()); + let _ = fs::remove_dir_all(&dir); + } + #[test] fn last_cr_segment_keeps_final_progress_redraw() { // A tqdm/HF-style in-place redraw collapses to its last segment. diff --git a/crates/rocm-core/src/lib.rs b/crates/rocm-core/src/lib.rs index 92e46e08..b5235598 100644 --- a/crates/rocm-core/src/lib.rs +++ b/crates/rocm-core/src/lib.rs @@ -112,6 +112,16 @@ pub fn parse_http_endpoint(endpoint_url: &str) -> Option<(String, u16)> { Some((host.to_owned(), port.parse().ok()?)) } +/// Suffix marking an in-progress download. Callers that sweep a cache directory +/// can recognize (and safely delete) leftovers by this pattern. +pub const PARTIAL_DOWNLOAD_SUFFIX: &str = ".part"; + +/// Download `url` to `destination`, streaming the body to disk. +/// +/// The body is written to a sibling `.part` file and renamed into place only +/// after the transfer completes, so an interrupted download never leaves a +/// truncated file at `destination` where callers would mistake it for a +/// complete artifact. The partial file is removed on every failure path. pub fn download_file_to_path(url: &str, destination: &Path, timeout: Duration) -> Result<()> { let response = ureq::get(url) .timeout(timeout) @@ -125,10 +135,58 @@ pub fn download_file_to_path(url: &str, destination: &Path, timeout: Duration) - .with_context(|| format!("failed to create {}", parent.display()))?; } let mut reader = response.into_reader(); - let mut file = fs::File::create(destination) - .with_context(|| format!("failed to create {}", destination.display()))?; - std::io::copy(&mut reader, &mut file) - .with_context(|| format!("failed to write {}", destination.display()))?; + stream_to_path_atomically(&mut reader, destination) + .with_context(|| format!("failed to download {url}")) +} + +/// Stream `reader` to `destination` via a sibling `.part` file, renaming into +/// place only once the whole body is written. Removes the partial file if +/// anything fails, so retries cannot accumulate orphans. +pub fn stream_to_path_atomically(reader: &mut dyn std::io::Read, destination: &Path) -> Result<()> { + let parent = destination + .parent() + .context("download destination has no parent directory")?; + fs::create_dir_all(parent).with_context(|| format!("failed to create {}", parent.display()))?; + + // Keep the original file name intact and append the marker, so a + // multi-extension artifact (`sdk.tar.gz`) yields `sdk.tar.gz.part-` + // rather than losing an extension the way `with_extension` would. + let file_name = destination + .file_name() + .context("download destination has no file name")? + .to_string_lossy() + .into_owned(); + let partial = parent.join(format!( + "{file_name}{PARTIAL_DOWNLOAD_SUFFIX}-{}-{}", + std::process::id(), + unix_time_millis() + )); + + let result = (|| -> Result<()> { + let mut file = fs::File::create(&partial) + .with_context(|| format!("failed to create {}", partial.display()))?; + std::io::copy(reader, &mut file) + .with_context(|| format!("failed to write {}", partial.display()))?; + file.sync_all() + .with_context(|| format!("failed to flush {}", partial.display()))?; + Ok(()) + })(); + + if let Err(error) = result { + let _ = fs::remove_file(&partial); + return Err(error); + } + + if let Err(error) = fs::rename(&partial, destination) { + let _ = fs::remove_file(&partial); + return Err(error).with_context(|| { + format!( + "failed to move {} into place at {}", + partial.display(), + destination.display() + ) + }); + } Ok(()) } @@ -8708,4 +8766,79 @@ last_installed_runtime_id = "therock-release" None ); } + + /// Regression: an interrupted transfer must not leave a truncated file at + /// the destination. Callers treat any file at that path as a complete, + /// cached artifact, so a partial one there poisons the cache permanently. + #[test] + fn download_leaves_no_truncated_file_at_destination() { + use std::io::Read; + + /// Yields a few bytes, then fails — a connection dropped mid-body. + struct TruncatedBody { + sent: bool, + } + impl Read for TruncatedBody { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + if self.sent { + return Err(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "response body closed before all bytes were read", + )); + } + self.sent = true; + let chunk = b"TRUNCATED-BYTES\n"; + buf[..chunk.len()].copy_from_slice(chunk); + Ok(chunk.len()) + } + } + + let dir = std::env::temp_dir().join(format!("rocm-core-partial-{}", std::process::id())); + let _ = fs::remove_dir_all(&dir); + let cache = dir.join("cache"); + let dest = cache.join("lemonade.tar.gz"); + + let mut body = TruncatedBody { sent: false }; + stream_to_path_atomically(&mut body, &dest) + .expect_err("a body that ends early should surface as an error"); + + assert!( + !dest.exists(), + "truncated download must not be left at {}", + dest.display() + ); + let leftovers: Vec = fs::read_dir(&cache) + .map(|entries| { + entries + .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) + .collect() + }) + .unwrap_or_default(); + assert!( + leftovers.is_empty(), + "failed download left files behind: {leftovers:?}" + ); + + let _ = fs::remove_dir_all(&dir); + } + + /// The happy path renames into place and keeps every extension. + #[test] + fn stream_to_path_atomically_writes_complete_body() { + let dir = std::env::temp_dir().join(format!("rocm-core-complete-{}", std::process::id())); + let _ = fs::remove_dir_all(&dir); + let dest = dir.join("cache").join("sdk.tar.gz"); + + let mut body = &b"complete-payload"[..]; + stream_to_path_atomically(&mut body, &dest).expect("complete body should be written"); + + assert_eq!(fs::read(&dest).unwrap(), b"complete-payload"); + let leftovers: Vec = fs::read_dir(dest.parent().unwrap()) + .unwrap() + .map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned()) + .collect(); + assert_eq!(leftovers, vec!["sdk.tar.gz".to_owned()]); + + let _ = fs::remove_dir_all(&dir); + } } From 7204154f918f9107e0c943d4f5a4863080d5e44c Mon Sep 17 00:00:00 2001 From: Roman Inflianskas Date: Fri, 7 Aug 2026 07:51:20 +0000 Subject: [PATCH 2/2] test(e2e): retry the Observe tab key until the dashboard acts on it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The two dashboard scenarios that open Observe send the `4` tab key straight after launching the TUI, with no assertion in between. A key written into the pseudo-terminal before the dashboard is reading input can be consumed by whatever holds the terminal at that moment, and nothing ever retries it — so the dashboard stays on Home and the scenario fails 30s later in an assertion about a view it never left. Send the key until the Observe chip is actually marked active, so the step depends on the dashboard having acted on the key rather than on it having been ready when the key was written. Reproduced by pointing ROCM_CLI_BINARY at a wrapper that drains the terminal before exec'ing the real binary: the scenario failed with exactly the CI symptom before this change and passes after it. Signed-off-by: Roman Inflianskas --- tests/e2e-cucumber/tests/e2e/dash_steps.rs | 8 ++++- tests/e2e-cucumber/tests/e2e/tui_driver.rs | 40 ++++++++++++++++++++++ 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/tests/e2e-cucumber/tests/e2e/dash_steps.rs b/tests/e2e-cucumber/tests/e2e/dash_steps.rs index aa618ff9..c9c553f0 100644 --- a/tests/e2e-cucumber/tests/e2e/dash_steps.rs +++ b/tests/e2e-cucumber/tests/e2e/dash_steps.rs @@ -121,7 +121,13 @@ async fn open_observe_view(world: &mut E2eWorld) { let tui = session(world); tui.use_detail_size() .unwrap_or_else(|e| panic!("failed to enlarge the dashboard: {e}")); - tui.send("4") + // Unlike the demo-data journeys, these scenarios open the dashboard and + // switch tabs with no assertion in between, so the keystroke can land + // before the dashboard is reading input. Repeat it until the Observe tab is + // actually selected (the `●` marks the active chip), so the step fails only + // if the dashboard never gets there — not if it was slow to start. + tui.send_until("4", "● Observe", DEFAULT_TIMEOUT) + .await .unwrap_or_else(|e| panic!("failed to switch to the Observe tab: {e}")); } diff --git a/tests/e2e-cucumber/tests/e2e/tui_driver.rs b/tests/e2e-cucumber/tests/e2e/tui_driver.rs index 9996dba4..16c88227 100644 --- a/tests/e2e-cucumber/tests/e2e/tui_driver.rs +++ b/tests/e2e-cucumber/tests/e2e/tui_driver.rs @@ -47,6 +47,10 @@ const DETAIL_COLS: u16 = 120; /// poll cadence, not a fixed readiness sleep: every wait has a deadline and /// returns the instant its condition holds. const POLL_INTERVAL: Duration = Duration::from_millis(20); +/// How long [`TuiSession::send_until`] waits for a key to take effect before +/// sending it again. Long enough that a busy host is not spammed with repeats, +/// short enough that several attempts fit inside a normal step timeout. +const KEY_RESEND_INTERVAL: Duration = Duration::from_millis(500); /// Maximum time to let the PTY reader consume the child's final frame after the /// process exits. This is bounded so a misbehaving PTY cannot stall a scenario. const DRAIN_TIMEOUT: Duration = Duration::from_millis(250); @@ -278,6 +282,42 @@ impl TuiSession { .map_err(|e| format!("failed to write to pty: {e}")) } + /// Send `bytes` until the screen shows `marker`, re-sending on an interval + /// until the deadline. + /// + /// A bare [`send`](Self::send) writes into the pseudo-terminal whether or + /// not the application is reading yet, so a keystroke typed during startup + /// can be consumed by whatever holds the terminal at that moment and never + /// reach the event loop. The key is then simply lost — nothing retries it, + /// and the scenario fails much later, in an assertion about a view it never + /// left. Re-sending until the expected view appears makes the step depend on + /// the application having acted on the key rather than on it having been + /// ready when the key was written. + /// + /// Only safe for idempotent keys (a tab jump, not a toggle). + pub async fn send_until( + &mut self, + bytes: &str, + marker: &str, + timeout: Duration, + ) -> Result<(), String> { + let deadline = Instant::now() + timeout; + loop { + self.send(bytes)?; + let remaining = deadline.saturating_duration_since(Instant::now()); + let attempt = KEY_RESEND_INTERVAL.min(remaining); + if self.wait_for_screen(marker, attempt).await.is_ok() { + return Ok(()); + } + if Instant::now() >= deadline { + return Err(format!( + "timed out after {timeout:?} waiting for {marker:?} while repeating {bytes:?}\n{}", + self.framed_screen() + )); + } + } + } + /// Poll the current screen until it contains `marker`, or fail with a /// deadline that includes the last screen for diagnosis. Also fails fast if /// the child exits before the marker appears.