From 69cb6b6fbeeb11572bb82bffe822d12e2e2972ef Mon Sep 17 00:00:00 2001 From: Rob Lyon Date: Tue, 21 Jul 2026 10:15:05 -0700 Subject: [PATCH] build: tune the release profile, drop the no-op feature flag, and share test helpers The release profile was never set, so a performance-sensitive server shipped untuned; it now uses thin LTO and a single codegen unit. CARGO_FEATURES was --all-features against a crate with no [features] table, a no-op on every target. AGENTS.md claimed Cargo auto-discovers the integration tests, but each is an explicit [[test]] target because they live in a subdirectory. Test scaffolding: the oneshot call wrapper duplicated across four integration suites moves to a shared common module, the bind-and-serve block repeated eight times in cacheprog_test.rs becomes one serve helper, and the SpacePolicy literal repeated four times in space_test.rs becomes ledger_from. Deletes foreground_get_waits_for_the_inflight_prefetch_download, which the queued variant that follows it strictly subsumes: begin_download runs before the semaphore acquire, so both tests observe the same entry through the same wait, and the queued one additionally asserts the get actually blocked. Also writes PrefetchObservation's completion flag through a method. Drop reads it, so the behaviour was already correct -- confirmed by driving a body to completion and asserting the recorder sees completed -- but a newer rustc reads the direct field assignment in the stream's final arm as a dead store, which would fail clippy -D warnings on the toolchain the container builds with. --- AGENTS.md | 9 +- Cargo.toml | 4 + Makefile | 13 ++- src/agent/ring_test.rs | 5 +- src/cache/space_test.rs | 43 +++------ src/cacheprog/cacheprog_test.rs | 136 ++++----------------------- src/telemetry.rs | 10 +- tests/integration/build_cache.rs | 7 +- tests/integration/channels.rs | 7 +- tests/integration/common/mod.rs | 11 +++ tests/integration/local_artifacts.rs | 79 ++++++++-------- tests/integration/maintenance.rs | 29 +++--- 12 files changed, 129 insertions(+), 224 deletions(-) create mode 100644 tests/integration/common/mod.rs diff --git a/AGENTS.md b/AGENTS.md index b246e43..092dbb8 100644 --- a/AGENTS.md +++ b/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 diff --git a/Cargo.toml b/Cargo.toml index 8b12783..558c42b 100644 --- a/Cargo.toml +++ b/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" diff --git a/Makefile b/Makefile index f04f6d4..33457c0 100644 --- a/Makefile +++ b/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. diff --git a/src/agent/ring_test.rs b/src/agent/ring_test.rs index 785f883..7cf46b5 100644 --- a/src/agent/ring_test.rs +++ b/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] diff --git a/src/cache/space_test.rs b/src/cache/space_test.rs index bd0f209..9686569 100644 --- a/src/cache/space_test.rs +++ b/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) -> 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 diff --git a/src/cacheprog/cacheprog_test.rs b/src/cacheprog/cacheprog_test.rs index 9d4a0cf..b5c67d1 100644 --- a/src/cacheprog/cacheprog_test.rs +++ b/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| { - 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(); diff --git a/src/telemetry.rs b/src/telemetry.rs index 5232163..69c301c 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -371,6 +371,14 @@ impl PrefetchObservation { 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 Drop for PrefetchObservation { @@ -414,7 +422,7 @@ where } Some(Err(error)) => Some((Err(error), (stream, observation))), None => { - observation.completed = true; + observation.complete(); None } } diff --git a/tests/integration/build_cache.rs b/tests/integration/build_cache.rs index d30e703..2bcba96 100644 --- a/tests/integration/build_cache.rs +++ b/tests/integration/build_cache.rs @@ -19,11 +19,10 @@ use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, sync::Notify, }; -use tower::ServiceExt; -async fn call(app: axum::Router, request: Request) -> axum::response::Response { - app.oneshot(request).await.unwrap() -} +#[path = "common/mod.rs"] +mod common; +use common::call; #[tokio::test] async fn generic_http_cache_replaces_opaque_keys_with_immutable_content() { diff --git a/tests/integration/channels.rs b/tests/integration/channels.rs index 589739b..8eb8508 100644 --- a/tests/integration/channels.rs +++ b/tests/integration/channels.rs @@ -12,11 +12,10 @@ use rocksdb::{DB, IteratorMode, Options}; use serde_json::{Value, json}; use sha2::{Digest as _, Sha256}; use tempfile::TempDir; -use tower::ServiceExt; -async fn call(app: axum::Router, request: Request) -> axum::response::Response { - app.oneshot(request).await.unwrap() -} +#[path = "common/mod.rs"] +mod common; +use common::call; async fn register(app: axum::Router, protected: bool) -> Value { let response = call( diff --git a/tests/integration/common/mod.rs b/tests/integration/common/mod.rs new file mode 100644 index 0000000..5e01f0d --- /dev/null +++ b/tests/integration/common/mod.rs @@ -0,0 +1,11 @@ +//! Helpers shared by the integration test targets. This file is not a test +//! target of its own: each test that needs it pulls it in with +//! `#[path = "common/mod.rs"] mod common;`. + +use axum::{body::Body, http::Request, response::Response}; +use tower::ServiceExt as _; + +/// Drives one request through a router without binding a port. +pub async fn call(app: axum::Router, request: Request) -> Response { + app.oneshot(request).await.unwrap() +} diff --git a/tests/integration/local_artifacts.rs b/tests/integration/local_artifacts.rs index 1db4663..e427a74 100644 --- a/tests/integration/local_artifacts.rs +++ b/tests/integration/local_artifacts.rs @@ -8,16 +8,15 @@ use sha2::{Digest as _, Sha256}; use std::{convert::Infallible, sync::Arc, time::Duration}; use tempfile::TempDir; use tokio::sync::Notify; -use tower::ServiceExt; + +#[path = "common/mod.rs"] +mod common; +use common::call; fn digest(body: &[u8]) -> String { hex::encode(Sha256::digest(body)) } -async fn request(app: axum::Router, request: Request) -> axum::response::Response { - app.oneshot(request).await.unwrap() -} - #[tokio::test] async fn responses_propagate_or_create_request_ids() { let directory = TempDir::new().unwrap(); @@ -26,7 +25,7 @@ async fn responses_propagate_or_create_request_ids() { .unwrap() .router(); - let generated = request( + let generated = call( app.clone(), Request::get("/health/ready").body(Body::empty()).unwrap(), ) @@ -34,7 +33,7 @@ async fn responses_propagate_or_create_request_ids() { let generated = generated.headers()["x-request-id"].to_str().unwrap(); assert!(ulid::Ulid::from_string(generated).is_ok()); - let provided = request( + let provided = call( app, Request::get("/health/ready") .header("x-request-id", "caller-request") @@ -56,7 +55,7 @@ async fn publishes_streams_ranges_and_recovers_local_artifacts() { let flywheel = Flywheel::open(Config::new(directory.path())).await.unwrap(); let app = flywheel.router(); - let response = request( + let response = call( app.clone(), Request::put(&path) .header(header::CONTENT_TYPE, "application/octet-stream") @@ -66,7 +65,7 @@ async fn publishes_streams_ranges_and_recovers_local_artifacts() { .await; assert_eq!(response.status(), StatusCode::CREATED); - let duplicate = request( + let duplicate = call( app.clone(), Request::put(&path) .body(Body::from(body.as_slice())) @@ -75,14 +74,14 @@ async fn publishes_streams_ranges_and_recovers_local_artifacts() { .await; assert_eq!(duplicate.status(), StatusCode::NO_CONTENT); - let corrupt = request( + let corrupt = call( app.clone(), Request::put(&path).body(Body::from("different")).unwrap(), ) .await; assert_eq!(corrupt.status(), StatusCode::CONFLICT); - let response = request( + let response = call( app.clone(), Request::get(&path).body(Body::empty()).unwrap(), ) @@ -101,7 +100,7 @@ async fn publishes_streams_ranges_and_recovers_local_artifacts() { body.as_slice() ); - let range = request( + let range = call( app.clone(), Request::get(&path) .header(header::RANGE, "bytes=6-12") @@ -119,7 +118,7 @@ async fn publishes_streams_ranges_and_recovers_local_artifacts() { &body[6..=12] ); - let head = request(app, Request::head(&path).body(Body::empty()).unwrap()).await; + let head = call(app, Request::head(&path).body(Body::empty()).unwrap()).await; assert_eq!(head.status(), StatusCode::OK); assert_eq!( head.headers()[header::CONTENT_LENGTH], @@ -134,7 +133,7 @@ async fn publishes_streams_ranges_and_recovers_local_artifacts() { } let recovered = Flywheel::open(Config::new(directory.path())).await.unwrap(); - let response = request( + let response = call( recovered.router(), Request::get(&path).body(Body::empty()).unwrap(), ) @@ -156,7 +155,7 @@ async fn identity_ranges_ignore_invalid_syntax_and_reject_only_non_overlapping_r let body = b"0123456789"; let path = format!("/artifacts/sha256/{}", digest(body)); assert_eq!( - request( + call( app.clone(), Request::put(&path) .body(Body::from(body.as_slice())) @@ -167,7 +166,7 @@ async fn identity_ranges_ignore_invalid_syntax_and_reject_only_non_overlapping_r StatusCode::CREATED ); - let range = request( + let range = call( app.clone(), Request::get(&path) .header(header::RANGE, "bytes=2-5") @@ -191,7 +190,7 @@ async fn identity_ranges_ignore_invalid_syntax_and_reject_only_non_overlapping_r "bytes=0-1,4-5", "bytes=5-4", ] { - let response = request( + let response = call( app.clone(), Request::get(&path) .header(header::RANGE, invalid) @@ -210,7 +209,7 @@ async fn identity_ranges_ignore_invalid_syntax_and_reject_only_non_overlapping_r } for unsatisfiable in ["bytes=20-30", "bytes=-0"] { - let response = request( + let response = call( app.clone(), Request::get(&path) .header(header::RANGE, unsatisfiable) @@ -227,7 +226,7 @@ async fn identity_ranges_ignore_invalid_syntax_and_reject_only_non_overlapping_r assert_eq!(response.headers()[header::CONTENT_RANGE], "bytes */10"); } - let head = request( + let head = call( app.clone(), Request::head(&path) .header(header::RANGE, "bytes=2-5") @@ -248,7 +247,7 @@ async fn identity_ranges_ignore_invalid_syntax_and_reject_only_non_overlapping_r // GET-only build-cache routes are also invoked for HEAD by axum; they must pass // that method through so the shared artifact responder ignores Range. - let cas_head = request( + let cas_head = call( app, Request::head(format!("/build-cache/bazel/cas/{}", digest(body))) .header(header::RANGE, "bytes=2-5") @@ -271,7 +270,7 @@ async fn empty_identity_representation_has_only_unsatisfiable_valid_ranges() { .router(); let path = format!("/artifacts/sha256/{}", digest(b"")); assert_eq!( - request( + call( app.clone(), Request::put(&path).body(Body::empty()).unwrap() ) @@ -280,7 +279,7 @@ async fn empty_identity_representation_has_only_unsatisfiable_valid_ranges() { StatusCode::CREATED ); - let unsatisfiable = request( + let unsatisfiable = call( app.clone(), Request::get(&path) .header(header::RANGE, "bytes=0-0") @@ -291,7 +290,7 @@ async fn empty_identity_representation_has_only_unsatisfiable_valid_ranges() { assert_eq!(unsatisfiable.status(), StatusCode::RANGE_NOT_SATISFIABLE); assert_eq!(unsatisfiable.headers()[header::CONTENT_RANGE], "bytes */0"); - let malformed = request( + let malformed = call( app, Request::get(&path) .header(header::RANGE, "bytes=garbage") @@ -336,7 +335,7 @@ async fn concurrent_duplicate_put_commits_independently_before_the_leader() { Ok::<_, Infallible>(body) } }); - let leader = tokio::spawn(request( + let leader = tokio::spawn(call( app.clone(), Request::put(&path) .body(Body::from_stream(leader_stream)) @@ -348,7 +347,7 @@ async fn concurrent_duplicate_put_commits_independently_before_the_leader() { // first to commit. let follower = tokio::time::timeout( Duration::from_secs(1), - request( + call( app.clone(), Request::put(&path).body(Body::from(body.clone())).unwrap(), ), @@ -357,7 +356,7 @@ async fn concurrent_duplicate_put_commits_independently_before_the_leader() { .expect("a duplicate PUT must not wait for the in-flight leader"); assert_eq!(follower.status(), StatusCode::CREATED); - let after_follower = request( + let after_follower = call( app.clone(), Request::get(&path).body(Body::empty()).unwrap(), ) @@ -373,7 +372,7 @@ async fn concurrent_duplicate_put_commits_independently_before_the_leader() { // The leader finishes second, finds the committed row, and reports the duplicate. release_leader_body.notify_one(); assert_eq!(leader.await.unwrap().status(), StatusCode::NO_CONTENT); - let after_leader = request( + let after_leader = call( app.clone(), Request::get(&path).body(Body::empty()).unwrap(), ) @@ -426,7 +425,7 @@ async fn failed_publication_does_not_block_a_later_put() { let failing_stream = stream::once(async { Err::(std::io::Error::other("injected upload failure")) }); - let failed = request( + let failed = call( app.clone(), Request::put(&path) .body(Body::from_stream(failing_stream)) @@ -435,13 +434,13 @@ async fn failed_publication_does_not_block_a_later_put() { .await; assert_eq!(failed.status(), StatusCode::INTERNAL_SERVER_ERROR); - let retry = request( + let retry = call( app.clone(), Request::put(&path).body(Body::from(body.clone())).unwrap(), ) .await; assert_eq!(retry.status(), StatusCode::CREATED); - let stored = request(app, Request::get(&path).body(Body::empty()).unwrap()).await; + let stored = call(app, Request::get(&path).body(Body::empty()).unwrap()).await; assert_eq!(stored.status(), StatusCode::OK); assert_eq!( to_bytes(stored.into_body(), usize::MAX).await.unwrap(), @@ -460,7 +459,7 @@ async fn logical_references_are_atomic_and_retryable() { let digest = digest(body); let artifact = format!("/artifacts/sha256/{digest}"); assert_eq!( - request( + call( app.clone(), Request::put(&artifact) .body(Body::from(body.as_slice())) @@ -473,7 +472,7 @@ async fn logical_references_are_atomic_and_retryable() { let binding = serde_json::json!({"algorithm": "sha256", "digest": digest}); for _ in 0..2 { - let response = request( + let response = call( app.clone(), Request::put("/references/toolchain") .header(header::CONTENT_TYPE, "application/json") @@ -484,7 +483,7 @@ async fn logical_references_are_atomic_and_retryable() { assert_eq!(response.status(), StatusCode::NO_CONTENT); } - let response = request( + let response = call( app.clone(), Request::get("/references/toolchain") .body(Body::empty()) @@ -502,7 +501,7 @@ async fn logical_references_are_atomic_and_retryable() { for _ in 0..2 { assert_eq!( - request( + call( app.clone(), Request::delete("/references/toolchain") .body(Body::empty()) @@ -526,7 +525,7 @@ async fn get_self_heals_metadata_pointing_at_a_missing_body() { let digest = digest(body); let path = format!("/artifacts/sha256/{digest}"); assert_eq!( - request( + call( app.clone(), Request::put(&path) .body(Body::from(body.as_slice())) @@ -548,7 +547,7 @@ async fn get_self_heals_metadata_pointing_at_a_missing_body() { for _ in 0..2 { assert_eq!( - request( + call( app.clone(), Request::get(&path).body(Body::empty()).unwrap() ) @@ -602,7 +601,7 @@ async fn streaming_get_holds_the_transport_budget_and_sheds_with_429() { let body = b"budgeted body"; let path = format!("/artifacts/sha256/{}", digest(body)); - let put = request( + let put = call( app.clone(), Request::put(&path) .body(Body::from(body.as_slice())) @@ -613,14 +612,14 @@ async fn streaming_get_holds_the_transport_budget_and_sheds_with_429() { // The permit is moved into the body stream when the response is built, so // holding the unconsumed response deterministically holds the budget's only slot. - let held = request( + let held = call( app.clone(), Request::get(&path).body(Body::empty()).unwrap(), ) .await; assert_eq!(held.status(), StatusCode::OK); - let shed = request( + let shed = call( app.clone(), Request::get(&path).body(Body::empty()).unwrap(), ) @@ -629,7 +628,7 @@ async fn streaming_get_holds_the_transport_budget_and_sheds_with_429() { assert_eq!(shed.headers()[header::RETRY_AFTER], "1"); drop(held); - let served = request(app, Request::get(&path).body(Body::empty()).unwrap()).await; + let served = call(app, Request::get(&path).body(Body::empty()).unwrap()).await; assert_eq!(served.status(), StatusCode::OK); assert_eq!( to_bytes(served.into_body(), usize::MAX).await.unwrap(), diff --git a/tests/integration/maintenance.rs b/tests/integration/maintenance.rs index 2958d53..178da7e 100644 --- a/tests/integration/maintenance.rs +++ b/tests/integration/maintenance.rs @@ -10,7 +10,10 @@ use std::sync::{ atomic::{AtomicU64, Ordering}, }; use tempfile::TempDir; -use tower::ServiceExt; + +#[path = "common/mod.rs"] +mod common; +use common::call; struct ManualClock(AtomicU64); @@ -54,7 +57,7 @@ fn artifact(body: &[u8]) -> (String, String) { } async fn status(app: axum::Router, request: Request) -> StatusCode { - app.oneshot(request).await.unwrap().status() + call(app, request).await.status() } // A large free-space configuration that keeps the controller firmly in Normal mode so @@ -174,18 +177,16 @@ async fn maintenance_uses_each_active_channels_persisted_expiry() { .await .unwrap(); let app = flywheel.router(); - let registered = app - .clone() - .oneshot( - Request::post("/channels") - .header(header::CONTENT_TYPE, "application/json") - .body(Body::from( - json!({"access_control": false, "expiry_seconds": 10}).to_string(), - )) - .unwrap(), - ) - .await - .unwrap(); + let registered = call( + app.clone(), + Request::post("/channels") + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from( + json!({"access_control": false, "expiry_seconds": 10}).to_string(), + )) + .unwrap(), + ) + .await; let registered: serde_json::Value = serde_json::from_slice(&to_bytes(registered.into_body(), usize::MAX).await.unwrap()) .unwrap();