diff --git a/src/cache/service.rs b/src/cache/service.rs index c7e4176..648b9ec 100644 --- a/src/cache/service.rs +++ b/src/cache/service.rs @@ -361,12 +361,11 @@ impl CacheService { // candidate row is not stale before anything is unlinked; under the // stripe nothing can requeue or republish in between (`commit_staged` // takes the same stripe), so `evict` then removes the row it just saw. - match self.metadata.artifact(channel, candidate.artifact).await? { - Some(metadata) if metadata.eligible_at == candidate.eligible_at => { - self.files.remove(channel, candidate.artifact).await?; - } - // Requeued or already gone: let `evict` drop the stale queue row. - _ => {} + // Requeued or already gone: let `evict` drop the stale queue row. + if let Some(metadata) = self.metadata.artifact(channel, candidate.artifact).await? + && metadata.eligible_at == candidate.eligible_at + { + self.files.remove(channel, candidate.artifact).await?; } match self .metadata diff --git a/src/cacheprog/mod.rs b/src/cacheprog/mod.rs index 4d0cab3..0b1b766 100644 --- a/src/cacheprog/mod.rs +++ b/src/cacheprog/mod.rs @@ -280,7 +280,9 @@ where ) .await; if args.ephemeral_cache { - tokio::fs::remove_dir_all(&directory).await?; + // Best-effort, like `prune_stale_files`: a cleanup failure after a successful + // build would otherwise exit non-zero, which Go reports as a cache error. + let _ = tokio::fs::remove_dir_all(&directory).await; } else { let object_max_age = (args.prune_days > 0).then(|| Duration::from_secs(args.prune_days * 24 * 60 * 60)); diff --git a/src/cacheprog/session.rs b/src/cacheprog/session.rs index 2338df9..3b08502 100644 --- a/src/cacheprog/session.rs +++ b/src/cacheprog/session.rs @@ -282,9 +282,14 @@ async fn prefetch( // Every zero-size action shares the one empty output; synthesize the // file locally instead of downloading nothing over the network. A // zero-size entry under any other digest can never verify, so it is - // dropped from the work list entirely. - if entry.output == EMPTY_OUTPUT { - super::write_disk_file(directory, &entry.output, &[]).await?; + // dropped from the work list entirely. A failed write is one future + // foreground miss, like any other per-object failure, so it neither + // counts as local nor stops the pass. + if entry.output == EMPTY_OUTPUT + && super::write_disk_file(directory, &entry.output, &[]) + .await + .is_ok() + { local += 1; } continue; diff --git a/src/cli.rs b/src/cli.rs index c4db8d0..6c43567 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -102,35 +102,32 @@ pub struct ServeArgs { } impl ServeArgs { - pub fn config(&self) -> Config { - let mut config = Config::new(&self.data_dir); - config.max_upload_bytes = self.max_upload_bytes; - config.default_expiry_seconds = self.default_expiry_seconds; - config.foreground_concurrency = self.foreground_concurrency; - config.reservation_extent_bytes = self.reservation_extent_bytes; - config.low_watermark_bytes = self.low_watermark_bytes; - config.high_watermark_bytes = self.high_watermark_bytes; - config.emergency_headroom_bytes = self.emergency_headroom_bytes; - config.bloom_bits = self.bloom_bits; - config.reclaim_candidate_limit = self.reclaim_candidate_limit; - config.reclaim_byte_limit = self.reclaim_byte_limit; - config.orphan_scan_limit = self.orphan_scan_limit; - config.go_upstream.clone_from(&self.go_upstream); - config.python_upstream.clone_from(&self.python_upstream); - config.npm_upstream.clone_from(&self.npm_upstream); - config - .cargo_index_upstream - .clone_from(&self.cargo_index_upstream); - config - .cargo_crate_upstream - .clone_from(&self.cargo_crate_upstream); - config.proxy_revalidation_seconds = self.proxy_revalidation_seconds; - config.proxy_concurrency = self.proxy_concurrency; - config.upstream_timeout_seconds = self.upstream_timeout_seconds; - config - .proxy_allowed_origins - .clone_from(&self.proxy_allowed_origins); - config + /// Names every `Config` field rather than overwriting `Config::new`'s defaults, so + /// a field added to `Config` is a compile error here instead of a silent default. + pub fn config(self) -> Config { + Config { + data_dir: self.data_dir, + max_upload_bytes: self.max_upload_bytes, + default_expiry_seconds: self.default_expiry_seconds, + go_upstream: self.go_upstream, + python_upstream: self.python_upstream, + npm_upstream: self.npm_upstream, + cargo_index_upstream: self.cargo_index_upstream, + cargo_crate_upstream: self.cargo_crate_upstream, + proxy_revalidation_seconds: self.proxy_revalidation_seconds, + proxy_concurrency: self.proxy_concurrency, + upstream_timeout_seconds: self.upstream_timeout_seconds, + proxy_allowed_origins: self.proxy_allowed_origins, + foreground_concurrency: self.foreground_concurrency, + reservation_extent_bytes: self.reservation_extent_bytes, + low_watermark_bytes: self.low_watermark_bytes, + high_watermark_bytes: self.high_watermark_bytes, + emergency_headroom_bytes: self.emergency_headroom_bytes, + bloom_bits: self.bloom_bits, + reclaim_candidate_limit: self.reclaim_candidate_limit, + reclaim_byte_limit: self.reclaim_byte_limit, + orphan_scan_limit: self.orphan_scan_limit, + } } } diff --git a/src/config.rs b/src/config.rs index f4ce9ab..a5eadb4 100644 --- a/src/config.rs +++ b/src/config.rs @@ -29,6 +29,10 @@ pub struct Config { } impl Config { + /// Defaults for embedders and tests. The serve command does not go through here: + /// clap owns the operator-facing defaults in `cli.rs` (`#[arg(default_value_t)]`) + /// and `ServeArgs::config` builds the whole struct, so the two default sets are + /// deliberately separate and must be changed together. pub fn new(data_dir: impl AsRef) -> Self { Self { data_dir: data_dir.as_ref().to_path_buf(), diff --git a/src/main.rs b/src/main.rs index f874a04..1f55220 100644 --- a/src/main.rs +++ b/src/main.rs @@ -56,6 +56,8 @@ where async fn serve(arguments: flywheel::cli::ServeArgs) -> anyhow::Result<()> { let startup = Instant::now(); let listen = arguments.listen; + let data_dir = arguments.data_dir.clone(); + let foreground_concurrency = arguments.foreground_concurrency; let flywheel = Arc::new(Flywheel::open(arguments.config()).await?); let listener = tokio::net::TcpListener::bind(listen).await?; let cancellation = CancellationToken::new(); @@ -68,8 +70,8 @@ async fn serve(arguments: flywheel::cli::ServeArgs) -> anyhow::Result<()> { component = "server", version = env!("CARGO_PKG_VERSION"), %listen, - data_dir = %arguments.data_dir.display(), - foreground_concurrency = arguments.foreground_concurrency, + data_dir = %data_dir.display(), + foreground_concurrency, startup_ms = startup.elapsed().as_millis() as u64, "Flywheel is ready" ); diff --git a/tests/integration/cli.rs b/tests/integration/cli.rs index 04078dd..e3257f0 100644 --- a/tests/integration/cli.rs +++ b/tests/integration/cli.rs @@ -1,5 +1,6 @@ use clap::Parser; use flywheel::cli::{Cli, Command}; +use std::path::Path; #[test] fn exposes_agent_command() { @@ -41,3 +42,84 @@ fn exposes_serve_and_cacheprog_commands() { .expect("cacheprog command parses"); assert!(matches!(cacheprog.command, Command::Cacheprog(_))); } + +/// Every serve flag carries a value distinct from every other flag's, so a field wired +/// to the wrong argument — the failure mode the struct literal in `ServeArgs::config` +/// cannot catch — surfaces as a mismatch rather than a coincidence. +#[test] +fn serve_arguments_map_onto_the_configuration() { + let serve = Cli::try_parse_from([ + "flywheel", + "serve", + "--data-dir", + "/tmp/flywheel-mapping", + "--max-upload-bytes", + "11", + "--default-expiry-seconds", + "12", + "--foreground-concurrency", + "13", + "--reservation-extent-bytes", + "14", + "--low-watermark-bytes", + "15", + "--high-watermark-bytes", + "16", + "--emergency-headroom-bytes", + "17", + "--bloom-bits", + "18", + "--reclaim-candidate-limit", + "19", + "--reclaim-byte-limit", + "20", + "--orphan-scan-limit", + "21", + "--proxy-revalidation-seconds", + "22", + "--proxy-concurrency", + "23", + "--upstream-timeout-seconds", + "24", + "--go-upstream", + "https://go.example/", + "--python-upstream", + "https://python.example/", + "--npm-upstream", + "https://npm.example/", + "--cargo-index-upstream", + "https://cargo-index.example/", + "--cargo-crate-upstream", + "https://cargo-crate.example/", + "--proxy-allowed-origin", + "https://mirror.example", + ]) + .expect("serve command parses"); + let Command::Serve(arguments) = serve.command else { + panic!("expected the serve command"); + }; + let config = arguments.config(); + config.validate().expect("mapped configuration is valid"); + + assert_eq!(config.data_dir, Path::new("/tmp/flywheel-mapping")); + assert_eq!(config.max_upload_bytes, 11); + assert_eq!(config.default_expiry_seconds, 12); + assert_eq!(config.foreground_concurrency, 13); + assert_eq!(config.reservation_extent_bytes, 14); + assert_eq!(config.low_watermark_bytes, 15); + assert_eq!(config.high_watermark_bytes, 16); + assert_eq!(config.emergency_headroom_bytes, 17); + assert_eq!(config.bloom_bits, 18); + assert_eq!(config.reclaim_candidate_limit, 19); + assert_eq!(config.reclaim_byte_limit, 20); + assert_eq!(config.orphan_scan_limit, 21); + assert_eq!(config.proxy_revalidation_seconds, 22); + assert_eq!(config.proxy_concurrency, 23); + assert_eq!(config.upstream_timeout_seconds, 24); + assert_eq!(config.go_upstream, "https://go.example/"); + assert_eq!(config.python_upstream, "https://python.example/"); + assert_eq!(config.npm_upstream, "https://npm.example/"); + assert_eq!(config.cargo_index_upstream, "https://cargo-index.example/"); + assert_eq!(config.cargo_crate_upstream, "https://cargo-crate.example/"); + assert_eq!(config.proxy_allowed_origins, ["https://mirror.example"]); +}