diff --git a/updater/src/app.rs b/updater/src/app.rs index 65b4a9143..33620501c 100644 --- a/updater/src/app.rs +++ b/updater/src/app.rs @@ -11,7 +11,6 @@ use crate::{ }; use anyhow::{Context, Result}; use chrono::{Duration as ChronoDuration, Utc}; -use reqwest::Client; use serde::Deserialize; use std::{ ffi::OsString, @@ -933,7 +932,7 @@ async fn run_check_cycle( return Ok(()); }; - let client = Client::builder().build()?; + let client = upstream::http_client()?; sync_runtime_state(config, state); state.status = UpdateStatus::CheckingUpstream; diff --git a/updater/src/upstream.rs b/updater/src/upstream.rs index 2a6687c31..68d390717 100644 --- a/updater/src/upstream.rs +++ b/updater/src/upstream.rs @@ -5,8 +5,35 @@ use chrono::{DateTime, Utc}; use futures_util::StreamExt; use reqwest::{header, Client}; use sha2::{Digest, Sha256}; -use std::path::{Path, PathBuf}; -use tokio::{fs::File, io::AsyncWriteExt}; +use std::time::Duration; +use std::{ + path::{Path, PathBuf}, + sync::atomic::{AtomicU64, Ordering}, +}; +use tokio::{fs::OpenOptions, io::AsyncWriteExt}; + +const HTTP_CONNECT_TIMEOUT: Duration = Duration::from_secs(30); +const HTTP_READ_TIMEOUT: Duration = Duration::from_secs(60); +const DOWNLOAD_TEMP_PREFIX: &str = ".Codex.dmg.download-"; +static DOWNLOAD_TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); + +struct DownloadTempFile { + path: PathBuf, +} + +impl DownloadTempFile { + fn commit(mut self) { + self.path = PathBuf::new(); + } +} + +impl Drop for DownloadTempFile { + fn drop(&mut self) { + if !self.path.as_os_str().is_empty() { + let _ = std::fs::remove_file(&self.path); + } + } +} #[derive(Debug, Clone, PartialEq, Eq)] /// Selected HTTP metadata used to detect upstream DMG changes. @@ -25,6 +52,15 @@ pub struct DownloadedDmg { pub candidate_version: String, } +/// Builds the HTTP client shared by upstream metadata and DMG requests. +pub fn http_client() -> Result { + Client::builder() + .connect_timeout(HTTP_CONNECT_TIMEOUT) + .read_timeout(HTTP_READ_TIMEOUT) + .build() + .context("Failed to build upstream HTTP client") +} + /// Fetches the upstream DMG headers used to detect candidate updates. pub async fn fetch_remote_metadata(client: &Client, dmg_url: &str) -> Result { let response = client @@ -80,11 +116,6 @@ pub async fn download_dmg( .await .with_context(|| format!("Failed to create {}", destination_dir.display()))?; - let destination = destination_dir.join("Codex.dmg"); - let mut file = File::create(&destination) - .await - .with_context(|| format!("Failed to create {}", destination.display()))?; - let response = client .get(dmg_url) .send() @@ -93,6 +124,9 @@ pub async fn download_dmg( .error_for_status() .with_context(|| format!("GET request for {dmg_url} returned an error status"))?; + let destination = destination_dir.join("Codex.dmg"); + let (temp, mut file) = create_download_temp(destination_dir).await?; + let mut hasher = Sha256::new(); let mut stream = response.bytes_stream(); @@ -100,13 +134,17 @@ pub async fn download_dmg( let chunk = chunk.with_context(|| format!("Failed downloading {dmg_url}"))?; file.write_all(&chunk) .await - .with_context(|| format!("Failed writing {}", destination.display()))?; + .with_context(|| format!("Failed writing {}", temp.path.display()))?; hasher.update(&chunk); } file.flush() .await - .with_context(|| format!("Failed flushing {}", destination.display()))?; + .with_context(|| format!("Failed flushing {}", temp.path.display()))?; + file.sync_all() + .await + .with_context(|| format!("Failed syncing {}", temp.path.display()))?; + drop(file); let sha256 = hasher .finalize() @@ -115,6 +153,16 @@ pub async fn download_dmg( .collect::(); let candidate_version = derive_candidate_version(&sha256, version_timestamp)?; + tokio::fs::rename(&temp.path, &destination) + .await + .with_context(|| { + format!( + "Failed to atomically replace {} with completed download", + destination.display() + ) + })?; + temp.commit(); + Ok(DownloadedDmg { path: destination, sha256, @@ -122,6 +170,30 @@ pub async fn download_dmg( }) } +async fn create_download_temp( + destination_dir: &Path, +) -> Result<(DownloadTempFile, tokio::fs::File)> { + loop { + let id = DOWNLOAD_TEMP_COUNTER.fetch_add(1, Ordering::Relaxed); + let path = destination_dir.join(format!( + "{DOWNLOAD_TEMP_PREFIX}{}-{id}.tmp", + std::process::id() + )); + match OpenOptions::new() + .write(true) + .create_new(true) + .open(&path) + .await + { + Ok(file) => return Ok((DownloadTempFile { path }, file)), + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue, + Err(error) => { + return Err(error).with_context(|| format!("Failed to create {}", path.display())) + } + } + } +} + /// Derives a local package version from the DMG hash and download timestamp. pub fn derive_candidate_version(sha256: &str, timestamp: DateTime) -> Result { let short_hash = sha256 @@ -139,6 +211,12 @@ mod tests { use super::*; use anyhow::Result; use chrono::TimeZone; + use std::{ + io::{Read, Write}, + net::TcpListener, + thread, + time::Duration, + }; use tempfile::tempdir; use wiremock::{ matchers::{method, path}, @@ -184,6 +262,7 @@ mod tests { let client = Client::builder().build()?; let temp = tempdir()?; + std::fs::write(temp.path().join("Codex.dmg"), b"old-valid-cache")?; let downloaded = download_dmg( &client, &format!("{}/Codex.dmg", server.uri()), @@ -198,6 +277,99 @@ mod tests { "678cd508ffe0071e217020a7a4eecbebe25362c022ac78c13a5ae87b7a3a0c92" ); assert_eq!(downloaded.candidate_version, "2026.03.24.120000+678cd508"); + assert_eq!(std::fs::read(&downloaded.path)?, body); + assert_no_download_temps(temp.path())?; + Ok(()) + } + + #[tokio::test] + async fn http_error_preserves_existing_dmg() -> Result<()> { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/Codex.dmg")) + .respond_with(ResponseTemplate::new(503)) + .mount(&server) + .await; + + let temp = tempdir()?; + let destination = temp.path().join("Codex.dmg"); + std::fs::write(&destination, b"old-valid-cache")?; + let error = download_dmg( + &Client::builder().build()?, + &format!("{}/Codex.dmg", server.uri()), + temp.path(), + Utc::now(), + ) + .await + .expect_err("error response should fail"); + + assert!(error.to_string().contains("returned an error status")); + assert_eq!(std::fs::read(destination)?, b"old-valid-cache"); + assert_no_download_temps(temp.path())?; + Ok(()) + } + + #[tokio::test] + async fn interrupted_body_preserves_existing_dmg_and_cleans_temp() -> Result<()> { + let listener = TcpListener::bind("127.0.0.1:0")?; + let address = listener.local_addr()?; + let server = thread::spawn(move || -> std::io::Result<()> { + let (mut connection, _) = listener.accept()?; + let mut request = [0_u8; 1024]; + let mut request_len = 0; + while request_len < request.len() { + let read = connection.read(&mut request[request_len..])?; + if read == 0 { + break; + } + request_len += read; + if request[..request_len] + .windows(4) + .any(|window| window == b"\r\n\r\n") + { + break; + } + } + connection.write_all( + b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\nConnection: close\r\n\r\n", + )?; + connection.flush()?; + thread::sleep(Duration::from_millis(100)); + connection.write_all(b"partial")?; + connection.flush()?; + thread::sleep(Duration::from_millis(100)); + Ok(()) + }); + + let temp = tempdir()?; + let destination = temp.path().join("Codex.dmg"); + std::fs::write(&destination, b"old-valid-cache")?; + let result = download_dmg( + &Client::builder().build()?, + &format!("http://{address}/Codex.dmg"), + temp.path(), + Utc::now(), + ) + .await; + server.join().expect("test server thread panicked")?; + + assert!(result.is_err(), "truncated body should fail"); + assert_eq!(std::fs::read(destination)?, b"old-valid-cache"); + assert_no_download_temps(temp.path())?; + Ok(()) + } + + fn assert_no_download_temps(directory: &Path) -> Result<()> { + let leftovers = std::fs::read_dir(directory)? + .filter_map(|entry| entry.ok()) + .filter(|entry| { + entry + .file_name() + .to_string_lossy() + .starts_with(DOWNLOAD_TEMP_PREFIX) + }) + .collect::>(); + assert!(leftovers.is_empty(), "temporary downloads remain"); Ok(()) } diff --git a/updater/src/wrapper_apply.rs b/updater/src/wrapper_apply.rs index 4414d757f..da0770650 100644 --- a/updater/src/wrapper_apply.rs +++ b/updater/src/wrapper_apply.rs @@ -18,6 +18,7 @@ use serde_json::Value; use std::{ collections::HashSet, fs, + io::{BufReader, Read}, os::unix::fs::{self as unix_fs, PermissionsExt}, path::{Path, PathBuf}, process::Command, @@ -489,7 +490,7 @@ async fn cached_or_downloaded_dmg( } } - let client = reqwest::Client::builder().build()?; + let client = upstream::http_client()?; let downloads_dir = config.workspace_root.join("downloads"); let downloaded = upstream::download_dmg(&client, &config.dmg_url, &downloads_dir, chrono::Utc::now()) @@ -504,10 +505,20 @@ async fn cached_or_downloaded_dmg( /// contents, matching the DMG update path's scheme. fn derive_package_version(dmg_path: &Path) -> Result { use sha2::{Digest, Sha256}; - let bytes = std::fs::read(dmg_path) - .with_context(|| format!("Failed to read {}", dmg_path.display()))?; + let file = fs::File::open(dmg_path) + .with_context(|| format!("Failed to open {}", dmg_path.display()))?; + let mut reader = BufReader::new(file); let mut hasher = Sha256::new(); - hasher.update(&bytes); + let mut buffer = [0_u8; 64 * 1024]; + loop { + let bytes_read = reader + .read(&mut buffer) + .with_context(|| format!("Failed to read {}", dmg_path.display()))?; + if bytes_read == 0 { + break; + } + hasher.update(&buffer[..bytes_read]); + } let sha = hasher .finalize() .iter() @@ -581,6 +592,18 @@ mod tests { } } + #[test] + fn derives_package_version_with_streamed_dmg_hash() -> Result<()> { + let root = tempdir()?; + let dmg = root.path().join("Codex.dmg"); + std::fs::write(&dmg, b"codex-dmg-test-payload")?; + + let version = derive_package_version(&dmg)?; + + assert!(version.ends_with("+678cd508")); + Ok(()) + } + fn write_local_feature(root: &Path, id: &str) { let feature_dir = root.join("builder/linux-features/local").join(id); std::fs::create_dir_all(feature_dir.join("nested")).unwrap();