From 30afb977fff782c375f70bf8cd9cb4f8123bf830 Mon Sep 17 00:00:00 2001 From: Andras Schmelczer Date: Sun, 21 Jun 2026 13:52:42 +0100 Subject: [PATCH] Fix stuck updates --- src/compose/updater.rs | 71 +++++-- src/registry.rs | 349 ++++++++++++++++++++++++++--------- src/scheduler.rs | 17 ++ tests/test_error_handling.rs | 68 +++++++ 4 files changed, 395 insertions(+), 110 deletions(-) diff --git a/src/compose/updater.rs b/src/compose/updater.rs index dec4b05..0b213cd 100644 --- a/src/compose/updater.rs +++ b/src/compose/updater.rs @@ -25,6 +25,11 @@ pub struct ComposeUpdater { pub struct UpdateReport { pub files_found: usize, pub updated_files: Vec, + /// Human-readable failures collected during the run (one per failed service, + /// file, or path). Non-empty means the cycle was degraded but did not abort: + /// every other service and file was still processed and successful updates + /// were still written. + pub errors: Vec, } impl ComposeUpdater { @@ -40,15 +45,29 @@ impl ComposeUpdater { pub async fn update_all_compose_files(&self) -> Result { let mut updated_files = Vec::new(); let mut files_found = 0; + let mut errors = Vec::new(); for compose_path in &self.config.compose_paths { - let compose_files = self.find_compose_files(compose_path)?; + let compose_files = match self.find_compose_files(compose_path) { + Ok(files) => files, + Err(e) => { + errors.push(format!("path {}: {:#}", compose_path.display(), e)); + continue; + } + }; files_found += compose_files.len(); for file_path in compose_files { - let updated = self.update_compose_file(&file_path).await?; - if updated { - updated_files.push(file_path); + // A failure on one file (or one service within it) must not abort + // the rest of the run: collect it and keep going. + match self.update_compose_file(&file_path).await { + Ok((updated, file_errors)) => { + if updated { + updated_files.push(file_path); + } + errors.extend(file_errors); + } + Err(e) => errors.push(format!("file {}: {:#}", file_path, e)), } } } @@ -56,6 +75,7 @@ impl ComposeUpdater { Ok(UpdateReport { files_found, updated_files, + errors, }) } @@ -63,11 +83,16 @@ impl ComposeUpdater { self.parser.parse_file(file_path) } - async fn update_compose_file(&self, file_path: &str) -> Result { + /// Updates every service in a single compose file. Returns whether the file + /// was modified along with a list of per-service failures. A transient + /// registry error on one service is recorded and skipped rather than + /// aborting the file, so unrelated services still get updated and written. + async fn update_compose_file(&self, file_path: &str) -> Result<(bool, Vec)> { info!("Processing compose file: {}", file_path); let compose_file = self.parse_compose_file(file_path)?; let mut updated = false; + let mut errors = Vec::new(); let mut new_content = compose_file.content.clone(); for service in &compose_file.services { @@ -77,15 +102,25 @@ impl ComposeUpdater { } match self.update_service_image(service).await { - Ok(Some(new_image)) => { - new_content = - self.replace_image_in_content(&new_content, service, &new_image)?; - updated = true; - info!( - "Updated {}: {} -> {}", - service.service_name, service.image_ref, new_image - ); - } + Ok(Some(new_image)) => match self.replace_image_in_content( + &new_content, + service, + &new_image, + ) { + Ok(replaced) => { + new_content = replaced; + updated = true; + info!( + "Updated {}: {} -> {}", + service.service_name, service.image_ref, new_image + ); + } + Err(e) => { + let msg = format!("service {}: {:#}", service.service_name, e); + warn!("Failed to update {} in {}", msg, file_path); + errors.push(msg); + } + }, Ok(None) => { debug!( "No update needed for {}: {}", @@ -93,9 +128,9 @@ impl ComposeUpdater { ); } Err(e) => { - return Err( - e.context(format!("Failed to update service {}", service.service_name)) - ); + let msg = format!("service {}: {:#}", service.service_name, e); + warn!("Failed to update {} in {}", msg, file_path); + errors.push(msg); } } } @@ -104,7 +139,7 @@ impl ComposeUpdater { self.write_updated_content(file_path, new_content)?; } - Ok(updated) + Ok((updated, errors)) } fn write_updated_content(&self, file_path: &str, content: String) -> Result<()> { diff --git a/src/registry.rs b/src/registry.rs index 24e9586..f72ef0d 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -2,20 +2,35 @@ use crate::config::{Config, RegistryConfig}; use crate::version::VersionInfo; use anyhow::{anyhow, Result}; use chrono::{DateTime, Utc}; -use reqwest::{Client as HttpClient, Response, StatusCode}; +use reqwest::{header::HeaderMap, Client as HttpClient, Response, StatusCode}; use serde::Deserialize; use std::sync::LazyLock; use std::time::Duration; use tokio::time::sleep; use tracing::{debug, warn}; -const PAGE_SIZE: usize = 500; -const MAX_PAGES: usize = 100; +// Docker Hub silently clamps page_size to 100, so request exactly that. +const DOCKERHUB_PAGE_SIZE: usize = 100; +// ghcr.io / lscr.io (and distribution-spec registries) cap `n` at 1000; a larger +// value is clamped, so 1000 minimises the number of requests (and thus 429s). +const OCI_PAGE_SIZE: usize = 1000; +// Upper bound on tags fetched per image, as a runaway guard only. Tag listings on +// Docker Hub and ghcr.io/lscr.io are NOT ordered by version (Docker Hub's order is +// undefined; ghcr.io/lscr.io are chronological), so we must scan the whole list to +// reliably find the highest version rather than truncate and risk missing it. Real +// repos (e.g. jellyfin/jellyfin at ~13.6k tags) finish far below this; hitting it +// means the result may be incomplete and is logged loudly. +const MAX_TAGS_SCANNED: usize = 50_000; const MAX_RETRY_ATTEMPTS: u32 = 5; const MAX_AUTH_ATTEMPTS: u32 = 2; const INITIAL_RETRY_DELAY_SECS: u64 = 1; const HTTP_TIMEOUT_SECS: u64 = 30; const HTTP_CONNECT_TIMEOUT_SECS: u64 = 10; +// Evict idle keep-alive connections well before registries (Docker Hub, GHCR, +// lscr.io) close them server-side. Reusing a connection the server has already +// dropped is what surfaces as "connection closed before message completed"; +// recycling early makes that race rare, and `fetch_with_retry` covers the rest. +const POOL_IDLE_TIMEOUT_SECS: u64 = 30; #[derive(Debug, Clone)] pub struct ImageRef { @@ -97,6 +112,16 @@ struct DockerHubTag { name: String, } +/// A fully-read HTTP response. The body is read inside the retry scope so that a +/// connection dropped mid-body is retried like any other transport failure; +/// callers therefore receive the body as an owned string rather than a streaming +/// `Response`, while retaining the status and headers they need. +struct FetchResult { + status: StatusCode, + headers: HeaderMap, + body: String, +} + pub struct Client { http_client: HttpClient, config: Config, @@ -107,6 +132,7 @@ impl Client { let http_client = HttpClient::builder() .timeout(Duration::from_secs(HTTP_TIMEOUT_SECS)) .connect_timeout(Duration::from_secs(HTTP_CONNECT_TIMEOUT_SECS)) + .pool_idle_timeout(Duration::from_secs(POOL_IDLE_TIMEOUT_SECS)) .build() .expect("failed to build HTTP client"); @@ -132,11 +158,11 @@ impl Client { None } - async fn request_with_retry( + async fn fetch_with_retry( &self, request_fn: F, operation_name: &str, - ) -> Result + ) -> Result where F: Fn() -> Fut, Fut: std::future::Future>, @@ -145,9 +171,49 @@ impl Client { let mut delay = Duration::from_secs(INITIAL_RETRY_DELAY_SECS); loop { - let response = request_fn().await?; + // Send the request AND read the body as one retryable unit: a + // connection dropped while streaming the body (`is_body`) is just as + // transient as one dropped while sending the request (`is_request`), + // and reading it here lets both be retried. + let fetched = async { + let response = request_fn().await?; + let status = response.status(); + let headers = response.headers().clone(); + let body = response.text().await?; + Ok::<_, reqwest::Error>(FetchResult { + status, + headers, + body, + }) + } + .await; - if response.status() == StatusCode::TOO_MANY_REQUESTS { + let fetched = match fetched { + Ok(fetched) => fetched, + Err(err) => { + attempt += 1; + + // Transient transport failures (connect timeouts, dropped + // keep-alive connections) never reached this retry path + // before and aborted the whole update run. Retry them with + // the same bounded, exponential backoff used for 429s. + if !is_transient_transport_error(&err) || attempt > MAX_RETRY_ATTEMPTS { + return Err(anyhow::Error::new(err) + .context(format!("{operation_name} failed"))); + } + + warn!( + "{} transient transport error ({}), attempt {}/{}, waiting {:?} before retry", + operation_name, err, attempt, MAX_RETRY_ATTEMPTS, delay + ); + + sleep(delay).await; + delay = delay.saturating_mul(2); + continue; + } + }; + + if fetched.status == StatusCode::TOO_MANY_REQUESTS { attempt += 1; if attempt > MAX_RETRY_ATTEMPTS { @@ -158,8 +224,8 @@ impl Client { )); } - let wait_duration = if let Some(retry_after) = response - .headers() + let wait_duration = if let Some(retry_after) = fetched + .headers .get("retry-after") .and_then(|h| h.to_str().ok()) { @@ -177,7 +243,7 @@ impl Client { delay = delay.saturating_mul(2); } else { - return Ok(response); + return Ok(fetched); } } } @@ -199,13 +265,17 @@ impl Client { } } + /// Returns the semver-parseable versions on the page, the URL of the next + /// page (if any), and the raw number of tags on the page (including + /// non-semver tags, used to budget the total scan). fn parse_dockerhub_response( &self, response_text: &str, - ) -> Result<(Vec, Option)> { + ) -> Result<(Vec, Option, usize)> { let dockerhub_response: DockerHubTagsResponse = serde_json::from_str(response_text) .map_err(|e| anyhow!("Failed to parse Docker Hub response: {}", e))?; + let raw_count = dockerhub_response.results.len(); let mut versions = Vec::new(); for tag in dockerhub_response.results { @@ -214,19 +284,29 @@ impl Client { } } - Ok((versions, dockerhub_response.next)) + Ok((versions, dockerhub_response.next, raw_count)) } - fn parse_v2_response(&self, response_text: &str) -> Result> { + /// Returns the semver-parseable versions on the page, the raw last tag (the + /// `last=` cursor for the next page, regardless of whether it parsed as + /// semver), and the raw number of tags on the page. + fn parse_v2_response( + &self, + response_text: &str, + ) -> Result<(Vec, Option, usize)> { debug!("Parsing registry v2 response: {}", response_text); let tags_response: serde_json::Value = serde_json::from_str(response_text) .map_err(|e| anyhow!("Failed to parse registry response: {}", e))?; let mut versions = Vec::new(); + let mut raw_last_tag = None; + let mut raw_count = 0; if let Some(tags) = tags_response.get("tags").and_then(|t| t.as_array()) { for tag in tags { if let Some(tag_str) = tag.as_str() { + raw_count += 1; + raw_last_tag = Some(tag_str.to_string()); if let Some(version_info) = VersionInfo::from_tag(tag_str) { versions.push(version_info); } @@ -234,7 +314,7 @@ impl Client { } } - Ok(versions) + Ok((versions, raw_last_tag, raw_count)) } pub async fn get_available_versions(&self, image_ref: &ImageRef) -> Result> { @@ -248,50 +328,50 @@ impl Client { async fn get_dockerhub_versions(&self, image_ref: &ImageRef) -> Result> { let repo_path = self.build_repository_path(image_ref); let mut results = Vec::new(); - let mut page_count: usize = 0; + let mut tags_scanned: usize = 0; let mut url = format!( - "https://hub.docker.com/v2/repositories/{repo_path}/tags/?page_size={PAGE_SIZE}" + "https://hub.docker.com/v2/repositories/{repo_path}/tags/?page_size={DOCKERHUB_PAGE_SIZE}" ); loop { debug!("Docker Hub API URL: {}", url); let url_clone = url.clone(); - let response = self - .request_with_retry( + let fetched = self + .fetch_with_retry( || async { self.http_client.get(&url_clone).send().await }, "Docker Hub API request", ) .await?; - debug!("Docker Hub response status: {}", response.status()); + debug!("Docker Hub response status: {}", fetched.status); - if !response.status().is_success() { + if !fetched.status.is_success() { return Err(anyhow!( "Docker Hub request failed with status {}: {}", - response.status(), - response.text().await.unwrap_or_default() + fetched.status, + fetched.body )); } - let response_text = response.text().await?; - - let (new_tags, next_page) = self.parse_dockerhub_response(&response_text)?; + let (new_tags, next_page, raw_count) = self.parse_dockerhub_response(&fetched.body)?; + tags_scanned += raw_count; results.extend(new_tags); - page_count += 1; - if page_count >= MAX_PAGES { + if tags_scanned >= MAX_TAGS_SCANNED { warn!( - "Reached maximum page count ({}) for {}, stopping pagination", - MAX_PAGES, image_ref + "Reached the {}-tag scan limit for {} before exhausting the registry; \ + the newest version may have been missed", + MAX_TAGS_SCANNED, image_ref ); break; } - if let Some(next) = next_page { - url = next; - } else { - break; + match next_page { + // Advance only on a non-empty page with a distinct next link; + // stop on the last page, an empty page, or a self-referential link. + Some(next) if raw_count > 0 && next != url => url = next, + _ => break, } } @@ -307,7 +387,7 @@ impl Client { let mut next_url: Option = None; let mut bearer_token: Option = None; let mut auth_attempts: u32 = 0; - let mut page_count: usize = 0; + let mut tags_scanned: usize = 0; loop { let url = if let Some(next) = next_url.take() { @@ -318,7 +398,7 @@ impl Client { } } else { format!( - "{}/v2/{repo_path}/tags/list?n={PAGE_SIZE}{}", + "{}/v2/{repo_path}/tags/list?n={OCI_PAGE_SIZE}{}", registry_config.url, if let Some(ref last) = last_tag { format!("&last={last}") @@ -332,8 +412,8 @@ impl Client { let url_clone = url.clone(); let bearer_token_clone = bearer_token.clone(); - let response = self - .request_with_retry( + let fetched = self + .fetch_with_retry( || async { let mut request_builder = self.http_client.get(&url_clone); if let Some(token) = &bearer_token_clone { @@ -345,70 +425,78 @@ impl Client { ) .await?; - let (new_tags, link_next) = if response.status() == reqwest::StatusCode::UNAUTHORIZED { - if auth_attempts >= MAX_AUTH_ATTEMPTS { - return Err(anyhow!( - "Authentication failed after {} attempts for registry {}", - auth_attempts, - image_ref.registry - )); - } - if let Some(token) = ®istry_config.auth_token { - if let Some(auth_header) = response.headers().get("www-authenticate") { - let auth_str = auth_header.to_str().map_err(|e| { - anyhow!("Invalid WWW-Authenticate header encoding: {}", e) - })?; - bearer_token = self.try_registry_v2_auth(auth_str, token).await?; - auth_attempts += 1; - continue; - } else { + let (new_tags, raw_last_tag, raw_count, link_next) = + if fetched.status == reqwest::StatusCode::UNAUTHORIZED { + if auth_attempts >= MAX_AUTH_ATTEMPTS { return Err(anyhow!( - "Unauthorized request but no WWW-Authenticate header found" + "Authentication failed after {} attempts for registry {}", + auth_attempts, + image_ref.registry )); } + if let Some(token) = ®istry_config.auth_token { + if let Some(auth_header) = fetched.headers.get("www-authenticate") { + let auth_str = auth_header.to_str().map_err(|e| { + anyhow!("Invalid WWW-Authenticate header encoding: {}", e) + })?; + bearer_token = self.try_registry_v2_auth(auth_str, token).await?; + auth_attempts += 1; + continue; + } else { + return Err(anyhow!( + "Unauthorized request but no WWW-Authenticate header found" + )); + } + } else { + return Err(anyhow!("Unauthorized request but no auth token configured")); + } + } else if fetched.status.is_success() { + let link_next = fetched + .headers + .get("link") + .and_then(|h| h.to_str().ok()) + .and_then(|h| self.parse_link_header(h)); + + let (tags, raw_last_tag, raw_count) = self.parse_v2_response(&fetched.body)?; + (tags, raw_last_tag, raw_count, link_next) } else { - return Err(anyhow!("Unauthorized request but no auth token configured")); - } - } else if response.status().is_success() { - let link_next = response - .headers() - .get("link") - .and_then(|h| h.to_str().ok()) - .and_then(|h| self.parse_link_header(h)); + return Err(anyhow!( + "Registry request failed with status {}: {}", + fetched.status, + fetched.body + )); + }; - let response_text = response.text().await?; - let tags = self.parse_v2_response(&response_text)?; - (tags, link_next) - } else { - let status = response.status(); - let error_text = response.text().await.unwrap_or_default(); - return Err(anyhow!( - "Registry request failed with status {}: {}", - status, - error_text - )); - }; + tags_scanned += raw_count; + results.extend(new_tags); - page_count += 1; - if page_count >= MAX_PAGES { + if tags_scanned >= MAX_TAGS_SCANNED { warn!( - "Reached maximum page count ({}) for {}, stopping pagination", - MAX_PAGES, image_ref + "Reached the {}-tag scan limit for {} before exhausting the registry; \ + the newest version may have been missed", + MAX_TAGS_SCANNED, image_ref ); - results.extend(new_tags); break; } - let maybe_last_tag = new_tags.last().map(|v| v.original.clone()); - results.extend(new_tags); - + // Advance via the registry's Link header when present, else the raw + // last-tag cursor. Stop on an empty page, a Link header that points + // back at the current page, or a cursor that did not move. if let Some(next) = link_next { - next_url = Some(next); - } else if let Some(ref last) = maybe_last_tag { - if last_tag.as_ref() == Some(last) { + let resolved = if next.starts_with("http://") || next.starts_with("https://") { + next + } else { + format!("{}{next}", registry_config.url) + }; + if resolved == url { break; } - last_tag = maybe_last_tag; + next_url = Some(resolved); + } else if let Some(last) = raw_last_tag { + if raw_count == 0 || last_tag.as_ref() == Some(&last) { + break; + } + last_tag = Some(last); } else { break; } @@ -432,7 +520,7 @@ impl Client { let auth_url_clone = auth_url.clone(); let token_clone = token.to_string(); let token_response = self - .request_with_retry( + .fetch_with_retry( || async { self.http_client .get(&auth_url_clone) @@ -444,15 +532,16 @@ impl Client { ) .await?; - if !token_response.status().is_success() { + if !token_response.status.is_success() { return Err(anyhow!( "Failed to get token: {} - {}", - token_response.status(), - token_response.text().await.unwrap_or_default() + token_response.status, + token_response.body )); } - let token_json: serde_json::Value = token_response.json().await?; + let token_json: serde_json::Value = serde_json::from_str(&token_response.body) + .map_err(|e| anyhow!("Failed to parse auth token response: {}", e))?; if let Some(registry_token) = token_json.get("token").and_then(|t| t.as_str()) { return Ok(Some(registry_token.to_string())); @@ -484,6 +573,18 @@ impl Client { static AUTH_PARAM_REGEX: LazyLock = LazyLock::new(|| regex::Regex::new(r#"(\w+)="([^"]+)""#).unwrap()); +/// Returns true for transient transport-level failures worth retrying: +/// connection-establishment failures (`tcp connect error: deadline has elapsed`), +/// read/overall timeouts, request-send failures such as hyper's `IncompleteMessage` +/// (`connection closed before message completed`), and body-read failures from a +/// connection dropped mid-stream. Because `fetch_with_retry` reads the body inside +/// the retry scope, `is_body` is meaningful here. Non-transient kinds (`is_decode`, +/// redirect, builder, and HTTP status errors, which arrive as a successful +/// `Response` rather than an `Err`) are not retried. +fn is_transient_transport_error(err: &reqwest::Error) -> bool { + err.is_connect() || err.is_timeout() || err.is_request() || err.is_body() +} + fn extract_auth_param(auth_str: &str, param: &str) -> Result { for caps in AUTH_PARAM_REGEX.captures_iter(auth_str) { if caps.get(1).map(|m| m.as_str()) == Some(param) { @@ -595,6 +696,70 @@ mod tests { assert_eq!(client.build_repository_path(&image_ref), "user/app"); } + #[tokio::test] + async fn test_connect_failure_is_transient() { + // 192.0.2.1 is TEST-NET-1 (RFC 5737): guaranteed non-routable, so the + // connect attempt fails (timeout or unreachable). Either way it must be + // classified as a transient transport error so it gets retried. + let client = HttpClient::builder() + .connect_timeout(Duration::from_millis(50)) + .timeout(Duration::from_millis(200)) + .build() + .unwrap(); + + let err = client + .get("http://192.0.2.1:9/") + .send() + .await + .expect_err("connecting to a non-routable address must fail"); + + assert!( + is_transient_transport_error(&err), + "connect failure should be retryable, got: {err:?}" + ); + } + + #[test] + fn test_parse_v2_response_reports_raw_cursor_and_count() { + let client = Client::new(Config::default()); + + // Mixed semver and non-semver tags. The raw cursor must be the literal + // last tag (even though it is not semver) so pagination advances past it, + // and the raw count must include every tag for the scan budget. + let body = r#"{"name":"linuxserver/radarr","tags":["6.1.0","6.1.1","latest","amd64-nightly"]}"#; + let (versions, raw_last_tag, raw_count) = client.parse_v2_response(body).unwrap(); + + assert_eq!(raw_count, 4); + assert_eq!(raw_last_tag.as_deref(), Some("amd64-nightly")); + let parsed: Vec<_> = versions.iter().map(|v| v.original.as_str()).collect(); + assert_eq!(parsed, vec!["6.1.0", "6.1.1"]); + } + + #[test] + fn test_parse_v2_response_handles_empty_and_null_tags() { + let client = Client::new(Config::default()); + + for body in [r#"{"name":"x","tags":[]}"#, r#"{"name":"x","tags":null}"#] { + let (versions, raw_last_tag, raw_count) = client.parse_v2_response(body).unwrap(); + assert!(versions.is_empty()); + assert_eq!(raw_last_tag, None); + assert_eq!(raw_count, 0, "body {body} should yield zero raw tags"); + } + } + + #[test] + fn test_parse_dockerhub_response_reports_raw_count_and_next() { + let client = Client::new(Config::default()); + + let body = r#"{"count":3,"next":"https://hub.docker.com/v2/repositories/jellyfin/jellyfin/tags/?page=2&page_size=100","results":[{"name":"10.10.7"},{"name":"latest"},{"name":"10.10.6"}]}"#; + let (versions, next, raw_count) = client.parse_dockerhub_response(body).unwrap(); + + assert_eq!(raw_count, 3); + assert!(next.unwrap().contains("page=2")); + let parsed: Vec<_> = versions.iter().map(|v| v.original.as_str()).collect(); + assert_eq!(parsed, vec!["10.10.7", "10.10.6"]); + } + #[test] fn test_parse_link_header() { let config = Config::default(); diff --git a/src/scheduler.rs b/src/scheduler.rs index 8dabc51..9c51044 100644 --- a/src/scheduler.rs +++ b/src/scheduler.rs @@ -83,6 +83,23 @@ impl Scheduler { )); } + // A degraded run (some services/files failed) still applied the + // updates that succeeded, but the operator must see the failure: + // health stays binary and fails closed. + if !report.errors.is_empty() { + for err in &report.errors { + error!("Update error: {}", err); + } + if let Some(health) = &self.health_handle { + health.report_update_failure(); + } + return Err(anyhow!( + "Update cycle completed with {} error(s); {} file(s) updated", + report.errors.len(), + report.updated_files.len() + )); + } + if report.updated_files.is_empty() { info!( "Update cycle completed in {:?} - scanned {} files, none updated", diff --git a/tests/test_error_handling.rs b/tests/test_error_handling.rs index 5045000..f633aa2 100644 --- a/tests/test_error_handling.rs +++ b/tests/test_error_handling.rs @@ -172,6 +172,74 @@ async fn test_compose_update_with_empty_paths() { let result = updater.update_all_compose_files().await.unwrap(); assert_eq!(result.files_found, 0); assert!(result.updated_files.is_empty()); + assert!(result.errors.is_empty()); +} + +#[tokio::test] +async fn test_update_continues_after_service_failure() { + // Two services point at registries absent from the config, so each fails + // fast with "Unknown registry" (no network). With an empty registry map and + // a non-semver tag, the third service is skipped without error. This proves + // a failure on the first service no longer aborts the run before the second. + let temp_dir = TempDir::new().unwrap(); + let compose_path = temp_dir.path().join("stack.yml"); + std::fs::write( + &compose_path, + "services:\n \ + skipped:\n image: nginx:latest\n \ + bad_one:\n image: unknown.registry.example/foo/app:1.0.0\n \ + bad_two:\n image: another.unknown.example/bar/app:2.0.0\n", + ) + .unwrap(); + + let config = Config { + compose_paths: vec![temp_dir.path().to_path_buf()], + schedule: "0 0 2 * * *".to_string(), + registries: HashMap::new(), + update_strategy: UpdateStrategy::Latest, + ignore_images: vec![], + dry_run: true, + }; + + let updater = ComposeUpdater::new(config); + let report = updater.update_all_compose_files().await.unwrap(); + + assert_eq!(report.files_found, 1); + assert!(report.updated_files.is_empty()); + assert_eq!( + report.errors.len(), + 2, + "both failing services should be attempted and recorded, got: {:?}", + report.errors + ); +} + +#[tokio::test] +async fn test_scheduler_reports_failure_on_service_error() { + // A degraded cycle (a service that cannot be resolved) must surface as an + // error from run_once even though the run itself did not abort. + let temp_dir = TempDir::new().unwrap(); + let compose_path = temp_dir.path().join("stack.yml"); + std::fs::write( + &compose_path, + "services:\n bad:\n image: unknown.registry.example/foo/app:1.0.0\n", + ) + .unwrap(); + + let config = Config { + compose_paths: vec![temp_dir.path().to_path_buf()], + schedule: "0 0 2 * * *".to_string(), + registries: HashMap::new(), + update_strategy: UpdateStrategy::Latest, + ignore_images: vec![], + dry_run: true, + }; + + let scheduler = Scheduler::new(config, None).unwrap(); + assert!( + scheduler.run_once().await.is_err(), + "a cycle where a service fails to update should report an error" + ); } #[test]