fix: correct four latent defects in negotiation, caching, staging, and routing

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)
     }

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)));

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());

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)?,

src/proxy/mod.rs

@@ -242,9 +242,10 @@ fn negotiate_npm(accept: Option<&str>) -> Option<bool> {
         (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),
+        ),
     }
 }
 

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);

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() {

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));

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)],
+                        "<!doctype html><html><body></body></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();