diff --git a/src/agent/discovery.rs b/src/agent/discovery.rs index cb05b1a..ca01705 100644 --- a/src/agent/discovery.rs +++ b/src/agent/discovery.rs @@ -162,6 +162,12 @@ impl RingState { } } let mut inner = self.inner.write().expect("ring state lock"); + // Whoever held the write lock first may already have re-admitted; the + // rest take the continuum it built rather than each rehashing the whole + // membership while the lock blocks all routing. + if inner.earliest_retry.is_none_or(|deadline| now < deadline) { + return Arc::clone(&inner.ring); + } rebuild(&mut inner, now); Arc::clone(&inner.ring) } diff --git a/src/agent/discovery_test.rs b/src/agent/discovery_test.rs index 2f5b44d..854c73e 100644 --- a/src/agent/discovery_test.rs +++ b/src/agent/discovery_test.rs @@ -100,6 +100,43 @@ fn ejected_member_becomes_eligible_after_the_retry_timeout() { assert_eq!(state.ring().members().len(), 2); } +/// Re-admission rebuilds the continuum once, not once per concurrent request. +/// Every caller that queued on the write lock re-checks the retry deadline, so +/// a degraded cluster does not pay one full rebuild per in-flight request while +/// the lock blocks all routing; the identical `Arc` proves a single rebuild. +#[test] +fn concurrent_readmission_rebuilds_the_continuum_once() { + let clock = Arc::new(TestClock(AtomicU64::new(0))); + let state = state_with(&clock, 1, 30, three_members()); + for _ in 0..16 { + assert!(state.record_failure("flywheel-1")); + clock.advance(30); + let barrier = std::sync::Barrier::new(8); + let rings: Vec<_> = std::thread::scope(|scope| { + let handles: Vec<_> = (0..8) + .map(|_| { + let (state, barrier) = (&state, &barrier); + scope.spawn(move || { + barrier.wait(); + state.ring() + }) + }) + .collect(); + handles + .into_iter() + .map(|handle| handle.join().unwrap()) + .collect() + }); + for ring in &rings { + assert_eq!(ring.members().len(), 3); + assert!( + Arc::ptr_eq(ring, &rings[0]), + "every concurrent caller must observe the one rebuilt continuum" + ); + } + } +} + #[test] fn success_resets_the_consecutive_failure_count() { let clock = Arc::new(TestClock(AtomicU64::new(0))); diff --git a/src/cacheprog/cacheprog_test.rs b/src/cacheprog/cacheprog_test.rs index 6f9ab37..9d4a0cf 100644 --- a/src/cacheprog/cacheprog_test.rs +++ b/src/cacheprog/cacheprog_test.rs @@ -1,7 +1,7 @@ use super::{Request, Response, read_protocol_line, run_with_io, run_with_shutdown, session}; use crate::{ cli::CacheprogArgs, - manifest::{MANIFEST_VERSION, Manifest, ManifestEntry, manifest_key}, + manifest::{MANIFEST_MAX_AGE_SECONDS, MANIFEST_VERSION, Manifest, ManifestEntry, manifest_key}, }; use axum::{ Json, Router, @@ -385,6 +385,65 @@ async fn manifest_known_gets_answer_locally_without_any_request() { assert!(std::path::Path::new(&response.disk_path).is_file()); } +/// A get answered out of the local object directory is still usage. The merged +/// manifest has to carry it forward with a fresh `last_seen`, or the actions +/// prefetch predicts perfectly are exactly the ones that age out and get +/// evicted first — the manifest would decay to only what it mispredicted. +#[tokio::test] +async fn locally_answered_gets_refresh_their_manifest_entry() { + // Unreachable on purpose: a local answer must never touch the network. + let base = reqwest::Url::parse("http://127.0.0.1:9/build-cache/http/").unwrap(); + let client = reqwest::Client::new(); + let cache_dir = tempfile::tempdir().unwrap(); + let body = b"perfectly predicted object"; + let output = hex::encode(sha2::Sha256::digest(body)); + tokio::fs::write(cache_dir.path().join(&output), body) + .await + .unwrap(); + let action = hex::encode([9u8; 32]); + let stored = Manifest { + version: MANIFEST_VERSION, + entries: HashMap::from([( + action.clone(), + ManifestEntry { + output: output.clone(), + size: body.len() as u64, + last_seen: 100, + }, + )]), + }; + let session = session::SessionState::default(); + session.manifest.set(stored.clone()).unwrap(); + + let response = super::get( + &client, + &base, + None, + cache_dir.path(), + &session, + Request { + id: 3, + command: "get".into(), + action_id: hex::decode(&action).unwrap(), + output_id: Vec::new(), + body_size: 0, + }, + ) + .await + .unwrap(); + assert!(!response.miss); + + // Finalize past the retention window: only a recorded use keeps the entry. + let now = 100 + MANIFEST_MAX_AGE_SECONDS + 1; + let used = session.used.lock().unwrap().clone(); + let merged = session::merge_manifest(Some(stored), &used, now); + assert_eq!( + merged.entries.get(&action).map(|entry| entry.last_seen), + Some(now), + "a locally answered get must refresh its manifest entry" + ); +} + #[tokio::test] async fn preserves_a_put_body_while_an_earlier_request_completes() { let get_started = Arc::new(Notify::new()); diff --git a/src/cacheprog/mod.rs b/src/cacheprog/mod.rs index 287d984..150db64 100644 --- a/src/cacheprog/mod.rs +++ b/src/cacheprog/mod.rs @@ -441,6 +441,10 @@ async fn get( } if file_has_size(&path, entry.size).await { touch(&path); + // A local answer is still a use: without this the manifest retains the + // entry against its stale `last_seen`, so the best-predicted actions + // are the first to age out and the first evicted at the entry cap. + session.record_used(action, entry.output.clone(), entry.size); return Ok(Response { id: request.id, output_id: hex::decode(&entry.output)?, diff --git a/src/proxy/mod.rs b/src/proxy/mod.rs index b5fab2b..1bf178c 100644 --- a/src/proxy/mod.rs +++ b/src/proxy/mod.rs @@ -242,9 +242,10 @@ fn negotiate_npm(accept: Option<&str>) -> Option { (None, None) => None, (Some(_), None) => Some(true), (None, Some(_)) => Some(false), - (Some((specificity, quality)), Some((_, full_quality))) => { - Some(quality > full_quality || (quality == full_quality && specificity > 0)) - } + (Some((specificity, quality)), Some((full_specificity, full_quality))) => Some( + quality > full_quality + || (quality == full_quality && specificity > 0 && specificity >= full_specificity), + ), } } diff --git a/src/storage/local/artifact_files.rs b/src/storage/local/artifact_files.rs index ecc7f0c..d4d866e 100644 --- a/src/storage/local/artifact_files.rs +++ b/src/storage/local/artifact_files.rs @@ -187,7 +187,9 @@ impl Reservation { impl Drop for Reservation { fn drop(&mut self) { - if self.outstanding == 0 { + // A zero-length body reserves nothing yet still stages a file, so the + // shortcut has to clear the filesystem state as well as the accounting. + if self.outstanding == 0 && !matches!(self.state, ReservationState::Temporary(_)) { return; } let state = std::mem::replace(&mut self.state, ReservationState::Vacant); diff --git a/src/storage/local/artifact_files/tests.rs b/src/storage/local/artifact_files/tests.rs index bbfc2b9..35ccf1d 100644 --- a/src/storage/local/artifact_files/tests.rs +++ b/src/storage/local/artifact_files/tests.rs @@ -201,6 +201,41 @@ async fn cancelling_staging_schedules_cleanup_and_reuses_capacity() { assert!(matches!(retry, StageOutcome::Ready(_))); } +/// A zero-length body reserves nothing, but staging still creates its `.part` +/// file. Dropping the stage has to remove that file: cleanup keys off the +/// filesystem state, not off the reserved byte count. +#[tokio::test] +async fn dropping_a_zero_length_stage_removes_its_temporary_file() { + let directory = TempDir::new().unwrap(); + let files = ArtifactFiles::open(directory.path(), 4).await.unwrap(); + let accounting = Arc::new(Accounting::new(4)); + let outcome = files + .stage( + ChannelId::DEFAULT, + body([]), + 4, + Some(0), + accounting.clone(), + Durability::BestEffort, + StoredEncoding::Identity, + ) + .await + .unwrap(); + let StageOutcome::Ready(staged) = outcome else { + panic!("an empty body always fits"); + }; + assert_eq!(staged.len, 0); + assert_eq!(temporary_file_count(directory.path()), 1); + + drop(staged); + + accounting + .wait_for(|| temporary_file_count(directory.path()) == 0) + .await; + assert_eq!(accounting.reserved(), 0); + assert_eq!(accounting.committed(), 0); +} + #[cfg(unix)] #[tokio::test] async fn failed_deletion_conservatively_commits_capacity() { diff --git a/tests/integration/build_cache.rs b/tests/integration/build_cache.rs index d913f3f..d30e703 100644 --- a/tests/integration/build_cache.rs +++ b/tests/integration/build_cache.rs @@ -522,16 +522,29 @@ async fn warm_build_costs_one_manifest_get_plus_one_get_per_distinct_output() { ); } - // The complete warm-build request set: one manifest GET, no request for any - // zero-size action, one GET per distinct nonzero output, nothing repeated. + // The complete warm-build request set: the bootstrap manifest GET, no + // request for any zero-size action, one GET per distinct nonzero output, + // nothing repeated, and the close-time read-modify-write that carries the + // locally answered actions forward with a fresh `last_seen`. let warm = requests.lock().unwrap()[populate_requests..].to_vec(); - assert_eq!(warm.len(), 4, "exactly the bootstrap traffic: {warm:?}"); + assert_eq!( + warm.len(), + 6, + "the bootstrap traffic plus the manifest write-back: {warm:?}" + ); assert_eq!( warm.iter() .filter(|(method, path)| method == "GET" && path == &manifest_path) .count(), + 2, + "one bootstrap read and one finalize read: {warm:?}" + ); + assert_eq!( + warm.iter() + .filter(|(method, path)| method == "PUT" && path == &manifest_path) + .count(), 1, - "warm requests: {warm:?}" + "a wholly local build still refreshes its manifest: {warm:?}" ); for action in &zero_actions { let path = format!("/build-cache/http/go-{}", hex::encode(action)); diff --git a/tests/integration/package_proxies.rs b/tests/integration/package_proxies.rs index 060a6bd..e0fbbf8 100644 --- a/tests/integration/package_proxies.rs +++ b/tests/integration/package_proxies.rs @@ -568,6 +568,125 @@ async fn npm_negotiates_and_separately_caches_install_and_full_metadata() { server.abort(); } +/// A wildcard range never outranks a representation the client named explicitly +/// at the same quality: `application/json, application/*` asks for the full +/// document and merely tolerates the rest, so it must not be answered with the +/// abbreviated one. Only an explicit q-value reorders the two. +#[tokio::test] +async fn wildcard_accept_ranges_never_outrank_a_named_representation() { + const NPM_INSTALL: &str = "application/vnd.npm.install-v1+json"; + const NPM_FULL: &str = "application/json"; + const PYTHON_JSON: &str = "application/vnd.pypi.simple.v1+json"; + const PYTHON_HTML: &str = "application/vnd.pypi.simple.v1+html"; + + let upstream = Router::new() + .route( + "/npm/demo", + get(|headers: HeaderMap| async move { + match headers + .get(header::ACCEPT) + .and_then(|value| value.to_str().ok()) + { + Some(NPM_INSTALL) => ( + [(header::CONTENT_TYPE, NPM_INSTALL)], + json!({"name": "demo", "dist-tags": {"latest": "1.0.0"}}).to_string(), + ) + .into_response(), + Some(NPM_FULL) => ( + [(header::CONTENT_TYPE, NPM_FULL)], + json!({"name": "demo", "readme": "full metadata"}).to_string(), + ) + .into_response(), + _ => StatusCode::NOT_ACCEPTABLE.into_response(), + } + }), + ) + .route( + "/python/demo/", + get(|headers: HeaderMap| async move { + match headers + .get(header::ACCEPT) + .and_then(|value| value.to_str().ok()) + { + Some(PYTHON_JSON) => ( + [(header::CONTENT_TYPE, PYTHON_JSON)], + json!({"meta":{"api-version":"1.0"},"name":"demo","files":[]}).to_string(), + ) + .into_response(), + Some(PYTHON_HTML) => ( + [(header::CONTENT_TYPE, PYTHON_HTML)], + "", + ) + .into_response(), + _ => StatusCode::NOT_ACCEPTABLE.into_response(), + } + }), + ); + let (base, server) = serve(upstream).await; + let directory = TempDir::new().unwrap(); + let mut config = Config::new(directory.path()); + config.npm_upstream = format!("{base}/npm/"); + config.python_upstream = format!("{base}/python/"); + let app = Flywheel::open(config).await.unwrap().router(); + + for (accept, expected) in [ + ("application/json, application/*", NPM_FULL), + ("application/*, application/json", NPM_FULL), + ("application/json, */*", NPM_FULL), + ( + "application/vnd.npm.install-v1+json, application/*", + NPM_INSTALL, + ), + ( + "application/*, application/vnd.npm.install-v1+json", + NPM_INSTALL, + ), + // A bare wildcard names neither document, so npm's own abbreviated + // install metadata stays the answer. + ("application/*", NPM_INSTALL), + // Specificity only breaks ties: an explicit q-value still decides. + ("application/json;q=0.2, application/*", NPM_INSTALL), + ] { + let response = call_accepting(app.clone(), "/proxy/npm/demo", accept).await; + assert_eq!(response.status(), StatusCode::OK, "Accept: {accept}"); + assert_eq!( + response.headers()[header::CONTENT_TYPE], + expected, + "Accept: {accept}" + ); + } + + for (accept, expected) in [ + ( + "application/vnd.pypi.simple.v1+json, application/*", + PYTHON_JSON, + ), + ( + "application/*, application/vnd.pypi.simple.v1+json", + PYTHON_JSON, + ), + ( + "application/vnd.pypi.simple.v1+html, application/*", + PYTHON_HTML, + ), + ("text/html, */*", PYTHON_HTML), + ( + "application/vnd.pypi.simple.v1+json;q=0.2, text/*", + PYTHON_HTML, + ), + ] { + let response = call_accepting(app.clone(), "/proxy/python/simple/demo/", accept).await; + assert_eq!(response.status(), StatusCode::OK, "Accept: {accept}"); + assert_eq!( + response.headers()[header::CONTENT_TYPE], + expected, + "Accept: {accept}" + ); + } + + server.abort(); +} + #[tokio::test] async fn protected_cargo_registry_accepts_credential_provider_authorization() { let directory = TempDir::new().unwrap();