diff --git a/README.md b/README.md index 9015bb8b..53e8c1e8 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ Sync OCI container images across registries - efficiently.

- ocync - 4x faster sync, 30% fewer API requests, adaptive rate control + ocync - 4x faster sync, 30% fewer API requests, adaptive rate control

[![CI](https://github.com/clowdhaus/ocync/actions/workflows/ci.yml/badge.svg)](https://github.com/clowdhaus/ocync/actions/workflows/ci.yml) diff --git a/crates/ocync-distribution/src/auth/docker.rs b/crates/ocync-distribution/src/auth/docker.rs index 46e401f2..ace7aca2 100644 --- a/crates/ocync-distribution/src/auth/docker.rs +++ b/crates/ocync-distribution/src/auth/docker.rs @@ -238,7 +238,7 @@ async fn run_credential_helper(helper: &str, registry: &str) -> Result Result Result Result) -> TargetCh while let Some((target_name, target_client, batch_checker, result)) = head_checks.next().await { match result { Ok(Some(head)) if head.digest == *compare_digest => { - info!( + debug!( source_repo = %source.repo, source_tag = %source.tag, target_repo = %target.repo, @@ -1574,7 +1574,7 @@ async fn full_pull_and_build_tasks(params: FullPullParams<'_>) -> DiscoveryOutco while let Some((target_name, target_client, batch_checker, result)) = head_checks.next().await { match result { Ok(Some(head)) if head.digest == *source_digest => { - info!( + debug!( source_repo = %source.repo, source_tag = %source.tag, target_repo = %target.repo, @@ -1763,7 +1763,7 @@ async fn execute_item( .notify_repo_failed(&item.target_name, &item.target.repo); if is_immutable_tag_error(&err) { - info!( + debug!( source_repo = %item.source.repo, target_repo = %item.target.repo, target_tag = %item.target.tag, @@ -2868,7 +2868,7 @@ async fn discover_referrers( } }, Err(e) if e.is_not_found() => { - info!( + debug!( repo = %source_repo, digest = %parent_digest, "no referrers found (API 404, tag fallback 404)" @@ -2877,7 +2877,7 @@ async fn discover_referrers( } Err(e) => { // Non-404 error on fallback is not fatal; log and continue. - info!( + debug!( repo = %source_repo, digest = %parent_digest, error = %e, @@ -2891,7 +2891,7 @@ async fn discover_referrers( Err(e) => { // Non-404 error from referrers API. Log and continue rather // than failing the entire image sync for an artifact query. - info!( + debug!( repo = %source_repo, digest = %parent_digest, error = %e, diff --git a/crates/ocync-sync/src/filter.rs b/crates/ocync-sync/src/filter.rs index 1acaa97c..ed1a63ff 100644 --- a/crates/ocync-sync/src/filter.rs +++ b/crates/ocync-sync/src/filter.rs @@ -6,7 +6,7 @@ use std::sync::OnceLock; use globset::{Glob, GlobBuilder, GlobSet, GlobSetBuilder}; use serde::{Deserialize, Serialize}; -use tracing::warn; +use tracing::debug; use crate::Error; @@ -111,6 +111,48 @@ impl FilterConfig { self.run_pipeline(tags, true) } + /// One-line summary of the active filter clauses, e.g. + /// `semver >=1.0.0, latest=5`. Returns `None` when no filtering applies. + /// + /// Sole formatter for filter rationale shown in non-dry-run logs; uses + /// the same `FilterConfig` fields the pipeline operates on so adding a + /// new field to the config will fail tests here before it ships. + pub fn describe(&self) -> Option { + let mut parts = Vec::new(); + if !self.glob.is_empty() { + parts.push(format!("glob {}", self.glob.join(","))); + } + if let Some(ref s) = self.semver { + parts.push(format!("semver {s}")); + } + if !self.exclude.is_empty() { + parts.push(format!("exclude {}", self.exclude.join(","))); + } + if !self.include.is_empty() { + parts.push(format!("include {}", self.include.join(","))); + } + if let Some(order) = self.sort { + parts.push(format!( + "sort {}", + match order { + SortOrder::Semver => "semver", + SortOrder::Alpha => "alpha", + } + )); + } + if let Some(n) = self.latest { + parts.push(format!("latest={n}")); + } + if let Some(n) = self.min_tags { + parts.push(format!("min_tags={n}")); + } + if parts.is_empty() { + None + } else { + Some(parts.join(", ")) + } + } + /// Shared pipeline implementation. When `track` is false (the real-sync /// hot path), per-stage `StageDelta` and per-reason `DropReason` are not /// constructed; the resulting `Filtered.report` carries empty vectors. @@ -171,18 +213,18 @@ impl FilterConfig { range: range.to_owned(), reason: e.to_string(), })?; - let (kept, dropped) = - partition_with_drop( - &pipeline, - track, - |t| match crate::version::TagVersion::parse(t) { - Some(ver) => req.matches(&ver), - None => { - warn!(tag = t, "tag is not parseable as a version, dropping"); - false - } - }, - ); + let (kept, dropped) = partition_with_drop(&pipeline, track, |t| { + if is_referrers_fallback_tag(t) { + return false; + } + match crate::version::TagVersion::parse(t) { + Some(ver) => req.matches(&ver), + None => { + debug!(tag = t, "tag is not parseable as a version, dropping"); + false + } + } + }); push_drop_reason( &mut drop_reasons, track, @@ -248,7 +290,19 @@ impl FilterConfig { } if let Some(order) = self.sort { + let before = pipeline.len(); sort_tags_in_place(&mut pipeline, order); + if track { + let label = match order { + SortOrder::Semver => "sort semver desc", + SortOrder::Alpha => "sort alpha desc", + }; + pipeline_stages.push(StageDelta { + label: label.to_string(), + count_in: before, + count_out: pipeline.len(), + }); + } } if let Some(n) = self.latest { let before = pipeline.len(); @@ -265,13 +319,8 @@ impl FilterConfig { pipeline.truncate(n); } if track { - let label = match self.sort { - Some(SortOrder::Semver) => format!("sort semver desc, latest {n}"), - Some(SortOrder::Alpha) => format!("sort alpha desc, latest {n}"), - None => format!("latest {n}"), - }; pipeline_stages.push(StageDelta { - label, + label: format!("keep latest {n}"), count_in: before, count_out: pipeline.len(), }); @@ -477,6 +526,18 @@ fn push_drop_reason( // Individual stages // --------------------------------------------------------------------------- +/// True for OCI 1.1 referrers fallback tags (`-` and the cosign +/// `.sig`/`.sbom`/`.att` variants). These are pointers to artifacts, not image +/// versions, and will never satisfy a semver range -- skip parsing so they do +/// not appear in the unparseable-tag log channel. +/// +/// Public so that observability/UX code outside the filter pipeline (e.g. the +/// CLI's "no tags matched" warn) can partition source tag lists without +/// reintroducing a duplicate prefix check that would drift over time. +pub fn is_referrers_fallback_tag(tag: &str) -> bool { + tag.starts_with("sha256-") || tag.starts_with("sha512-") +} + /// Build a [`GlobSet`] from patterns, returning an error on invalid patterns. pub fn build_glob_set(patterns: &[String]) -> Result { let mut builder = GlobSetBuilder::new(); @@ -520,8 +581,11 @@ fn filter_semver<'a>(tags: &[&'a str], range: &str) -> Result, Erro .iter() .copied() .filter(|tag| { + if is_referrers_fallback_tag(tag) { + return false; + } let Some(ver) = crate::version::TagVersion::parse(tag) else { - warn!(tag, "tag is not parseable as a version, dropping"); + debug!(tag, "tag is not parseable as a version, dropping"); return false; }; req.matches(&ver) @@ -728,6 +792,24 @@ mod tests { assert_eq!(result, vec!["1.0.0"]); } + /// Referrers fallback tags (cosign signatures, SBOMs, attestations) bypass + /// the version parser. They drop silently so noisy unparseable-tag logs do + /// not fire once per artifact tag per image. + #[test] + fn semver_skips_referrers_fallback_tags() { + let tags = vec![ + "1.0.0", + "sha256-abc123def456.sig", + "sha256-abc123def456.sbom", + "sha256-abc123def456.att", + "sha256-abc123def456", + "sha512-deadbeef.sig", + "2.0.0", + ]; + let result = filter_semver(&tags, ">=1.0.0").unwrap(); + assert_eq!(result, vec!["1.0.0", "2.0.0"]); + } + // - pipeline tests ---------------------------------------------------- #[test] @@ -1378,4 +1460,68 @@ mod tests { assert_eq!(reason.count, reason.samples.len()); } } + + // - describe ---------------------------------------------------------- + + #[test] + fn describe_default_returns_none() { + assert!(FilterConfig::default().describe().is_none()); + } + + #[test] + fn describe_combines_clauses_in_pipeline_order() { + let config = FilterConfig { + glob: vec!["1.*".into()], + semver: Some(">=1.0.0".into()), + exclude: vec!["*-rc*".into()], + sort: Some(SortOrder::Semver), + latest: Some(5), + min_tags: Some(1), + ..FilterConfig::default() + }; + assert_eq!( + config.describe().as_deref(), + Some("glob 1.*, semver >=1.0.0, exclude *-rc*, sort semver, latest=5, min_tags=1") + ); + } + + /// Regression guard: every `FilterConfig` field that influences selection + /// must contribute to `describe()`. Adding a new field without updating + /// `describe()` would silently leave INFO-line filter rationale stale. + #[test] + fn describe_covers_every_selection_field() { + let cfg = FilterConfig { + include: vec!["latest".into()], + glob: vec!["v*".into()], + semver: Some("^1".into()), + exclude: vec!["nightly".into()], + sort: Some(SortOrder::Alpha), + latest: Some(3), + min_tags: Some(2), + }; + let desc = cfg.describe().expect("non-empty config describes"); + for needle in [ + "include latest", + "glob v*", + "semver ^1", + "exclude nightly", + "sort alpha", + "latest=3", + "min_tags=2", + ] { + assert!(desc.contains(needle), "missing {needle:?} in {desc:?}"); + } + } + + // - is_referrers_fallback_tag ---------------------------------------- + + #[test] + fn referrers_fallback_detection() { + assert!(is_referrers_fallback_tag("sha256-abcdef")); + assert!(is_referrers_fallback_tag("sha256-deadbeef.sig")); + assert!(is_referrers_fallback_tag("sha512-cafef00d.sbom")); + assert!(!is_referrers_fallback_tag("v1.0.0")); + assert!(!is_referrers_fallback_tag("latest")); + assert!(!is_referrers_fallback_tag("sha256")); // no dash + } } diff --git a/crates/ocync-sync/src/lib.rs b/crates/ocync-sync/src/lib.rs index 3cf88e26..579cf6be 100644 --- a/crates/ocync-sync/src/lib.rs +++ b/crates/ocync-sync/src/lib.rs @@ -184,7 +184,7 @@ impl std::fmt::Display for ErrorKind { } /// Aggregate statistics for a sync run. -#[derive(Debug, Default, Serialize)] +#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize)] pub struct SyncStats { /// Number of images successfully synced. pub images_synced: u64, diff --git a/docs/public/ecr-banner.png b/docs/public/ecr-banner.png new file mode 100644 index 00000000..bf47edf9 Binary files /dev/null and b/docs/public/ecr-banner.png differ diff --git a/docs/src/content/cli-reference.md b/docs/src/content/cli-reference.md index c96e076e..f54dd58f 100644 --- a/docs/src/content/cli-reference.md +++ b/docs/src/content/cli-reference.md @@ -48,11 +48,11 @@ ocync sync -c config.yaml --json `--dry-run` runs the full filter pipeline against each mapping's source tags and prints, per mapping: -- **`source candidates: N`** -- the number of tags fetched from the source. +- **`source tags: N`** -- the number of tags fetched from the source. - **`include path:`** -- tags rescued via `include:` (bypasses `glob:`/`semver:` and the system-exclude defaults). Default cap is 5 names; `-v` removes the cap. -- **`pipeline:`** -- per-stage attrition (`glob`, `semver`, `exclude`, `latest`). Each row shows count_in -> count_out and the drop count. +- **`filter:`** -- per-stage attrition (`glob`, `semver`, `exclude`, `sort`, `keep latest`). Each row shows count_in -> count_out and the drop count. - **`kept (N):`** -- the final tags. When `include:` is used, rescued tags are listed first and tagged `[via include]` so the rescue path is visible. -- **`dropped N:`** -- Pareto-sorted drop attribution (largest cause first), with sample tag names per reason. Default cap is 5 names per reason; `-v` removes the cap. +- **`dropped (N):`** -- Pareto-sorted drop attribution (largest cause first), with sample tag names per reason. Default cap is 5 names per reason; `-v` removes the cap. - **`min_tags: N`** -- when `min_tags:` is configured, the line prints `kept M, satisfied` or `kept M, real sync will FAIL with BelowMinTags`. Real-sync (no `--dry-run`) errors out below `min_tags`; dry-run shows the report and surfaces the gap so the configuration can be fixed before running. ## copy diff --git a/src/cli/commands/analyze.rs b/src/cli/commands/analyze.rs index d5fdd09f..c0974b9a 100644 --- a/src/cli/commands/analyze.rs +++ b/src/cli/commands/analyze.rs @@ -18,7 +18,7 @@ use ocync_distribution::{Digest, RepositoryName}; use ocync_sync::ShutdownSignal; -use crate::cli::commands::synchronize::{build_clients, resolve_mapping}; +use crate::cli::commands::synchronize::{MappingResolution, build_clients, resolve_mapping}; use crate::cli::config::load_config; use crate::cli::output::format_bytes; use crate::cli::{CliError, ExitCode}; @@ -68,8 +68,8 @@ pub(crate) async fn run( let resolved = match resolve_mapping(mapping, &config, &clients, &no_checkers, false).await? { - Some(r) => r, - None => continue, + MappingResolution::Resolved(r) => r, + MappingResolution::NoMatchingTags(_) => continue, }; for tag_pair in &resolved.tags { diff --git a/src/cli/commands/dry_run.rs b/src/cli/commands/dry_run.rs index d807cb82..69bd83fa 100644 --- a/src/cli/commands/dry_run.rs +++ b/src/cli/commands/dry_run.rs @@ -49,7 +49,7 @@ fn write_mapping(w: &mut W, m: &ResolvedMapping, verbose: bool) -> io: let target_names: Vec<&str> = m.targets.iter().map(|t| &*t.name).collect(); writeln!( w, - "dry-run: {} -> {} => [{}]", + "dry-run: {} -> {} [{}]", m.source_repo, m.target_repo, target_names.join(", ") @@ -61,7 +61,7 @@ fn write_mapping(w: &mut W, m: &ResolvedMapping, verbose: bool) -> io: return write_simple_tag_list(w, &m.tags); }; - writeln!(w, " source candidates: {}", report.candidate_count)?; + writeln!(w, " source tags: {}", report.candidate_count)?; writeln!(w)?; if !report.include_kept.is_empty() { @@ -107,7 +107,7 @@ fn write_include_path(w: &mut W, report: &FilterReport, verbose: bool) } fn write_pipeline(w: &mut W, report: &FilterReport) -> io::Result<()> { - writeln!(w, " pipeline:")?; + writeln!(w, " filter:")?; for stage in &report.pipeline { let delta = stage.count_in as isize - stage.count_out as isize; let delta_str = if delta != 0 { @@ -115,11 +115,11 @@ fn write_pipeline(w: &mut W, report: &FilterReport) -> io::Result<()> } else { String::new() }; - writeln!( - w, + let line = format!( " {:<28} {:>4} -> {:<4}{}", stage.label, stage.count_in, stage.count_out, delta_str - )?; + ); + writeln!(w, "{}", line.trim_end())?; } Ok(()) } @@ -150,7 +150,7 @@ fn write_dropped(w: &mut W, report: &FilterReport, verbose: bool) -> i if total == 0 { return Ok(false); } - writeln!(w, " dropped {total}:")?; + writeln!(w, " dropped ({total}):")?; for reason in &report.dropped { let samples_display = render_samples(&reason.samples, verbose); // `LatestCap` reads as a complete clause ("over latest=N limit"); every @@ -167,8 +167,7 @@ fn write_dropped(w: &mut W, report: &FilterReport, verbose: bool) -> i if matches!(reason.kind, DropKind::SystemExclude) { writeln!( w, - " {:<28}to keep prereleases, list patterns under include: (globs supported)", - "" + " hint: to keep prereleases, list patterns under include: (globs supported)" )?; } } @@ -394,8 +393,8 @@ mod tests { assert!(out.contains("tags (2):"), "{out}"); assert!(out.contains(" v1.0.0\n"), "{out}"); assert!(out.contains(" v1.1.0\n"), "{out}"); - // No pipeline/kept/dropped sections appear. - assert!(!out.contains("pipeline:"), "{out}"); + // No filter/kept/dropped sections appear. + assert!(!out.contains("filter:"), "{out}"); assert!(!out.contains("kept ("), "{out}"); assert!(!out.contains("dropped"), "{out}"); } diff --git a/src/cli/commands/synchronize.rs b/src/cli/commands/synchronize.rs index 498c05b3..210c87b7 100644 --- a/src/cli/commands/synchronize.rs +++ b/src/cli/commands/synchronize.rs @@ -16,7 +16,7 @@ use ocync_sync::engine::{ DEFAULT_MAX_CONCURRENT_TRANSFERS, RegistryAlias, ResolvedArtifacts, ResolvedMapping, SyncEngine, TagPair, TargetEntry, }; -use ocync_sync::filter::{FilterConfig, build_glob_set}; +use ocync_sync::filter::{FilterConfig, build_glob_set, is_referrers_fallback_tag}; use ocync_sync::retry::RetryConfig; use ocync_sync::shutdown::ShutdownSignal; use ocync_sync::staging::BlobStage; @@ -25,6 +25,7 @@ use crate::SyncArgs; use crate::cli::config::{ AuthType, Config, GlobOrList, MappingConfig, TagsConfig, load_config, resolve_target_names, }; +use crate::cli::output::{format_bytes, format_duration}; use crate::cli::{CliError, ExitCode, bare_hostname, build_registry_client}; /// Default cache TTL: 12 hours. @@ -33,6 +34,174 @@ pub(crate) const DEFAULT_CACHE_TTL: Duration = Duration::from_secs(12 * 3600); /// Default cache file name within the cache directory. const CACHE_FILE_NAME: &str = "transfer_state.bin"; +/// Sample cap for the source-tag list shown in the no-tags-matched WARN. +/// Mirrors `dry_run::SAMPLE_CAP` so both surfaces show the same depth of +/// example data without overwhelming the log line. +const NO_TAGS_SAMPLE_CAP: usize = 5; + +/// Outcome of resolving a single mapping. Either the mapping is ready for the +/// engine, or no source tag survived filtering and the caller decides whether +/// to log a WARN (sync mode: always; watch mode: only on transition). +/// +/// The size disparity between variants is intentional: `ResolvedMapping` flows +/// directly into `Vec` for the engine, so boxing it would +/// just add a heap round-trip per success. The error variant is rare; we pay +/// the disparity instead of the allocation traffic. +#[allow(clippy::large_enum_variant)] +pub(crate) enum MappingResolution { + Resolved(ResolvedMapping), + NoMatchingTags(NoTagsInfo), +} + +/// Diagnostic context for a mapping whose filter rejected every source tag. +/// +/// Fields together let an operator see, in one log line, the size and +/// composition of the source repo (image tags vs OCI 1.1 referrer fallbacks), +/// the active filter clauses, and example image tag names so the cause is +/// obvious without spelunking. +pub(crate) struct NoTagsInfo { + pub from: String, + pub image_count: usize, + pub artifact_count: usize, + /// Active filter clauses (e.g. `semver >=1.0.0, latest=5`). `None` only + /// when no filter is configured -- distinct from "filter description + /// missing" so the formatter can render an explicit fallback string. + pub filter_desc: Option, + /// Up to [`NO_TAGS_SAMPLE_CAP`] image-tag names. Excludes referrer + /// fallback tags so the example list is meaningful on cosign-heavy + /// repos like `cgr.dev/chainguard/*` (otherwise dominated by + /// `sha256-(.sig|.sbom|.att)` entries). + pub samples: Vec, +} + +impl NoTagsInfo { + /// Total tags returned by `/v2//tags/list`. Derived: image + artifact. + fn source_total(&self) -> usize { + self.image_count + self.artifact_count + } + + /// True when the source had more image tags than `samples` shows. + fn samples_truncated(&self) -> bool { + self.image_count > self.samples.len() + } +} + +impl std::fmt::Display for NoTagsInfo { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let total = self.source_total(); + let total_phrase = if self.artifact_count > 0 { + format!( + "{total} source tags ({} image tags, {} referrer artifacts)", + self.image_count, self.artifact_count + ) + } else { + format!("{total} source tags") + }; + let filter = self + .filter_desc + .as_deref() + .unwrap_or("no filter configured"); + let samples = if self.samples.is_empty() { + "".to_string() + } else if self.samples_truncated() { + format!("[{}, ...]", self.samples.join(", ")) + } else { + format!("[{}]", self.samples.join(", ")) + }; + write!( + f, + "{}: 0 of {total_phrase} matched filter ({filter}); skipping. Source: {samples}", + self.from + ) + } +} + +/// Per-process state that lets watch-mode log on transitions instead of +/// every cycle. Sync mode passes `None`. State lives in `watch::run` so it +/// spans loop iterations. +/// +/// Tracks three pieces of cross-cycle context: +/// +/// 1. No-tags failure set: mappings whose filter rejected every source tag. +/// 2. Per-mapping outcomes: the prior cycle's [`MappingOutcome`] keyed by +/// `mapping.from`, used to detect both repeated and recovery transitions. +/// 3. Per-cycle emit counter: bumped by every `observe_*` method that +/// reports a transition; the watch loop reads it to gate the idle +/// heartbeat. +#[derive(Debug, Default)] +pub(crate) struct WatchLogState { + warned_no_tags: HashSet, + last_outcomes: HashMap, + cycle_emit_count: u32, +} + +impl WatchLogState { + pub(crate) fn begin_cycle(&mut self) { + self.cycle_emit_count = 0; + } + + pub(crate) fn cycle_emit_count(&self) -> u32 { + self.cycle_emit_count + } + + /// Record a no-match observation. Returns `true` on transition into the + /// failure state (caller emits a WARN); `false` when already failing. + fn observe_no_match(&mut self, from: &str) -> bool { + let changed = self.warned_no_tags.insert(from.to_string()); + if changed { + self.cycle_emit_count = self.cycle_emit_count.saturating_add(1); + } + changed + } + + /// Record a successful resolution. Returns `true` when the mapping was + /// previously in the failure set (caller emits a recovery INFO). + fn observe_resolved(&mut self, from: &str) -> bool { + let changed = self.warned_no_tags.remove(from); + if changed { + self.cycle_emit_count = self.cycle_emit_count.saturating_add(1); + } + changed + } + + /// Record `outcome` as the latest result for `from`. + /// + /// Returns: + /// - `None` when the outcome is identical to the prior cycle (suppress). + /// - `Some(false)` on a non-recovery transition (emit normally). + /// - `Some(true)` when transitioning from `failed > 0` to `failed == 0` + /// (emit with `[recovered]` marker). + fn observe_mapping_outcome(&mut self, from: &str, outcome: &MappingOutcome) -> Option { + use std::collections::hash_map::Entry; + match self.last_outcomes.entry(from.to_string()) { + Entry::Occupied(mut slot) => { + let prev = *slot.get(); + if &prev == outcome { + return None; + } + slot.insert(*outcome); + self.cycle_emit_count = self.cycle_emit_count.saturating_add(1); + Some(prev.failed > 0 && outcome.failed == 0) + } + Entry::Vacant(slot) => { + slot.insert(*outcome); + self.cycle_emit_count = self.cycle_emit_count.saturating_add(1); + Some(false) + } + } + } + + /// Drop entries for mappings no longer in the active set so the state + /// does not grow unbounded across edits to the config. + fn retain_active<'a>(&mut self, active: impl IntoIterator) { + let active_set: HashSet<&str> = active.into_iter().collect(); + self.warned_no_tags + .retain(|k| active_set.contains(k.as_str())); + self.last_outcomes + .retain(|k, _| active_set.contains(k.as_str())); + } +} + /// Resolve the cache directory and file path from config. /// /// Uses `global.cache_dir` if configured, otherwise places the cache @@ -78,6 +247,7 @@ pub(crate) async fn run( shutdown: Option<&ShutdownSignal>, external_cache: Option>>, verbose: bool, + mut watch_log: Option<&mut WatchLogState>, ) -> Result { let config = load_config(&args.config)?; @@ -86,14 +256,32 @@ pub(crate) async fn run( let mut mappings = Vec::new(); for mapping in &config.mappings { - match resolve_mapping(mapping, &config, &clients, &batch_checkers, args.dry_run).await { - Ok(Some(resolved)) => mappings.push(resolved), - Ok(None) => {} // no tags after filtering, logged inside - Err(err) => return Err(err), + match resolve_mapping(mapping, &config, &clients, &batch_checkers, args.dry_run).await? { + MappingResolution::Resolved(resolved) => mappings.push(resolved), + MappingResolution::NoMatchingTags(info) => { + let should_warn = match watch_log.as_mut() { + Some(state) => state.observe_no_match(&info.from), + None => true, + }; + if should_warn { + emit_no_tags_warn(&info); + } + } } } - log_resolved_mappings(&mappings); + if let Some(state) = watch_log.as_mut() { + for resolved in &mappings { + let from = resolved.source_repo.as_str(); + if state.observe_resolved(from) { + tracing::info!( + from = %from, + "{from}: filter now matches at least one tag; resuming sync" + ); + } + } + state.retain_active(config.mappings.iter().map(|m| m.from.as_str())); + } if args.dry_run { crate::cli::commands::dry_run::print(&mappings, verbose); @@ -157,6 +345,18 @@ pub(crate) async fn run( .map_or(DEFAULT_MAX_CONCURRENT_TRANSFERS, |g| { g.max_concurrent_transfers }); + // Capture per-mapping metadata before the engine consumes `mappings`. + // Used to emit one INFO line per mapping after the engine returns, + // grouped from the report's per-image outcomes. + let descriptors: Vec = mappings + .iter() + .map(|m| MappingDescriptor { + from: m.source_repo.as_str().to_string(), + target_repo: m.target_repo.as_str().to_string(), + target_names: m.targets.iter().map(|t| (*t.name).to_string()).collect(), + }) + .collect(); + let engine = SyncEngine::new(RetryConfig::default(), max_concurrent); let report = engine .run(mappings, cache.clone(), staging, progress, shutdown) @@ -169,11 +369,180 @@ pub(crate) async fn run( } } + emit_mapping_outcomes(&descriptors, &report, watch_log.as_deref_mut()); + // Watch mode: suppress the cycle tail when no per-mapping line emitted + // (steady-state idle); sync mode: always emit as the final marker. + let cycle_had_activity = watch_log + .as_deref() + .is_none_or(|s| s.cycle_emit_count() > 0); + if cycle_had_activity { + emit_cycle_tail(&descriptors, &report); + } + write_output(&report, args.json)?; Ok(ExitCode::from_report(report.exit_code())) } +/// Per-mapping metadata captured before the engine consumes `mappings`, +/// so we can join it with the engine's per-image report after the fact +/// to emit one log line per mapping (with source/target context). +struct MappingDescriptor { + from: String, + target_repo: String, + target_names: Vec, +} + +/// Per-mapping aggregated outcome derived from [`SyncReport.images`]. +/// Used for log emission and watch-mode change detection. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub(crate) struct MappingOutcome { + pub synced: u64, + pub skipped: u64, + pub failed: u64, + pub bytes: u64, +} + +impl MappingOutcome { + fn is_empty(&self) -> bool { + self.synced == 0 && self.skipped == 0 && self.failed == 0 + } +} + +/// Emit one INFO (or WARN, on failures) per mapping summarizing what its +/// configured tags did this cycle. In watch mode (when `watch_log` is +/// `Some`), suppress mappings whose outcome is unchanged from the prior +/// cycle so steady-state pods log only on transition. +fn emit_mapping_outcomes( + descriptors: &[MappingDescriptor], + report: &SyncReport, + mut watch_log: Option<&mut WatchLogState>, +) { + for d in descriptors { + let outcome = aggregate_mapping_outcome(&d.from, &d.target_repo, report); + // No images for this mapping in the report (e.g. the mapping was + // resolved to zero tags by an upstream filter that the engine + // never saw). The no-tags WARN already covered it; skip here. + if outcome.is_empty() { + continue; + } + let recovered = match watch_log.as_deref_mut() { + Some(state) => match state.observe_mapping_outcome(&d.from, &outcome) { + Some(r) => r, + None => continue, + }, + None => false, + }; + let line = format_mapping_outcome(d, &outcome, recovered); + // `from` / `to` are intentionally NOT structured fields here -- the + // message already names them as `from -> to`, and tracing's text + // formatter would otherwise tail the line with a redundant + // `from=... to=...` block. The count fields remain because they + // carry zero values the terse message elides. + // + // The two arms differ only in level. `tracing::event!` would let + // us pick at runtime, but it requires a const-expression level. + if outcome.failed > 0 { + tracing::warn!( + synced = outcome.synced, + skipped = outcome.skipped, + failed = outcome.failed, + bytes = outcome.bytes, + recovered, + "{line}" + ); + } else { + tracing::info!( + synced = outcome.synced, + skipped = outcome.skipped, + failed = outcome.failed, + bytes = outcome.bytes, + recovered, + "{line}" + ); + } + } +} + +fn aggregate_mapping_outcome( + source_repo: &str, + target_repo: &str, + report: &SyncReport, +) -> MappingOutcome { + let src_prefix = format!("{source_repo}:"); + let tgt_prefix = format!("{target_repo}:"); + let mut o = MappingOutcome::default(); + for r in &report.images { + if !(r.source.starts_with(&src_prefix) && r.target.starts_with(&tgt_prefix)) { + continue; + } + match r.status { + ocync_sync::ImageStatus::Synced => { + o.synced += 1; + o.bytes += r.bytes_transferred; + } + ocync_sync::ImageStatus::Skipped { .. } => o.skipped += 1, + ocync_sync::ImageStatus::Failed { .. } => o.failed += 1, + } + } + o +} + +fn format_mapping_outcome(d: &MappingDescriptor, o: &MappingOutcome, recovered: bool) -> String { + let mut parts = Vec::with_capacity(3); + if o.synced > 0 { + parts.push(format!("synced {}", o.synced)); + } + if o.skipped > 0 { + parts.push(format!("skipped {}", o.skipped)); + } + if o.failed > 0 { + parts.push(format!("failed {}", o.failed)); + } + let counts = parts.join(", "); + let bytes_clause = if o.bytes > 0 { + format!(" ({})", format_bytes(o.bytes)) + } else { + String::new() + }; + let recovered_clause = if recovered { " [recovered]" } else { "" }; + // Multi-target mappings need the bracket to disambiguate which targets + // the line refers to. Single-target mappings: omit -- the destination + // is already in the `from -> to` arrow. + let targets_clause = if d.target_names.len() > 1 { + format!(" [{}]", d.target_names.join(", ")) + } else { + String::new() + }; + format!( + "{} -> {}{targets_clause}: {counts}{bytes_clause}{recovered_clause}", + d.from, d.target_repo + ) +} + +/// One-line cycle tail rolling up totals across all mappings. The caller +/// is responsible for gating this in watch mode (skip on idle cycles). +fn emit_cycle_tail(descriptors: &[MappingDescriptor], report: &SyncReport) { + let s = &report.stats; + let line = format!( + "summary: {} mappings | {} synced, {} skipped, {} failed | {} in {}", + descriptors.len(), + s.images_synced, + s.images_skipped, + s.images_failed, + format_bytes(s.bytes_transferred), + format_duration(report.duration), + ); + // Counts are already in the message; structured fields would be a + // verbatim restatement in text output. JSON aggregators parse the + // message (or use the SyncReport via `--json`). + if s.images_failed > 0 { + tracing::warn!("{line}"); + } else { + tracing::info!("{line}"); + } +} + /// Parse a human-readable duration string into a [`Duration`]. /// /// Accepts: @@ -273,18 +642,19 @@ async fn build_batch_checkers( Ok(checkers) } -/// Resolve a single mapping config into a `ResolvedMapping`, or `None` if no -/// tags survive filtering. +/// Resolve a single mapping config into a [`MappingResolution`]. /// -/// Falls back to `defaults.source`, `defaults.targets`, and `defaults.tags` -/// when the mapping does not specify its own values. +/// Returns [`MappingResolution::Resolved`] when at least one tag survives the +/// filter pipeline, or [`MappingResolution::NoMatchingTags`] carrying the +/// diagnostic context the caller needs to render a WARN. Pulls fallbacks from +/// `defaults.source`, `defaults.targets`, and `defaults.tags`. pub(crate) async fn resolve_mapping( mapping: &MappingConfig, config: &Config, clients: &HashMap>, batch_checkers: &HashMap>, with_report: bool, -) -> Result, CliError> { +) -> Result { // --- Source registry --- let source_name = mapping .source @@ -353,20 +723,49 @@ pub(crate) async fn resolve_mapping( // enumerating all tags from the source registry. This avoids // hundreds of paginated tags/list requests for repos with thousands // of tags. - let (filtered, candidate_count, filter_report) = - if let Some(exact) = tags_config.and_then(|t| t.exact_tags()) { - (exact, None, None) - } else { - let all_tags = source_client.list_tags(&source_repo_path).await?; - select_filtered_tags(tags_config, all_tags, with_report)? + // The image/artifact partition + sample collection happen in the same + // pass that prepares input for `select_filtered_tags`, so the filter and + // the no-match WARN both see consistent counts. The pre-built `NoTagsInfo` + // is only consumed when filtering yields zero tags. + let (filtered, candidate_count, filter_report, no_tags_template): ( + Vec, + Option, + Option, + Option, + ) = if let Some(exact) = tags_config.and_then(|t| t.exact_tags()) { + (exact, None, None, None) + } else { + let all_tags = source_client.list_tags(&source_repo_path).await?; + let mut samples: Vec = Vec::with_capacity(NO_TAGS_SAMPLE_CAP); + let mut image_count = 0usize; + for t in &all_tags { + if !is_referrers_fallback_tag(t) { + image_count += 1; + if samples.len() < NO_TAGS_SAMPLE_CAP { + samples.push(t.clone()); + } + } + } + let template = NoTagsInfo { + from: mapping.from.clone(), + image_count, + artifact_count: all_tags.len() - image_count, + filter_desc: describe_filter(tags_config), + samples, }; + let (kept, count, report) = select_filtered_tags(tags_config, all_tags, with_report)?; + (kept, count, report, Some(template)) + }; if filtered.is_empty() { - tracing::warn!( - from = %mapping.from, - "no tags matched after filtering, skipping mapping" - ); - return Ok(None); + let info = no_tags_template.unwrap_or_else(|| NoTagsInfo { + from: mapping.from.clone(), + image_count: 0, + artifact_count: 0, + filter_desc: describe_filter(tags_config), + samples: Vec::new(), + }); + return Ok(MappingResolution::NoMatchingTags(info)); } // --- Target repo --- @@ -433,7 +832,7 @@ pub(crate) async fn resolve_mapping( None => ResolvedArtifacts::default(), }; - Ok(Some(ResolvedMapping { + Ok(MappingResolution::Resolved(ResolvedMapping { source_authority, source_client, source_repo: RepositoryName::new(mapping.from.clone())?, @@ -449,6 +848,23 @@ pub(crate) async fn resolve_mapping( })) } +/// Emit a tracing WARN for a [`NoTagsInfo`] with both a human-readable +/// message (via [`Display`](std::fmt::Display)) and structured fields for +/// log aggregators. +fn emit_no_tags_warn(info: &NoTagsInfo) { + // `from` is omitted as a structured field -- the message renders it + // first, so the text formatter would otherwise tail with a redundant + // `from=...`. Counts and filter remain (numeric, terse, useful for + // both grep and JSON aggregation). + tracing::warn!( + source_total = info.source_total(), + image_count = info.image_count, + artifact_count = info.artifact_count, + filter = info.filter_desc.as_deref().unwrap_or(""), + "{info}" + ); +} + /// Build a `FilterConfig` from a `TagsConfig`, falling back to defaults. fn build_filter(tags: Option<&TagsConfig>) -> FilterConfig { let Some(tags) = tags else { @@ -511,43 +927,14 @@ fn select_filtered_tags( } } -/// Format one per-mapping plan line for `INFO`-level emission. +/// One-line summary of a [`TagsConfig`] suitable for log emission, e.g. +/// `semver >=1.0.0, latest=5`. Returns `None` when no filter applies. /// -/// `{source} -> {target}: {kept} of {N} tags => [t1, t2]` when the -/// candidate count is known, or `{kept} tags` on the exact-tag fast path. -fn format_plan_line( - source_repo: &str, - target_repo: &str, - kept: usize, - candidate_count: Option, - target_names: &[&str], -) -> String { - let count_phrase = match candidate_count { - Some(n) => format!("{kept} of {n} tags"), - None => format!("{kept} tags"), - }; - format!( - "{source_repo} -> {target_repo}: {count_phrase} => [{}]", - target_names.join(", ") - ) -} - -/// Emit one INFO log line per resolved mapping summarizing kept/considered -/// tag counts and target list. -fn log_resolved_mappings(mappings: &[ResolvedMapping]) { - for m in mappings { - let target_names: Vec<&str> = m.targets.iter().map(|t| &*t.name).collect(); - tracing::info!( - "{}", - format_plan_line( - m.source_repo.as_str(), - m.target_repo.as_str(), - m.tags.len(), - m.candidate_count, - &target_names, - ) - ); - } +/// Single source of truth: delegates to [`FilterConfig::describe`] after +/// the same conversion the engine uses, so dry-run stage labels and the +/// no-tags-matched WARN rationale cannot drift. +fn describe_filter(tags: Option<&TagsConfig>) -> Option { + build_filter(tags).describe() } /// Write sync output as JSON when `--json` is passed. @@ -823,32 +1210,354 @@ latest: 5 } #[test] - fn format_plan_line_filtered_path_includes_of_n() { - let line = format_plan_line( - "docker.io/library/alpine", - "alpine", - 3, - Some(50), - &["ecr-prod", "ghcr-mirror"], + fn describe_filter_combines_semver_and_latest() { + let tags = TagsConfig { + semver: Some(">=1.0.0".into()), + latest: Some(5), + ..TagsConfig::default() + }; + assert_eq!( + describe_filter(Some(&tags)).as_deref(), + Some("semver >=1.0.0, latest=5") + ); + } + + #[test] + fn describe_filter_returns_none_when_empty() { + let tags = TagsConfig::default(); + assert!(describe_filter(Some(&tags)).is_none()); + assert!(describe_filter(None).is_none()); + } + + // -- NoTagsInfo Display --------------------------------------------- + + fn no_tags_info( + from: &str, + image_count: usize, + artifact_count: usize, + filter_desc: Option<&str>, + samples: &[&str], + ) -> NoTagsInfo { + NoTagsInfo { + from: from.into(), + image_count, + artifact_count, + filter_desc: filter_desc.map(String::from), + samples: samples.iter().map(|s| (*s).into()).collect(), + } + } + + #[test] + fn no_tags_warn_renders_simple_repo() { + // 2 image tags, both shown -- no truncation, no artifact split. + let info = no_tags_info( + "library/nginx", + 2, + 0, + Some("semver >=2.0"), + &["v1.0", "v1.1"], ); assert_eq!( - line, - "docker.io/library/alpine -> alpine: 3 of 50 tags => [ecr-prod, ghcr-mirror]" + info.to_string(), + "library/nginx: 0 of 2 source tags matched filter (semver >=2.0); skipping. Source: [v1.0, v1.1]" ); } + /// Cosign-heavy repos: WARN must split image vs referrer counts so the + /// 14289-tag chainguard case is not misread as 14289 missing image tags. #[test] - fn format_plan_line_exact_tag_path_omits_of_n() { - let line = format_plan_line("ghcr.io/foo/bar", "bar", 3, None, &["ghcr-mirror"]); - assert_eq!(line, "ghcr.io/foo/bar -> bar: 3 tags => [ghcr-mirror]"); + fn no_tags_warn_splits_image_and_artifact_counts() { + let info = no_tags_info( + "chainguard/nginx", + 2, + 14287, + Some("semver >=1.0.0, latest=5"), + &["latest", "latest-dev"], + ); + let msg = info.to_string(); + assert!( + msg.contains( + "0 of 14289 source tags (2 image tags, 14287 referrer artifacts) matched filter" + ), + "{msg}" + ); } + /// Truncation appends `, ...` so the user knows the list is sampled. #[test] - fn format_plan_line_single_target() { - let line = format_plan_line("src/repo", "dst/repo", 5, Some(80), &["only-target"]); + fn no_tags_warn_appends_ellipsis_when_truncated() { + let info = NoTagsInfo { + from: "library/alpine".into(), + // image_count > samples.len() drives the truncation marker. + image_count: 100, + artifact_count: 0, + filter_desc: Some("semver >=99.0".into()), + samples: (0..5).map(|i| format!("v{i}")).collect(), + }; + assert!(info.samples_truncated()); + let msg = info.to_string(); + assert!(msg.ends_with("Source: [v0, v1, v2, v3, v4, ...]"), "{msg}"); + } + + /// Empty samples render as `` and a missing filter description + /// renders as `no filter configured` -- both ensure the message never + /// has bare parens or `[]`. + #[test] + fn no_tags_warn_renders_empty_markers() { + let info = no_tags_info("x/y", 0, 0, None, &[]); + let msg = info.to_string(); + assert!(msg.contains("(no filter configured)"), "{msg}"); + assert!(msg.ends_with("Source: "), "{msg}"); + } + + // -- WatchLogState transitions -------------------------------------- + + /// First observation triggers a WARN; repeats within the same failure + /// run are suppressed; recovery clears the entry so a relapse warns + /// again. Encodes the contract `run()` depends on. + #[test] + fn watch_log_state_emits_once_per_transition() { + let mut state = WatchLogState::default(); + assert!(state.observe_no_match("repo-a")); + assert!(!state.observe_no_match("repo-a")); + assert!(state.observe_resolved("repo-a")); + assert!(!state.observe_resolved("repo-a")); + assert!(state.observe_no_match("repo-a")); + } + + /// `retain_active` drops entries for mappings no longer in the config + /// so the set does not grow unbounded across the watch process. + #[test] + fn watch_log_state_prunes_removed_mappings() { + let mut state = WatchLogState::default(); + state.observe_no_match("repo-a"); + state.observe_no_match("repo-b"); + state.observe_no_match("repo-removed"); + + state.retain_active(["repo-a", "repo-b"]); + + // After pruning, `repo-removed` re-warns (gap means transition); + // surviving entries continue to suppress. + assert!(!state.observe_no_match("repo-a")); + assert!(!state.observe_no_match("repo-b")); + assert!(state.observe_no_match("repo-removed")); + } + + // -- per-mapping outcome aggregation + dedup --------------------------- + + fn img( + source: &str, + target: &str, + status: ocync_sync::ImageStatus, + bytes: u64, + ) -> ocync_sync::ImageResult { + use ocync_sync::{BlobTransferStats, ImageResult}; + ImageResult { + image_id: uuid::Uuid::now_v7(), + source: source.into(), + target: target.into(), + status, + bytes_transferred: bytes, + blob_stats: BlobTransferStats::default(), + duration: Duration::from_secs(1), + artifacts_skipped: false, + } + } + + fn report_with(images: Vec) -> SyncReport { + SyncReport { + run_id: uuid::Uuid::now_v7(), + images, + stats: ocync_sync::SyncStats::default(), + duration: Duration::from_secs(1), + } + } + + /// Aggregation groups by `source_repo:` + `target_repo:` prefix so + /// images from a different mapping (same source repo, different target) + /// don't bleed into this mapping's counts. + #[test] + fn aggregate_mapping_outcome_groups_by_source_and_target() { + let report = report_with(vec![ + img( + "library/alpine:3.20", + "mirror/a:3.20", + ocync_sync::ImageStatus::Synced, + 100, + ), + img( + "library/alpine:3.21", + "mirror/a:3.21", + ocync_sync::ImageStatus::Synced, + 200, + ), + img( + "library/alpine:3.21", + "mirror/b:3.21", + ocync_sync::ImageStatus::Skipped { + reason: ocync_sync::SkipReason::DigestMatch, + }, + 0, + ), + ]); + let o = aggregate_mapping_outcome("library/alpine", "mirror/a", &report); + assert_eq!(o.synced, 2); + assert_eq!(o.skipped, 0); + assert_eq!(o.bytes, 300); + } + + /// Empty mappings (no images in the report) skip silently -- the + /// no-tags WARN is the right surface for that case, not this one. + #[test] + fn empty_outcome_is_recognized() { + let outcome = MappingOutcome::default(); + assert!(outcome.is_empty()); + let with_skip = MappingOutcome { + skipped: 1, + ..MappingOutcome::default() + }; + assert!(!with_skip.is_empty()); + } + + /// First observation emits; identical follow-up suppresses; outcome + /// change emits again. Mirrors the no-tags transition contract. + #[test] + fn watch_log_state_dedupes_identical_mapping_outcomes() { + let mut state = WatchLogState::default(); + let steady = MappingOutcome { + skipped: 5, + ..MappingOutcome::default() + }; + let active = MappingOutcome { + synced: 1, + skipped: 4, + bytes: 1024, + ..MappingOutcome::default() + }; + + // First observation: emit, no prior so not a recovery. + assert_eq!( + state.observe_mapping_outcome("repo-a", &steady), + Some(false) + ); + // Same outcome twice: suppress. + assert_eq!(state.observe_mapping_outcome("repo-a", &steady), None); + assert_eq!(state.observe_mapping_outcome("repo-a", &steady), None); + // Different outcome: emit, neither prior nor current was a failure + // so not a recovery either. + assert_eq!( + state.observe_mapping_outcome("repo-a", &active), + Some(false) + ); + assert_eq!(state.observe_mapping_outcome("repo-a", &active), None); + assert_eq!( + state.observe_mapping_outcome("repo-a", &steady), + Some(false) + ); + } + + /// Recovery detection: prior outcome with `failed > 0` followed by an + /// outcome with `failed == 0` returns `Some(true)` so the caller can + /// attach the `[recovered]` marker. Cycle counter advances on each + /// transition so the watch loop sees activity. + #[test] + fn watch_log_state_surfaces_failure_to_clean_transition() { + let mut state = WatchLogState::default(); + let failing = MappingOutcome { + failed: 1, + ..MappingOutcome::default() + }; + let healthy = MappingOutcome { + synced: 1, + ..MappingOutcome::default() + }; + + // First observation can never be a recovery (no prior). + assert_eq!(state.observe_mapping_outcome("r", &failing), Some(false)); + assert_eq!(state.observe_mapping_outcome("r", &healthy), Some(true)); + assert_eq!(state.cycle_emit_count(), 2); + } + + /// `retain_active` also prunes per-mapping outcome cache so a removed + /// mapping doesn't keep its stale entry forever. Re-observation after + /// pruning emits as a fresh first-observation. + #[test] + fn watch_log_state_retain_active_also_prunes_outcomes() { + let mut state = WatchLogState::default(); + let outcome = MappingOutcome { + skipped: 1, + ..MappingOutcome::default() + }; + state.observe_mapping_outcome("keep", &outcome); + state.observe_mapping_outcome("drop", &outcome); + + state.retain_active(["keep"]); + + assert_eq!(state.observe_mapping_outcome("drop", &outcome), Some(false)); + assert_eq!(state.observe_mapping_outcome("keep", &outcome), None); + } + + /// `format_mapping_outcome` omits zero counts, elides the bytes clause + /// when nothing transferred, tags recovery transitions, and drops the + /// `[targets]` bracket on single-target mappings (the destination is + /// already in the `from -> to` arrow). + #[test] + fn format_mapping_outcome_single_target_omits_bracket() { + let d = MappingDescriptor { + from: "library/alpine".into(), + target_repo: "mirror/alpine".into(), + target_names: vec!["ttl".into()], + }; + let synced_only = MappingOutcome { + synced: 3, + bytes: 1024, + ..MappingOutcome::default() + }; + assert_eq!( + format_mapping_outcome(&d, &synced_only, false), + "library/alpine -> mirror/alpine: synced 3 (1.0 KB)" + ); + let skipped_only = MappingOutcome { + skipped: 5, + ..MappingOutcome::default() + }; + assert_eq!( + format_mapping_outcome(&d, &skipped_only, false), + "library/alpine -> mirror/alpine: skipped 5" + ); + let mixed = MappingOutcome { + synced: 1, + skipped: 2, + failed: 1, + bytes: 2048, + }; + assert_eq!( + format_mapping_outcome(&d, &mixed, false), + "library/alpine -> mirror/alpine: synced 1, skipped 2, failed 1 (2.0 KB)" + ); + assert_eq!( + format_mapping_outcome(&d, &synced_only, true), + "library/alpine -> mirror/alpine: synced 3 (1.0 KB) [recovered]" + ); + } + + /// Multi-target mappings keep the `[targets]` bracket so the operator + /// can see which destinations the outcome covers. + #[test] + fn format_mapping_outcome_multi_target_keeps_bracket() { + let d = MappingDescriptor { + from: "library/alpine".into(), + target_repo: "mirror/alpine".into(), + target_names: vec!["ecr-prod".into(), "ghcr-mirror".into()], + }; + let synced = MappingOutcome { + synced: 1, + bytes: 1024, + ..MappingOutcome::default() + }; assert_eq!( - line, - "src/repo -> dst/repo: 5 of 80 tags => [only-target]" + format_mapping_outcome(&d, &synced, false), + "library/alpine -> mirror/alpine [ecr-prod, ghcr-mirror]: synced 1 (1.0 KB)" ); } @@ -1002,7 +1711,7 @@ latest: 5 assert!(out.contains("FAIL"), "{out}"); assert!(out.contains("BelowMinTags"), "{out}"); // And carries the full pipeline trace. - assert!(out.contains("source candidates: 10"), "{out}"); - assert!(out.contains("dropped 10:"), "{out}"); + assert!(out.contains("source tags: 10"), "{out}"); + assert!(out.contains("dropped (10):"), "{out}"); } } diff --git a/src/cli/commands/watch.rs b/src/cli/commands/watch.rs index ac83be49..d00c478b 100644 --- a/src/cli/commands/watch.rs +++ b/src/cli/commands/watch.rs @@ -2,12 +2,12 @@ use std::cell::RefCell; use std::rc::Rc; -use std::time::Duration; +use std::time::{Duration, Instant}; use ocync_sync::cache::TransferStateCache; use tokio::net::TcpListener; -use crate::cli::commands::synchronize; +use crate::cli::commands::synchronize::{self, WatchLogState}; use crate::cli::config::load_config; use crate::cli::health::HealthState; use crate::cli::shutdown::ShutdownSignal; @@ -69,6 +69,18 @@ pub(crate) async fn run( }) }; + // Watch-mode log state: tracks which mappings have already + // emitted a no-tags-matched WARN so we emit one per transition, + // not one per cycle. Pruned each cycle to mappings still in config. + let mut watch_log = WatchLogState::default(); + + // Heartbeat: when no log line has been emitted for an extended + // period (idle steady-state watch), emit a "still alive" INFO + // so log scrapers have a recent anchor confirming the process + // is doing its job. Cadence is decoupled from the sync interval. + const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(3600); + let mut last_emit_at = Instant::now(); + // Track consecutive config reload failures for backoff. let mut config_failures: u32 = 0; const BACKOFF_THRESHOLD: u32 = 3; @@ -139,17 +151,21 @@ pub(crate) async fn run( json: args.json, }; + watch_log.begin_cycle(); match synchronize::run( &sync_args, progress, Some(&shutdown), Some(Rc::clone(&cache)), verbose, + Some(&mut watch_log), ) .await { Ok(code) => { - tracing::info!(exit_code = ?code, "sync cycle complete"); + // Cycle completion is conveyed by the dedup-aware + // per-mapping line + cycle tail (when stats change) + // and by /healthz; no separate INFO needed. if matches!(code, ExitCode::Success | ExitCode::PartialFailure) { health_state.borrow_mut().record_success(); } @@ -159,6 +175,21 @@ pub(crate) async fn run( } } + // Heartbeat gate: if anything emitted this cycle, the user + // already has a recent log anchor; reset the timer. Else + // check the elapsed-since-last-emit and fire the heartbeat + // when we've been silent past the threshold. + if watch_log.cycle_emit_count() > 0 { + last_emit_at = Instant::now(); + } else if last_emit_at.elapsed() >= HEARTBEAT_INTERVAL { + let idle_secs = last_emit_at.elapsed().as_secs(); + tracing::info!( + idle_secs, + "watch alive: no state changes; sync loop healthy" + ); + last_emit_at = Instant::now(); + } + // Wait for the interval to elapse, or return early on shutdown. tokio::select! { () = tokio::time::sleep(interval) => {} diff --git a/src/cli/config.rs b/src/cli/config.rs index 787cf2bd..79ab07ce 100644 --- a/src/cli/config.rs +++ b/src/cli/config.rs @@ -754,7 +754,7 @@ fn validate_artifacts(context: &str, artifacts: &ArtifactsConfig) -> Result<(), ))); } if !artifacts.enabled { - tracing::warn!( + tracing::debug!( context, "artifacts.enabled is false: signatures and SBOMs will be stripped from synced images" ); @@ -769,7 +769,7 @@ fn validate_tags(context: &str, tags: &TagsConfig) -> Result<(), ConfigError> { )); } if tags.semver.is_some() && tags.latest.is_none() { - tracing::warn!( + tracing::debug!( context, "tags.semver is set without tags.latest: every tag matching the version range will sync. \ For long-running mirrors of images with many tags, consider adding 'latest: N' (with 'sort: semver') \ diff --git a/src/cli/progress.rs b/src/cli/progress.rs index 3429640d..343088ea 100644 --- a/src/cli/progress.rs +++ b/src/cli/progress.rs @@ -1,7 +1,9 @@ //! Verbosity-aware progress reporters for sync output. //! -//! [`TextProgress`] writes plain status lines to stderr (non-TTY and TTY -//! alike). The run summary always goes to stdout. +//! [`TextProgress`] writes per-image status lines to stderr. There is no +//! per-cycle aggregate stdout line: the CLI driver emits a per-mapping +//! INFO line via tracing after the engine returns, which carries the +//! source/target context an aggregate cannot. use std::cell::RefCell; use std::io::{self, Write}; @@ -49,111 +51,29 @@ fn format_image_line(result: &ImageResult, verbosity: u8) -> Option { } } -/// Write the run summary to `stdout`, or do nothing if `suppress_summary` -/// is true or the report contains no images. -fn write_run_summary( - stdout: &RefCell>, - report: &SyncReport, - suppress_summary: bool, -) { - if suppress_summary { - return; - } - if report.images.is_empty() { - return; - } - let s = &report.stats; - let has_discovery = s.discovery_cache_hits > 0 - || s.discovery_cache_misses > 0 - || s.discovery_head_first_skips > 0 - || s.immutable_tag_skips > 0; - let discovery = if has_discovery { - let head_first_suffix = if s.discovery_head_first_skips > 0 { - format!(", {} head_first", s.discovery_head_first_skips) - } else { - String::new() - }; - let immutable_suffix = if s.immutable_tag_skips > 0 { - format!(", {} immutable", s.immutable_tag_skips) - } else { - String::new() - }; - format!( - " | discovery: {} cached, {} pulled{}{}", - s.discovery_cache_hits, s.discovery_cache_misses, head_first_suffix, immutable_suffix, - ) - } else { - String::new() - }; - let artifacts_warn = if s.artifacts_skipped > 0 { - format!(" | {} artifacts skipped", s.artifacts_skipped) - } else { - String::new() - }; - if let Err(e) = writeln!( - stdout.borrow_mut(), - "sync complete: {} synced, {} skipped, {} failed | blobs: {} transferred, {} skipped, {} mounted | {} in {}{}{}", - s.images_synced, - s.images_skipped, - s.images_failed, - s.blobs_transferred, - s.blobs_skipped, - s.blobs_mounted, - format_bytes(s.bytes_transferred), - format_duration(report.duration), - discovery, - artifacts_warn, - ) { - tracing::warn!(error = %e, "failed to write progress summary to stdout"); - } -} - /// Text progress reporter with configurable verbosity. /// -/// Per-image status lines go to stderr (alongside tracing logs). -/// The run summary goes to stdout (pipeable, parseable). -/// -/// Uses [`RefCell`] for interior mutability because the -/// [`ProgressReporter`] trait takes `&self` and the engine runs on a -/// single-threaded tokio runtime. +/// Per-image status lines (`synced` / `FAILED`) go to stderr. There is no +/// stdout per-cycle aggregate; the CLI emits per-mapping INFO lines from +/// the sync driver instead. pub(crate) struct TextProgress { verbosity: u8, - /// When true, suppress the stdout summary line. Used when JSON output - /// owns stdout (`--json`) or when the summary is redundant (e.g., `copy` - /// with a single image where the per-image line says everything). - suppress_summary: bool, stderr: RefCell>, - stdout: RefCell>, } impl TextProgress { - /// Create a new text progress reporter writing to real stderr/stdout. - /// - /// When `suppress_summary` is true, the run summary is suppressed on - /// stdout. Per-image status lines still go to stderr. Used when JSON - /// output owns stdout or when the summary is redundant (single-image copy). - pub(crate) fn new(verbosity: u8, suppress_summary: bool) -> Self { + pub(crate) fn new(verbosity: u8) -> Self { Self { verbosity, - suppress_summary, stderr: RefCell::new(Box::new(io::stderr())), - stdout: RefCell::new(Box::new(io::stdout())), } } - /// Create a text progress reporter with custom writers (for testing). #[cfg(test)] - fn with_writers( - verbosity: u8, - suppress_summary: bool, - stderr: Box, - stdout: Box, - ) -> Self { + fn with_writer(verbosity: u8, stderr: Box) -> Self { Self { verbosity, - suppress_summary, stderr: RefCell::new(stderr), - stdout: RefCell::new(stdout), } } } @@ -171,8 +91,12 @@ impl ProgressReporter for TextProgress { } } - fn run_completed(&self, report: &SyncReport) { - write_run_summary(&self.stdout, report, self.suppress_summary); + fn run_completed(&self, _report: &SyncReport) { + // Per-cycle aggregate is no longer emitted here; the CLI driver + // emits a per-mapping line via tracing INFO after the engine + // returns, which carries the source/target context the aggregate + // lacked. The `--json` path writes the structured report to stdout + // separately in `write_output`. } } @@ -400,449 +324,54 @@ mod tests { "skipped source/repo:v1 -> target/repo:v1 (digest match)" ); } + // - TextProgress IO tests (stderr only) -- - // - write_run_summary tests -- - - #[test] - fn summary_exact_format() { - let buf: Buf = Rc::new(RefCell::new(Vec::new())); - let stdout: RefCell> = RefCell::new(Box::new(RcWriter(Rc::clone(&buf)))); - let report = SyncReport { - run_id: Uuid::now_v7(), - images: vec![make_result(ImageStatus::Synced, 1024)], - stats: SyncStats { - images_synced: 3, - images_skipped: 47, - images_failed: 1, - blobs_transferred: 12, - blobs_skipped: 5, - blobs_mounted: 34, - bytes_transferred: 432_000_000, - ..SyncStats::default() - }, - duration: Duration::from_secs(47), - }; - write_run_summary(&stdout, &report, false); - let output = String::from_utf8(buf.borrow().clone()).unwrap(); - assert_eq!( - output, - "sync complete: 3 synced, 47 skipped, 1 failed | blobs: 12 transferred, 5 skipped, 34 mounted | 432.0 MB in 47s\n" - ); + fn text_progress(verbosity: u8) -> (TextProgress, Buf) { + let stderr_buf: Buf = Rc::new(RefCell::new(Vec::new())); + let progress = + TextProgress::with_writer(verbosity, Box::new(RcWriter(Rc::clone(&stderr_buf)))); + (progress, stderr_buf) } #[test] - fn summary_with_discovery_stats() { - let buf: Buf = Rc::new(RefCell::new(Vec::new())); - let stdout: RefCell> = RefCell::new(Box::new(RcWriter(Rc::clone(&buf)))); - let report = SyncReport { - run_id: Uuid::now_v7(), - images: vec![make_result(ImageStatus::Synced, 1024)], - stats: SyncStats { - images_synced: 3, - images_skipped: 47, - images_failed: 0, - blobs_transferred: 12, - blobs_skipped: 5, - blobs_mounted: 34, - bytes_transferred: 432_000_000, - discovery_cache_hits: 40, - discovery_cache_misses: 10, - discovery_head_failures: 2, - discovery_target_stale: 1, - discovery_head_first_skips: 0, - immutable_tag_skips: 0, - artifacts_skipped: 0, - }, - duration: Duration::from_secs(47), - }; - write_run_summary(&stdout, &report, false); - let output = String::from_utf8(buf.borrow().clone()).unwrap(); - assert_eq!( - output, - "sync complete: 3 synced, 47 skipped, 0 failed | blobs: 12 transferred, 5 skipped, 34 mounted | 432.0 MB in 47s | discovery: 40 cached, 10 pulled\n" - ); - } - - #[test] - fn summary_omits_discovery_when_zero() { - let buf: Buf = Rc::new(RefCell::new(Vec::new())); - let stdout: RefCell> = RefCell::new(Box::new(RcWriter(Rc::clone(&buf)))); - let report = make_report(vec![make_result(ImageStatus::Synced, 1024)]); - write_run_summary(&stdout, &report, false); - let output = String::from_utf8(buf.borrow().clone()).unwrap(); - assert!(!output.contains("discovery"), "got: {output}"); - } - - #[test] - fn summary_with_only_cache_hits_includes_discovery() { - // Distinguishes the `||` from `&&` in the discovery condition: - // even when misses == 0, hits > 0 should show the discovery suffix. - let buf: Buf = Rc::new(RefCell::new(Vec::new())); - let stdout: RefCell> = RefCell::new(Box::new(RcWriter(Rc::clone(&buf)))); - let report = SyncReport { - run_id: Uuid::now_v7(), - images: vec![make_result(ImageStatus::Synced, 1024)], - stats: SyncStats { - images_synced: 1, - discovery_cache_hits: 5, - discovery_cache_misses: 0, - ..SyncStats::default() - }, - duration: Duration::from_secs(1), - }; - write_run_summary(&stdout, &report, false); - let output = String::from_utf8(buf.borrow().clone()).unwrap(); - assert!( - output.contains("discovery: 5 cached, 0 pulled"), - "got: {output}" - ); - } - - #[test] - fn summary_with_only_cache_misses_includes_discovery() { - let buf: Buf = Rc::new(RefCell::new(Vec::new())); - let stdout: RefCell> = RefCell::new(Box::new(RcWriter(Rc::clone(&buf)))); - let report = SyncReport { - run_id: Uuid::now_v7(), - images: vec![make_result(ImageStatus::Synced, 1024)], - stats: SyncStats { - images_synced: 1, - discovery_cache_hits: 0, - discovery_cache_misses: 3, - ..SyncStats::default() - }, - duration: Duration::from_secs(1), - }; - write_run_summary(&stdout, &report, false); - let output = String::from_utf8(buf.borrow().clone()).unwrap(); - assert!( - output.contains("discovery: 0 cached, 3 pulled"), - "got: {output}" - ); - } - - #[test] - fn summary_with_head_first_skips_includes_suffix() { - let buf: Buf = Rc::new(RefCell::new(Vec::new())); - let stdout: RefCell> = RefCell::new(Box::new(RcWriter(Rc::clone(&buf)))); - let report = SyncReport { - run_id: Uuid::now_v7(), - images: vec![make_result(ImageStatus::Synced, 1024)], - stats: SyncStats { - images_synced: 1, - images_skipped: 3, - discovery_cache_hits: 2, - discovery_cache_misses: 1, - discovery_head_first_skips: 3, - ..SyncStats::default() - }, - duration: Duration::from_secs(5), - }; - write_run_summary(&stdout, &report, false); - let output = String::from_utf8(buf.borrow().clone()).unwrap(); - assert!( - output.contains("discovery: 2 cached, 1 pulled, 3 head_first"), - "got: {output}" - ); - } - - #[test] - fn summary_with_only_head_first_skips_includes_discovery() { - let buf: Buf = Rc::new(RefCell::new(Vec::new())); - let stdout: RefCell> = RefCell::new(Box::new(RcWriter(Rc::clone(&buf)))); - let report = SyncReport { - run_id: Uuid::now_v7(), - images: vec![make_result(ImageStatus::Synced, 1024)], - stats: SyncStats { - images_synced: 1, - discovery_head_first_skips: 5, - ..SyncStats::default() - }, - duration: Duration::from_secs(1), - }; - write_run_summary(&stdout, &report, false); - let output = String::from_utf8(buf.borrow().clone()).unwrap(); - assert!( - output.contains("discovery: 0 cached, 0 pulled, 5 head_first"), - "got: {output}" - ); - } - - #[test] - fn summary_with_immutable_skips_includes_suffix() { - let buf: Buf = Rc::new(RefCell::new(Vec::new())); - let stdout: RefCell> = RefCell::new(Box::new(RcWriter(Rc::clone(&buf)))); - let report = SyncReport { - run_id: Uuid::now_v7(), - images: vec![make_result(ImageStatus::Synced, 1024)], - stats: SyncStats { - images_synced: 1, - images_skipped: 10, - discovery_cache_hits: 2, - discovery_cache_misses: 1, - immutable_tag_skips: 8, - ..SyncStats::default() - }, - duration: Duration::from_secs(3), - }; - write_run_summary(&stdout, &report, false); - let output = String::from_utf8(buf.borrow().clone()).unwrap(); - assert!( - output.contains("discovery: 2 cached, 1 pulled, 8 immutable"), - "got: {output}" - ); - } - - #[test] - fn summary_with_only_immutable_skips_includes_discovery() { - let buf: Buf = Rc::new(RefCell::new(Vec::new())); - let stdout: RefCell> = RefCell::new(Box::new(RcWriter(Rc::clone(&buf)))); - let report = SyncReport { - run_id: Uuid::now_v7(), - images: vec![make_result(ImageStatus::Synced, 1024)], - stats: SyncStats { - images_synced: 1, - immutable_tag_skips: 50, - ..SyncStats::default() - }, - duration: Duration::from_secs(1), - }; - write_run_summary(&stdout, &report, false); - let output = String::from_utf8(buf.borrow().clone()).unwrap(); - assert!( - output.contains("discovery: 0 cached, 0 pulled, 50 immutable"), - "got: {output}" - ); - } - - #[test] - fn summary_with_artifacts_skipped() { - let buf: Buf = Rc::new(RefCell::new(Vec::new())); - let stdout: RefCell> = RefCell::new(Box::new(RcWriter(Rc::clone(&buf)))); - let report = SyncReport { - run_id: Uuid::now_v7(), - images: vec![make_result(ImageStatus::Synced, 1024)], - stats: SyncStats { - images_synced: 5, - artifacts_skipped: 2, - ..SyncStats::default() - }, - duration: Duration::from_secs(10), - }; - write_run_summary(&stdout, &report, false); - let output = String::from_utf8(buf.borrow().clone()).unwrap(); - assert!(output.contains("2 artifacts skipped"), "got: {output}"); - } - - #[test] - fn summary_without_artifacts_skipped_omits_suffix() { - let buf: Buf = Rc::new(RefCell::new(Vec::new())); - let stdout: RefCell> = RefCell::new(Box::new(RcWriter(Rc::clone(&buf)))); - let report = make_report(vec![make_result(ImageStatus::Synced, 1024)]); - write_run_summary(&stdout, &report, false); - let output = String::from_utf8(buf.borrow().clone()).unwrap(); - assert!(!output.contains("artifacts skipped"), "got: {output}"); - } - - #[test] - fn summary_suppressed_produces_no_output() { - let buf: Buf = Rc::new(RefCell::new(Vec::new())); - let stdout: RefCell> = RefCell::new(Box::new(RcWriter(Rc::clone(&buf)))); - let report = make_report(vec![make_result(ImageStatus::Synced, 1024)]); - write_run_summary(&stdout, &report, true); - assert!(buf.borrow().is_empty()); - } - - #[test] - fn summary_empty_report_produces_no_output() { - let buf: Buf = Rc::new(RefCell::new(Vec::new())); - let stdout: RefCell> = RefCell::new(Box::new(RcWriter(Rc::clone(&buf)))); - let report = SyncReport { - run_id: Uuid::now_v7(), - images: vec![], - stats: SyncStats::default(), - duration: Duration::ZERO, - }; - write_run_summary(&stdout, &report, false); - assert!(buf.borrow().is_empty()); - } - - // - TextProgress tests (wiring: writes to correct streams) -- - - fn test_text_progress(verbosity: u8) -> (TextProgress, Buf, Buf) { - test_text_progress_with_suppress(verbosity, false) - } - - fn test_text_progress_with_suppress( - verbosity: u8, - suppress_summary: bool, - ) -> (TextProgress, Buf, Buf) { - let stderr_buf = Rc::new(RefCell::new(Vec::new())); - let stdout_buf = Rc::new(RefCell::new(Vec::new())); - let progress = TextProgress::with_writers( - verbosity, - suppress_summary, - Box::new(RcWriter(Rc::clone(&stderr_buf))), - Box::new(RcWriter(Rc::clone(&stdout_buf))), - ); - (progress, stderr_buf, stdout_buf) - } - - #[test] - fn text_image_started_is_noop() { - let (progress, stderr, stdout) = test_text_progress(1); - progress.image_started("source/repo:v1", "target/repo:v1"); - assert!( - stderr.borrow().is_empty(), - "image_started should not write to stderr" - ); - assert!( - stdout.borrow().is_empty(), - "image_started should not write to stdout" - ); - } - - #[test] - fn text_image_completed_writes_to_stderr() { - let (progress, stderr, stdout) = test_text_progress(1); - let result = make_result(ImageStatus::Synced, 187_000_000); - progress.image_completed(&result); - let output = String::from_utf8(stderr.borrow().clone()).unwrap(); - assert!(output.starts_with("synced "), "got: {output}"); - assert!( - stdout.borrow().is_empty(), - "per-image output must NOT go to stdout" - ); + fn text_image_completed_writes_to_stderr_at_v1() { + let (progress, stderr) = text_progress(1); + progress.image_completed(&make_result(ImageStatus::Synced, 1024)); + let out = String::from_utf8(stderr.borrow().clone()).unwrap(); + assert!(out.starts_with("synced "), "got: {out}"); } #[test] fn text_image_completed_silent_at_v0() { - let (progress, stderr, stdout) = test_text_progress(0); - let result = make_result(ImageStatus::Synced, 187_000_000); - progress.image_completed(&result); + let (progress, stderr) = text_progress(0); + progress.image_completed(&make_result(ImageStatus::Synced, 1024)); assert!(stderr.borrow().is_empty()); - assert!(stdout.borrow().is_empty()); } #[test] fn text_failed_always_writes_to_stderr() { - let (progress, stderr, stdout) = test_text_progress(0); - let result = make_result( - ImageStatus::Failed { - kind: ErrorKind::ManifestPush, - error: "timeout".into(), - retries: 2, - status_code: None, - }, - 0, - ); - progress.image_completed(&result); - let output = String::from_utf8(stderr.borrow().clone()).unwrap(); - assert!(output.starts_with("FAILED "), "got: {output}"); - assert!( - stdout.borrow().is_empty(), - "per-image output must NOT go to stdout" - ); - } - - #[test] - fn text_stream_separation() { - let (progress, stderr, stdout) = test_text_progress(0); - let failed = make_result( - ImageStatus::Failed { - kind: ErrorKind::ManifestPull, - error: "timeout".into(), - retries: 2, - status_code: None, - }, - 0, - ); - progress.image_completed(&failed); - - let report = make_report(vec![make_result(ImageStatus::Synced, 1024)]); - progress.run_completed(&report); - - let stderr_text = String::from_utf8(stderr.borrow().clone()).unwrap(); - let stdout_text = String::from_utf8(stdout.borrow().clone()).unwrap(); - - // Per-image output on stderr, summary on stdout, never crossed. - assert!(stderr_text.contains("FAILED"), "FAILED should be on stderr"); - assert!( - !stdout_text.contains("FAILED"), - "FAILED must NOT be on stdout" - ); - assert!( - stdout_text.contains("sync complete:"), - "summary should be on stdout" - ); - assert!( - !stderr_text.contains("sync complete:"), - "summary must NOT be on stderr" - ); - } - - #[test] - fn text_multiple_images_mixed_status() { - let (progress, stderr, _stdout) = test_text_progress(1); - - progress.image_completed(&make_result(ImageStatus::Synced, 100_000_000)); - progress.image_completed(&make_result(ImageStatus::Synced, 200_000_000)); - progress.image_completed(&make_result( - ImageStatus::Skipped { - reason: SkipReason::DigestMatch, - }, - 0, - )); + let (progress, stderr) = text_progress(0); progress.image_completed(&make_result( ImageStatus::Failed { kind: ErrorKind::BlobTransfer, - error: "connection lost".into(), + error: "timeout".into(), retries: 1, status_code: None, }, 0, )); - - let output = String::from_utf8(stderr.borrow().clone()).unwrap(); - let lines: Vec<&str> = output.lines().collect(); - assert_eq!( - lines.iter().filter(|l| l.starts_with("synced ")).count(), - 2 - ); - assert_eq!( - lines.iter().filter(|l| l.starts_with("skipped ")).count(), - 1 - ); - assert_eq!( - lines.iter().filter(|l| l.starts_with("FAILED ")).count(), - 1 - ); + let out = String::from_utf8(stderr.borrow().clone()).unwrap(); + assert!(out.starts_with("FAILED "), "got: {out}"); } + /// `run_completed` is intentionally a no-op now -- the per-cycle + /// aggregate moved to per-mapping INFO lines emitted by the CLI driver. + /// Lock that contract: even with a non-empty report, nothing reaches + /// stderr from this method. #[test] - fn text_suppress_summary_still_prints_failures() { - let (progress, stderr, stdout) = test_text_progress_with_suppress(0, true); - let result = make_result( - ImageStatus::Failed { - kind: ErrorKind::ManifestPush, - error: "timeout".into(), - retries: 2, - status_code: None, - }, - 0, - ); - progress.image_completed(&result); - - let report = make_report(vec![make_result(ImageStatus::Synced, 1024)]); - progress.run_completed(&report); - - assert!( - !stderr.borrow().is_empty(), - "failures should still go to stderr" - ); - assert!( - stdout.borrow().is_empty(), - "suppress_summary should suppress stdout" - ); + fn text_run_completed_emits_nothing() { + let (progress, stderr) = text_progress(2); + progress.run_completed(&make_report(vec![make_result(ImageStatus::Synced, 1)])); + assert!(stderr.borrow().is_empty()); } } diff --git a/src/main.rs b/src/main.rs index 928aed4d..ad7ef602 100644 --- a/src/main.rs +++ b/src/main.rs @@ -313,16 +313,6 @@ async fn main() -> std::process::ExitCode { let shutdown = cli::shutdown::ShutdownSignal::new(); cli::shutdown::install_signal_handlers(shutdown.clone()); - // Suppress the text summary on stdout when JSON owns stdout or when - // the summary is redundant (single-image copy). - let suppress_summary = match &cli.command { - Commands::Sync(args) => args.json, - Commands::Watch(args) => args.json, - Commands::Analyze(args) => args.json, - Commands::Copy(_) => true, - _ => false, - }; - let effective_verbosity = match &cli.command { // Copy always shows per-image output -- users expect to see what was copied. Commands::Copy(_) => cli.verbose.max(1), @@ -332,10 +322,7 @@ async fn main() -> std::process::ExitCode { let progress: Box = if cli.quiet { Box::new(NullProgress) } else { - Box::new(cli::progress::TextProgress::new( - effective_verbosity, - suppress_summary, - )) + Box::new(cli::progress::TextProgress::new(effective_verbosity)) }; // Dry-run / log-emission verbose toggle: any -v level removes the sample @@ -344,7 +331,8 @@ async fn main() -> std::process::ExitCode { let verbose = cli.verbose >= 1; let result = match cli.command { Commands::Sync(args) => { - cli::commands::synchronize::run(&args, &*progress, Some(&shutdown), None, verbose).await + cli::commands::synchronize::run(&args, &*progress, Some(&shutdown), None, verbose, None) + .await } Commands::Copy(args) => cli::commands::copy::run(&args, &*progress, Some(&shutdown)).await, Commands::Tags(args) => cli::commands::tags::run(&args).await,