build: tune the release profile, drop the no-op feature flag, and share test helpers

AGENTS.md

@@ -43,9 +43,12 @@ src/
 
 ### Integration tests
 
-Integration tests live in `tests/integration/`. Cargo auto-discovers them.
-They exercise full Flywheel instances with HTTP routers, real TCP, and
-tempfile-backed storage.
+Integration tests live in `tests/integration/`. Cargo does not auto-discover
+files in that subdirectory, so each one is declared as a `[[test]]` target in
+`Cargo.toml`; a new file needs a new entry there. They exercise full Flywheel
+instances with HTTP routers, real TCP, and tempfile-backed storage. Helpers
+shared between them live in `tests/integration/common/mod.rs`, pulled in with
+`#[path = "common/mod.rs"] mod common;` rather than declared as a target.
 
 ### What to test
 

Cargo.toml

@@ -43,6 +43,10 @@ url = "2.5.7"
 [dev-dependencies]
 tempfile = "3.23.0"
 
+[profile.release]
+lto = "thin"
+codegen-units = 1
+
 # Integration tests in tests/integration/
 [[test]]
 name = "integration_agent"

Makefile

@@ -1,5 +1,4 @@
 CARGO ?= cargo
-CARGO_FEATURES ?= --all-features
 INSTALL ?= install
 PREFIX ?= /usr/local
 BINDIR ?= $(PREFIX)/bin
@@ -21,18 +20,18 @@ endif
 
 .PHONY: build release install check fmt fmt-check lint test ci clean help
 
-build: ## Build Flywheel with every feature enabled.
-	$(CARGO) build $(CARGO_FEATURES)
+build: ## Build the development binary.
+	$(CARGO) build
 
 release: ## Build the optimized production binary from the locked dependency graph.
-	$(CARGO) build --release --locked $(CARGO_FEATURES)
+	$(CARGO) build --release --locked
 
 install: release ## Install the production binary (PREFIX=/usr/local by default).
 	$(INSTALL) -d "$(DESTDIR)$(BINDIR)"
 	$(INSTALL) -m 0755 "$(RELEASE_BINARY)" "$(DESTDIR)$(BINDIR)/flywheel"
 
 check: ## Type-check all targets without producing binaries.
-	$(CARGO) check --all-targets $(CARGO_FEATURES)
+	$(CARGO) check --all-targets
 
 fmt: ## Format Rust sources.
 	$(CARGO) fmt --all
@@ -41,10 +40,10 @@ fmt-check: ## Verify Rust formatting without changing files.
 	$(CARGO) fmt --all -- --check
 
 lint: ## Run Clippy and reject every warning.
-	$(CARGO) clippy --all-targets $(CARGO_FEATURES) -- -D warnings
+	$(CARGO) clippy --all-targets -- -D warnings
 
 test: ## Run the hermetic test suite.
-	$(CARGO) test $(CARGO_FEATURES)
+	$(CARGO) test
 
 ci: fmt-check lint test ## Run every CI quality gate.
 

src/agent/ring_test.rs

@@ -88,9 +88,9 @@ fn three_member_owners_match_frozen_vectors() {
 }
 
 #[test]
