diff --git a/Cargo.lock b/Cargo.lock index c10fa56..1f515fd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -362,7 +362,6 @@ dependencies = [ "anyhow", "dirs", "flate2", - "futures-util", "libc", "libloading", "log", @@ -374,7 +373,6 @@ dependencies = [ "tar", "tempfile", "thiserror", - "tokio", "xz2", "zstd", ] @@ -502,17 +500,6 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" -[[package]] -name = "futures-macro" -version = "0.3.34" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" -dependencies = [ - "proc-macro2", - "quote", - "syn 3.0.9", -] - [[package]] name = "futures-sink" version = "0.3.34" @@ -533,7 +520,6 @@ checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ "futures-core", "futures-io", - "futures-macro", "futures-sink", "futures-task", "memchr", @@ -1229,14 +1215,12 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-rustls", - "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams", "web-sys", ] @@ -1691,21 +1675,9 @@ dependencies = [ "mio", "pin-project-lite", "socket2", - "tokio-macros", "windows-sys 0.61.2", ] -[[package]] -name = "tokio-macros" -version = "2.7.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" -dependencies = [ - "proc-macro2", - "quote", - "syn 3.0.9", -] - [[package]] name = "tokio-rustls" version = "0.26.6" @@ -1947,19 +1919,6 @@ dependencies = [ "unicode-ident", ] -[[package]] -name = "wasm-streams" -version = "0.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb" -dependencies = [ - "futures-util", - "js-sys", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", -] - [[package]] name = "web-sys" version = "0.3.106" diff --git a/crates/ecrou/Cargo.toml b/crates/ecrou/Cargo.toml index 9a52d0a..20f38b5 100644 --- a/crates/ecrou/Cargo.toml +++ b/crates/ecrou/Cargo.toml @@ -13,7 +13,7 @@ serde = { version = "1", features = ["derive"] } serde_yaml = "0.9" # HTTP downloads -reqwest = { version = "0.13", features = ["blocking", "stream"] } +reqwest = { version = "0.13", features = ["blocking"] } # Tarball extraction tar = "0.4" @@ -36,8 +36,6 @@ thiserror = "2" # Utilities dirs = "6" -tokio = { version = "1", features = ["rt-multi-thread", "macros", "io-util"] } -futures-util = "0.3" serde_json = "1" # libkrun FFI: raw ioctl for the KVM probe, runtime dlopen for the ABI diff --git a/crates/ecrou/src/distro.rs b/crates/ecrou/src/distro.rs index addfba7..683b5e0 100644 --- a/crates/ecrou/src/distro.rs +++ b/crates/ecrou/src/distro.rs @@ -1,4 +1,5 @@ use std::path::PathBuf; +use std::time::Duration; use anyhow::{anyhow, Context}; @@ -198,7 +199,10 @@ pub fn checksum_artifact_url(distro: &Distro, tarball_url: &str) -> Option anyhow::Result { - let text = reqwest::blocking::get(meta_release_url) + let client = crate::download::http_client(Some(Duration::from_secs(30)))?; + let text = client + .get(meta_release_url) + .send() .with_context(|| format!("Failed to fetch {}", meta_release_url))? .text() .with_context(|| format!("Failed to read {}", meta_release_url))?; @@ -240,7 +244,10 @@ fn fetch_alpine_version_from_branch(branch: &str, arch: &str) -> anyhow::Result< "https://dl-cdn.alpinelinux.org/alpine/{}/releases/{}/latest-releases.yaml", branch, arch ); - let text = reqwest::blocking::get(&url) + let client = crate::download::http_client(Some(Duration::from_secs(30)))?; + let text = client + .get(&url) + .send() .with_context(|| { format!( "Failed to fetch Alpine latest-releases.yaml for branch {}", @@ -305,7 +312,10 @@ pub fn resolve_distro_version( /// meta-release file (e.g. "24.04" → "noble", "22.04" → "jammy"). /// Uses the full meta-release (not -lts) so non-LTS versions are also covered. fn fetch_ubuntu_version_map() -> anyhow::Result> { - let text = reqwest::blocking::get("https://changelogs.ubuntu.com/meta-release") + let client = crate::download::http_client(Some(Duration::from_secs(30)))?; + let text = client + .get("https://changelogs.ubuntu.com/meta-release") + .send() .context("Failed to fetch Ubuntu meta-release")? .text() .context("Failed to read Ubuntu meta-release")?; diff --git a/crates/ecrou/src/download.rs b/crates/ecrou/src/download.rs index fdea007..52ba5fd 100644 --- a/crates/ecrou/src/download.rs +++ b/crates/ecrou/src/download.rs @@ -1,17 +1,32 @@ //! Download plumbing shared by both image sources: the digest hasher, -//! distro checksum files, and plain file transfers into the cache. +//! the shared HTTP client, distro checksum files, and plain file +//! transfers into the cache. use crate::progress::{Progress, Unit}; use anyhow::{anyhow, Context}; -use futures_util::StreamExt; -use reqwest::Client; +use reqwest::blocking::Client; use sha2::{Digest, Sha256, Sha512}; +use std::io::{Read, Write}; use std::path::{Path, PathBuf}; use std::time::Duration; use crate::distro::{Distro, ImageSource}; use crate::Result; +/// One HTTP client shape for the crate: a connect window, plus an +/// optional total request window. Metadata callers (checksum files, +/// token endpoints, version lookups) pass `Some` — small bodies, so a +/// total bound is safe. Transfer callers pass `None`: a slow transfer +/// of any size must finish, and the blocking client offers no per-read +/// window — a stalled stream is the operator's Ctrl-C. +pub(crate) fn http_client(request_window: Option) -> anyhow::Result { + let mut builder = Client::builder().connect_timeout(Duration::from_secs(30)); + if let Some(window) = request_window { + builder = builder.timeout(window); + } + builder.build().context("Failed to create HTTP client") +} + /// Path of the digest sidecar file for a cached OCI image. /// e.g. `foo.tar.gz` → `foo.tar.gz.digest` pub fn digest_sidecar(cache_path: &Path) -> PathBuf { @@ -91,11 +106,6 @@ pub fn download_image( arch: &str, progress: &dyn Progress, ) -> Result<()> { - // Create a single Tokio runtime for all async I/O in this download. - // Previously each of the two sync wrappers (download_file_sync and - // download_oci_image) created their own runtime; this consolidates them. - let rt = tokio::runtime::Runtime::new().context("Failed to create Tokio runtime")?; - match source { ImageSource::DirectTarball { distro, version } => { let url = crate::distro::resolve_distro_url(distro, version.as_deref(), arch)?; @@ -107,7 +117,7 @@ pub fn download_image( // next to it. A fetch failure (missing artifact, network error) // degrades to an unverified download with a warning rather than // hard-failing, mirroring the digest freshness check in rootfs. - let expected = match rt.block_on(fetch_published_checksum(distro, &url)) { + let expected = match fetch_published_checksum(distro, &url) { Ok(expected) => Some(expected), Err(e) => { log::warn!( @@ -118,12 +128,8 @@ pub fn download_image( None } }; - Ok(rt.block_on(download_file_async( - &url, - dest, - progress, - expected.as_deref(), - ))?) + download_file(&url, dest, progress, expected.as_deref())?; + Ok(()) } ImageSource::OciImage { registry, @@ -132,14 +138,15 @@ pub fn download_image( architecture, } => { log::debug!("Pulling OCI image: {}/{}:{}", registry, repository, tag); - Ok(rt.block_on(crate::registry::download_oci_image_async( + crate::registry::download_oci_image( registry, repository, tag, architecture, dest, progress, - ))?) + )?; + Ok(()) } // A local archive is read in place by the prepare step; there // is nothing to fetch. @@ -152,22 +159,15 @@ pub fn download_image( } /// Fetch the sha256 the distro publishes for the tarball at `tarball_url`. -async fn fetch_published_checksum(distro: &Distro, tarball_url: &str) -> anyhow::Result { +fn fetch_published_checksum(distro: &Distro, tarball_url: &str) -> anyhow::Result { let Some(checksum_url) = crate::distro::checksum_artifact_url(distro, tarball_url) else { return Err(anyhow!("no checksum artifact for {}", tarball_url)); }; - // Metadata call: a small body, so the read window alone bounds it. - let client = Client::builder() - .connect_timeout(Duration::from_secs(30)) - .read_timeout(Duration::from_secs(30)) - .build() - .context("Failed to create HTTP client")?; - + let client = http_client(Some(Duration::from_secs(30)))?; let response = client .get(&checksum_url) .send() - .await .with_context(|| format!("Failed to fetch checksum artifact {}", checksum_url))?; if !response.status().is_success() { @@ -178,7 +178,7 @@ async fn fetch_published_checksum(distro: &Distro, tarball_url: &str) -> anyhow: )); } - let text = response.text().await?; + let text = response.text()?; let filename = tarball_url.rsplit('/').next().unwrap_or(""); expected_digest_from_checksums(&text, filename).ok_or_else(|| { anyhow!( @@ -204,7 +204,7 @@ fn expected_digest_from_checksums(text: &str, filename: &str) -> Option None } -async fn download_file_async( +fn download_file( url: &str, dest: &Path, progress: &dyn Progress, @@ -212,20 +212,8 @@ async fn download_file_async( ) -> anyhow::Result<()> { log::debug!("Downloading: {}", url); - // No total timeout: a slow transfer of any size must finish. The - // connect window and the per-read window catch dead peers and - // stalled streams instead. - let client = Client::builder() - .connect_timeout(Duration::from_secs(30)) - .read_timeout(Duration::from_secs(60)) - .build() - .context("Failed to create HTTP client")?; - - let response = client - .get(url) - .send() - .await - .context("Failed to start download")?; + let client = http_client(None)?; + let mut response = client.get(url).send().context("Failed to start download")?; if !response.status().is_success() { return Err(anyhow!("Download failed: HTTP {}", response.status())); @@ -235,9 +223,7 @@ async fn download_file_async( // Create parent directories if let Some(parent) = dest.parent() { - tokio::fs::create_dir_all(parent) - .await - .context("Failed to create cache directory")?; + std::fs::create_dir_all(parent).context("Failed to create cache directory")?; } progress.begin( @@ -256,17 +242,19 @@ async fn download_file_async( // reaches the cache; on any failure after the temp file is created we // remove it so stale partial files don't accumulate. let mut hasher = DownloadHasher::new(); - let result: anyhow::Result<()> = async { - let mut file = tokio::fs::File::create(&temp_dest) - .await - .context("Failed to create temporary file")?; + let result: anyhow::Result<()> = (|| { + let mut file = + std::fs::File::create(&temp_dest).context("Failed to create temporary file")?; - let mut stream = response.bytes_stream(); - while let Some(chunk) = stream.next().await { - let chunk = chunk.context("Download interrupted")?; - tokio::io::copy(&mut &chunk[..], &mut file).await?; - hasher.update(&chunk); - progress.advance(chunk.len() as u64); + let mut buf = [0u8; 64 * 1024]; + loop { + let n = response.read(&mut buf).context("Download interrupted")?; + if n == 0 { + break; + } + file.write_all(&buf[..n]).context("Failed to write chunk")?; + hasher.update(&buf[..n]); + progress.advance(n as u64); } progress.finish(); @@ -275,18 +263,16 @@ async fn download_file_async( log::debug!("Verified content digest: {}", dest.display()); } - tokio::fs::rename(&temp_dest, dest) - .await + std::fs::rename(&temp_dest, dest) .context("Failed to move partial download to final destination")?; log::debug!("Download complete: {}", dest.display()); Ok(()) - } - .await; + })(); if result.is_err() { // Best-effort cleanup; ignore errors (file may not exist if creation failed). - let _ = tokio::fs::remove_file(&temp_dest).await; + let _ = std::fs::remove_file(&temp_dest); } result diff --git a/crates/ecrou/src/registry.rs b/crates/ecrou/src/registry.rs index 41d3cc2..6079edd 100644 --- a/crates/ecrou/src/registry.rs +++ b/crates/ecrou/src/registry.rs @@ -3,12 +3,13 @@ //! and content-addressed layer blobs bundled into a cache entry. use anyhow::{anyhow, Context}; -use futures_util::StreamExt; -use reqwest::Client; +use reqwest::blocking::Client; use serde_json::Value; +use std::io::{Read, Write}; use std::path::Path; use std::time::Duration; +use crate::download::http_client; use crate::download::{digest_sidecar, DownloadHasher}; use crate::progress::{Progress, Unit}; use crate::{Error, Result}; @@ -17,25 +18,13 @@ use crate::{Error, Result}; /// layers. The registry returns the content-addressable digest in the /// `Docker-Content-Digest` response header of any manifest GET. pub(crate) fn fetch_oci_digest(registry: &str, repository: &str, tag: &str) -> Result { - let digest = tokio::runtime::Runtime::new() - .context("Failed to create Tokio runtime")? - .block_on(fetch_oci_digest_async(registry, repository, tag))?; - Ok(digest) + fetch_oci_digest_inner(registry, repository, tag).map_err(Error::Other) } -async fn fetch_oci_digest_async( - registry: &str, - repository: &str, - tag: &str, -) -> anyhow::Result { - // Metadata call: a small body, so the read window alone bounds it. - let client = Client::builder() - .connect_timeout(Duration::from_secs(30)) - .read_timeout(Duration::from_secs(30)) - .build() - .context("Failed to create HTTP client")?; - - let token = get_auth_token(&client, registry, repository).await?; +fn fetch_oci_digest_inner(registry: &str, repository: &str, tag: &str) -> anyhow::Result { + // Metadata: a small body, so a total request bound is safe. + let client = http_client(Some(Duration::from_secs(30)))?; + let token = get_auth_token(&client, registry, repository)?; let manifest_url = if registry == "docker.io" { format!( @@ -57,7 +46,6 @@ async fn fetch_oci_digest_async( ) .header("Authorization", format!("Bearer {}", token)) .send() - .await .context("Failed to fetch manifest for digest check")?; if !response.status().is_success() { @@ -75,7 +63,7 @@ async fn fetch_oci_digest_async( .ok_or_else(|| anyhow!("Registry did not return a Docker-Content-Digest header")) } -pub(crate) async fn download_oci_image_async( +pub(crate) fn download_oci_image( registry: &str, repository: &str, tag: &str, @@ -83,17 +71,8 @@ pub(crate) async fn download_oci_image_async( dest: &Path, progress: &dyn Progress, ) -> anyhow::Result<()> { - // No total timeout: a slow transfer of any size must finish. The - // connect window and the per-read window catch dead peers and - // stalled streams instead. - let client = Client::builder() - .connect_timeout(Duration::from_secs(30)) - .read_timeout(Duration::from_secs(60)) - .build() - .context("Failed to create HTTP client")?; - - let digest = - try_download_oci_image(&client, registry, repository, tag, arch, dest, progress).await?; + let client = http_client(None)?; + let digest = try_download_oci_image(&client, registry, repository, tag, arch, dest, progress)?; // Persist the manifest digest so the next run can skip the download when // the tag hasn't changed (see fetch_oci_digest / digest_sidecar). @@ -104,7 +83,7 @@ pub(crate) async fn download_oci_image_async( Ok(()) } -async fn try_download_oci_image( +fn try_download_oci_image( client: &Client, registry: &str, repository: &str, @@ -114,7 +93,7 @@ async fn try_download_oci_image( progress: &dyn Progress, ) -> anyhow::Result { // Get authentication token for the registry - let token = get_auth_token(client, registry, repository).await?; + let token = get_auth_token(client, registry, repository)?; // Construct manifest URL based on registry let manifest_url = if registry == "docker.io" { @@ -135,7 +114,6 @@ async fn try_download_oci_image( .header("Accept", "application/vnd.docker.distribution.manifest.list.v2+json, application/vnd.docker.distribution.manifest.v2+json, application/vnd.oci.image.index.v1+json, application/vnd.oci.image.manifest.v1+json") .header("Authorization", format!("Bearer {}", token)) .send() - .await .context("Failed to get image manifest")?; if !response.status().is_success() { @@ -156,7 +134,7 @@ async fn try_download_oci_image( .map(|s| s.to_string()) .unwrap_or_default(); - let body = response.text().await?; + let body = response.text()?; let manifest: Value = serde_json::from_str(&body).context("Failed to parse manifest JSON")?; // Check if this is a manifest list (multi-arch) @@ -208,7 +186,6 @@ async fn try_download_oci_image( .header("Accept", "application/vnd.docker.distribution.manifest.v2+json, application/vnd.oci.image.manifest.v1+json") .header("Authorization", format!("Bearer {}", token)) .send() - .await .context("Failed to get architecture manifest")?; if !arch_response.status().is_success() { @@ -218,7 +195,7 @@ async fn try_download_oci_image( )); } - let arch_body = arch_response.text().await?; + let arch_body = arch_response.text()?; // The arch manifest was requested by digest: verify the response // matches, since the layer digests read from it are trusted next. @@ -293,8 +270,7 @@ async fn try_download_oci_image( &label, layer_digest, progress, - ) - .await?; + )?; layer_names.push(layer_name); } @@ -348,11 +324,7 @@ async fn try_download_oci_image( Ok(tag_digest) } -async fn get_auth_token( - client: &Client, - registry: &str, - repository: &str, -) -> anyhow::Result { +fn get_auth_token(client: &Client, registry: &str, repository: &str) -> anyhow::Result { // Docker Hub has a well-known, stable auth endpoint. if registry == "docker.io" { let url = format!( @@ -360,7 +332,7 @@ async fn get_auth_token( repository ); log::debug!("Getting auth token from: {}", url); - return fetch_bearer_token(client, &url).await; + return fetch_bearer_token(client, &url); } // For every other registry follow the OCI Distribution Spec §4.2: @@ -371,7 +343,6 @@ async fn get_auth_token( let probe = client .get(&probe_url) .send() - .await .with_context(|| format!("Failed to probe registry at {}", probe_url))?; match probe.status().as_u16() { @@ -412,7 +383,7 @@ async fn get_auth_token( realm, service, repository ); log::debug!("Getting auth token from: {}", token_url); - fetch_bearer_token(client, &token_url).await + fetch_bearer_token(client, &token_url) } /// Parse a `Bearer realm="...",service="..."[,scope="..."]` header value. @@ -452,11 +423,10 @@ fn parse_www_authenticate(header: &str) -> Option<(String, String)> { } /// Fetch a bearer token from a fully-constructed token URL. -async fn fetch_bearer_token(client: &Client, token_url: &str) -> anyhow::Result { +fn fetch_bearer_token(client: &Client, token_url: &str) -> anyhow::Result { let response = client .get(token_url) .send() - .await .context("Failed to request auth token")?; if !response.status().is_success() { @@ -464,7 +434,7 @@ async fn fetch_bearer_token(client: &Client, token_url: &str) -> anyhow::Result< return Ok(String::new()); } - let body = response.text().await?; + let body = response.text()?; let json: Value = serde_json::from_str(&body).context("Failed to parse auth token response")?; // OCI spec uses "token"; Docker also emits "access_token". @@ -475,7 +445,7 @@ async fn fetch_bearer_token(client: &Client, token_url: &str) -> anyhow::Result< .ok_or_else(|| anyhow!("No token field in auth response: {}", body)) } -async fn download_blob_with_auth( +fn download_blob_with_auth( client: &Client, url: &str, dest: &Path, @@ -489,7 +459,7 @@ async fn download_blob_with_auth( request = request.header("Authorization", format!("Bearer {}", token)); } - let response = request.send().await.context("Failed to start download")?; + let mut response = request.send().context("Failed to start download")?; if !response.status().is_success() { return Err(anyhow!( @@ -502,24 +472,21 @@ async fn download_blob_with_auth( progress.begin(label, Unit::Bytes, (total_size > 0).then_some(total_size)); - let mut file = tokio::fs::File::create(dest) - .await - .context("Failed to create destination file")?; + let mut file = std::fs::File::create(dest).context("Failed to create destination file")?; let mut hasher = DownloadHasher::new(); - let mut stream = response.bytes_stream(); - use tokio::io::AsyncWriteExt; - - while let Some(chunk) = stream.next().await { - let chunk = chunk.context("Failed to read chunk")?; - file.write_all(&chunk) - .await - .context("Failed to write chunk")?; - hasher.update(&chunk); - progress.advance(chunk.len() as u64); + let mut buf = [0u8; 64 * 1024]; + loop { + let n = response.read(&mut buf).context("Failed to read chunk")?; + if n == 0 { + break; + } + file.write_all(&buf[..n]).context("Failed to write chunk")?; + hasher.update(&buf[..n]); + progress.advance(n as u64); } - file.flush().await?; + file.flush().context("Failed to flush the layer file")?; progress.finish(); // The blob was addressed by this digest in the manifest; a mismatch means