-fn kinds_place_independently_and_cas_shares_artifact_placement() {
+fn distinct_kinds_place_the_same_id_independently() {
     // The action record and an output that reuses the same hex string are distinct
-    // routing objects, while the raw-artifact and Bazel CAS routes share a kind.
+    // routing objects.
     assert_ne!(
         key_position("artifact", DIGEST_A),
         key_position("bazel-action", DIGEST_A)
@@ -120,7 +120,6 @@ fn construction_is_independent_of_discovery_order() {
 fn duplicate_member_ids_collapse_to_one_member() {
     let ring = Ring::new(vec![member("flywheel-0"), member("flywheel-0")]);
     assert_eq!(ring.members().len(), 1);
-    // One member with VIRTUAL_NODES virtual nodes
 }
 
 #[test]

src/cache/space_test.rs

@@ -12,9 +12,11 @@ impl FreeSpace for Fixed {
     }
 }
 
-fn ledger(free: u64) -> SpaceLedger {
+/// Every case here exercises the same watermark policy; only the free-space
+/// source differs.
+fn ledger_from(source: Arc<dyn FreeSpace>) -> SpaceLedger {
     SpaceLedger::new(
-        Arc::new(Fixed(free)),
+        source,
         SpacePolicy {
             low_watermark: 10,
             high_watermark: 20,
@@ -23,6 +25,10 @@ fn ledger(free: u64) -> SpaceLedger {
     )
 }
 
+fn ledger(free: u64) -> SpaceLedger {
+    ledger_from(Arc::new(Fixed(free)))
+}
+
 #[test]
 fn concurrent_reservations_never_double_spend_the_same_capacity() {
     let ledger = ledger(100);
@@ -50,14 +56,7 @@ fn mode_uses_low_high_watermark_hysteresis() {
             Some(self.0.load(Ordering::SeqCst))
         }
     }
-    let ledger = SpaceLedger::new(
-        Arc::new(Dynamic(Arc::clone(&source))),
-        SpacePolicy {
-            low_watermark: 10,
-            high_watermark: 20,
-            emergency_headroom: 0,
-        },
-    );
+    let ledger = ledger_from(Arc::new(Dynamic(Arc::clone(&source))));
     assert_eq!(ledger.mode(), Mode::Normal);
 
     // Below the low watermark enters Reclaiming.
@@ -84,14 +83,7 @@ fn failed_observation_fails_closed_and_reports_degraded() {
             None
         }
     }
-    let ledger = SpaceLedger::new(
-        Arc::new(Failing),
-        SpacePolicy {
-            low_watermark: 10,
-            high_watermark: 20,
-            emergency_headroom: 0,
-        },
-    );
+    let ledger = ledger_from(Arc::new(Failing));
     // An unreadable filesystem starts degraded and admits nothing.
     assert!(ledger.degraded());
     assert!(!ledger.try_reserve(1));
@@ -114,17 +106,10 @@ fn refresh_failure_retains_last_observation_but_stops_reservations() {
         }
     }
     let fail = Arc::new(std::sync::atomic::AtomicBool::new(false));
-    let ledger = SpaceLedger::new(
-        Arc::new(Maybe {
-            free: Arc::clone(&source),
-            fail: Arc::clone(&fail),
-        }),
-        SpacePolicy {
-            low_watermark: 10,
-            high_watermark: 20,
-            emergency_headroom: 0,
-        },
-    );
+    let ledger = ledger_from(Arc::new(Maybe {
+        free: Arc::clone(&source),
+        fail: Arc::clone(&fail),
+    }));
     assert!(ledger.try_reserve(10));
 
     // The observation now fails: the ledger keeps the last free value for metrics

src/cacheprog/cacheprog_test.rs

@@ -13,6 +13,7 @@ use axum::{
 use sha2::Digest as _;
 use std::{
     collections::HashMap,
+    net::SocketAddr,
     sync::{
         Arc, Mutex,
         atomic::{AtomicBool, AtomicUsize, Ordering},
@@ -25,6 +26,15 @@ use tokio::{
     sync::{Notify, oneshot},
 };
 
+/// Serves `app` on an ephemeral loopback port, returning the bound address and
+/// the server task the caller aborts when the test is done.
+async fn serve(app: Router) -> (SocketAddr, tokio::task::JoinHandle<()>) {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    let task = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+    (address, task)
+}
+
 fn args(url: String, cache_dir: &std::path::Path) -> CacheprogArgs {
     CacheprogArgs {
         url,
@@ -46,9 +56,7 @@ async fn ephemeral_cache_is_removed_after_close() {
         "/{*key}",
         route_get(|| async { StatusCode::NOT_FOUND }).put(|| async { StatusCode::OK }),
     );
-    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-    let address = listener.local_addr().unwrap();
-    let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+    let (address, server) = serve(app).await;
     let parent = tempfile::tempdir().unwrap();
     let mut options = args(format!("http://{address}/"), parent.path());
     options.ephemeral_cache = true;
@@ -124,9 +132,7 @@ async fn shutdown_cancels_in_flight_requests_and_removes_ephemeral_cache() {
             }
         }),
     );
-    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-    let address = listener.local_addr().unwrap();
-    let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+    let (address, server) = serve(app).await;
     let parent = tempfile::tempdir().unwrap();
     let mut options = args(format!("http://{address}/"), parent.path());
     options.ephemeral_cache = true;
@@ -271,9 +277,7 @@ async fn completes_independent_requests_concurrently() {
             StatusCode::NOT_FOUND
         }),
     );
-    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-    let address = listener.local_addr().unwrap();
-    let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+    let (address, server) = serve(app).await;
     let input = concat!(
         "{\"ID\":1,\"Command\":\"get\",\"ActionID\":\"AQ==\"}\n\n",
         "{\"ID\":2,\"Command\":\"get\",\"ActionID\":\"Ag==\"}\n\n",
@@ -470,9 +474,7 @@ async fn preserves_a_put_body_while_an_earlier_request_completes() {
         })
         .put(|| async { StatusCode::OK }),
     );
-    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-    let address = listener.local_addr().unwrap();
-    let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+    let (address, server) = serve(app).await;
     let cache_dir = tempfile::tempdir().unwrap();
     let (mut input, cache_input) = tokio::io::duplex(1);
     let (cache_output, output) = tokio::io::duplex(4096);
@@ -612,9 +614,7 @@ async fn spawn_prefetch_backend(
             ),
         )
         .with_state(backend);
-    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-    let address = listener.local_addr().unwrap();
-    let task = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+    let (address, task) = serve(app).await;
     (
         reqwest::Url::parse(&format!("http://{address}/build-cache/http/")).unwrap(),
         task,
@@ -746,9 +746,7 @@ async fn prefetch_concurrency_zero_fetches_only_the_manifest() {
             }
         }
     });
-    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-    let address = listener.local_addr().unwrap();
-    let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+    let (address, server) = serve(app).await;
     let cache_dir = tempfile::tempdir().unwrap();
     let (mut input, cache_input) = tokio::io::duplex(4096);
     let (cache_output, output) = tokio::io::duplex(4096);
@@ -789,104 +787,6 @@ async fn prefetch_concurrency_zero_fetches_only_the_manifest() {
     );
 }
 
-/// A foreground get for an object the prefetch pool is mid-download must wait
-/// for that download and answer locally — one fetch total, not two.
-#[tokio::test]
-async fn foreground_get_waits_for_the_inflight_prefetch_download() {
-    let body = b"contended object body".to_vec();
-    let action = format!("{:064x}", 9);
-    let mut manifest = Manifest::empty();
-    manifest
-        .entries
-        .insert(action.clone(), manifest_entry_for(&body));
-    let object_requests = Arc::new(AtomicUsize::new(0));
-    let download_started = Arc::new(Notify::new());
-    let release_download = Arc::new(Notify::new());
-    let app = Router::new().route(
-        "/build-cache/http/{key}",
-        route_get({
-            let manifest = manifest.clone();
-            let object_requests = Arc::clone(&object_requests);
-            let download_started = Arc::clone(&download_started);
-            let release_download = Arc::clone(&release_download);
-            let body = body.clone();
-            move |Path(key): Path<String>| {
-                let manifest = manifest.clone();
-                let object_requests = Arc::clone(&object_requests);
-                let download_started = Arc::clone(&download_started);
-                let release_download = Arc::clone(&release_download);
-                let body = body.clone();
-                async move {
-                    if key == manifest_key("wait-test") {
-                        return Json(manifest).into_response();
-                    }
-                    object_requests.fetch_add(1, Ordering::Relaxed);
-                    download_started.notify_one();
-                    release_download.notified().await;
-                    body.into_response()
-                }
-            }
-        }),
-    );
-    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-    let address = listener.local_addr().unwrap();
-    let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
-    let base = reqwest::Url::parse(&format!("http://{address}/build-cache/http/")).unwrap();
-    let client = reqwest::Client::new();
-    let directory = tempfile::tempdir().unwrap();
-    let state = Arc::new(session::SessionState::default());
-
-    let prefetch = tokio::spawn(session::run_prefetch(
-        client.clone(),
-        base.clone(),
-        None,
-        directory.path().to_path_buf(),
-        manifest_key("wait-test"),
-        1,
-        Arc::clone(&state),
-    ));
-    // The prefetch download is now in flight, blocked inside the server handler.
-    download_started.notified().await;
-
-    let releaser = tokio::spawn({
-        let release_download = Arc::clone(&release_download);
-        async move {
-            tokio::time::sleep(Duration::from_millis(100)).await;
-            release_download.notify_one();
-        }
-    });
-    let response = super::get(
-        &client,
-        &base,
-        None,
-        directory.path(),
-        &state,
-        Request {
-            id: 4,
-            command: "get".into(),
-            action_id: hex::decode(&action).unwrap(),
-            output_id: Vec::new(),
-            body_size: 0,
-        },
-    )
-    .await
-    .unwrap();
-    prefetch.await.unwrap();
-    releaser.await.unwrap();
-    server.abort();
-
-    assert!(!response.miss);
-    assert_eq!(
-        hex::encode(&response.output_id),
-        hex::encode(sha2::Sha256::digest(&body))
-    );
-    assert_eq!(
-        object_requests.load(Ordering::Relaxed),
-        1,
-        "the waiting get must reuse the prefetch download, not repeat it"
-    );
-}
-
 /// A foreground get for an output whose download is still queued behind the
 /// saturated pool must also wait: the in-flight marker is registered when the
 /// download is enqueued, not when it reaches the semaphore.
@@ -942,9 +842,7 @@ async fn foreground_get_waits_for_a_queued_prefetch_download() {
             }
         }),
     );
-    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-    let address = listener.local_addr().unwrap();
-    let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+    let (address, server) = serve(app).await;
     let base = reqwest::Url::parse(&format!("http://{address}/build-cache/http/")).unwrap();
     let client = reqwest::Client::new();
     let directory = tempfile::tempdir().unwrap();

src/telemetry.rs

@@ -371,6 +371,14 @@ impl<R: PrefetchRecorder> PrefetchObservation<R> {
     fn add_bytes(&mut self, bytes: usize) {
         self.bytes = self.bytes.saturating_add(bytes as u64);
     }
+
+    /// Marks the transfer as having reached the end of the body. `Drop` reads this to
+    /// tell a completed transfer from a cancelled one, so it is written through a
+    /// method: assigning the field directly in the stream's final arm reads as a dead
+    /// store to a newer rustc, which `-D warnings` then rejects.
+    fn complete(&mut self) {
+        self.completed = true;
+    }
 }
 
 impl<R: PrefetchRecorder> Drop for PrefetchObservation<R> {
@@ -414,7 +422,7 @@ where
                 }
                 Some(Err(error)) => Some((Err(error), (stream, observation))),
                 None => {
-                    observation.completed = true;
+                    observation.complete();
                     None
                 }
             }