diff --git a/src/agent/mod.rs b/src/agent/mod.rs index 944ded7..6e85545 100644 --- a/src/agent/mod.rs +++ b/src/agent/mod.rs @@ -255,27 +255,29 @@ struct RoutedKey { } fn agent_request_span(request: &Request) -> tracing::Span { - let (operation, key, class) = match request.extensions().get::() { - Some(path) => (path.as_str().to_owned(), String::new(), "control"), - None => { - let routed = classify(request.method(), request.uri().path()); - (routed.kind, routed.id, routed.class.as_str()) - } - }; let request_id = request .headers() .get(&REQUEST_ID_HEADER) .and_then(|value| value.to_str().ok()) .unwrap_or("invalid"); - tracing::info_span!( + let span = tracing::info_span!( "http_request", component = "agent", %request_id, method = %request.method(), - %operation, - %key, - class, - ) + operation = tracing::field::Empty, + key = tracing::field::Empty, + class = tracing::field::Empty, + ); + // A control route is identified by its matched path; every other request is + // classified once by the forwarder, which records these fields itself rather + // than making the span repeat the work. + if let Some(path) = request.extensions().get::() { + span.record("operation", path.as_str()); + span.record("key", ""); + span.record("class", "control"); + } + span } /// Derives the routing object from a bare default-channel request path, mirroring the @@ -332,17 +334,19 @@ async fn forward(State(state): State>, request: Request) -> Resp .headers() .get(REQUEST_PURPOSE_HEADER) .is_some_and(|value| value.as_bytes() == REQUEST_PURPOSE_PREFETCH.as_bytes()); - let session = request - .headers() - .get(REQUEST_SESSION_HEADER) - .and_then(|value| value.to_str().ok()) - .map(str::to_owned); - let configured_concurrency = request - .headers() - .get(REQUEST_PREFETCH_CONCURRENCY_HEADER) - .and_then(|value| value.to_str().ok()) - .and_then(|value| value.parse::().ok()); + // The session and concurrency headers are read only to describe a prefetch, so + // foreground traffic never parses them. let observation = prefetch.then(|| { + let session = request + .headers() + .get(REQUEST_SESSION_HEADER) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let configured_concurrency = request + .headers() + .get(REQUEST_PREFETCH_CONCURRENCY_HEADER) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()); AgentPrefetchObservation::start(Arc::clone(&state), session, configured_concurrency) }); let started = Instant::now(); @@ -358,14 +362,19 @@ async fn forward(State(state): State>, request: Request) -> Resp } async fn forward_request(state: &Arc, request: Request, prefetch: bool) -> Response { + let routed = classify(request.method(), request.uri().path()); + let span = tracing::Span::current(); + span.record("operation", routed.kind.as_str()); + span.record("key", routed.id.as_str()); + span.record("class", routed.class.as_str()); if request.uri().path() == "/channels" || request.uri().path().starts_with("/channels/") { return StatusCode::NOT_IMPLEMENTED.into_response(); } - let routed = classify(request.method(), request.uri().path()); let Some(position) = ring::key_position(&routed.kind, &routed.id) else { return StatusCode::URI_TOO_LONG.into_response(); }; - let Some(owner) = state.ring.ring().owner(position).cloned() else { + let ring = state.ring.ring(); + let Some(owner) = ring.owner(position) else { if prefetch { state.metrics.prefetch_unavailable.inc(); } @@ -376,17 +385,17 @@ async fn forward_request(state: &Arc, request: Request, prefetch: bo .path_and_query() .map_or("/", |path_and_query| path_and_query.as_str()); let url = format!("http://{}{}", owner.address, path_and_query); - let method = request.method().clone(); - let reqwest_method = - reqwest::Method::from_bytes(method.as_str().as_bytes()).expect("method round-trips"); let (parts, body) = request.into_parts(); - let mut upstream = state.client.request(reqwest_method, url); + // Axum and reqwest share one `http` crate, so the inbound method is the outbound + // method — no textual round-trip. + let bodyless = matches!(parts.method, Method::GET | Method::HEAD); + let mut upstream = state.client.request(parts.method, url); for (name, value) in &parts.headers { if !skip_request_header(name.as_str()) { upstream = upstream.header(name.as_str(), value.as_bytes()); } } - if !matches!(method, Method::GET | Method::HEAD) { + if !bodyless { upstream = upstream.body(reqwest::Body::wrap_stream(body.into_data_stream())); } match upstream.send().await { @@ -408,7 +417,7 @@ async fn forward_request(state: &Arc, request: Request, prefetch: bo } Err(error) => { tracing::warn!(%error, owner = owner.id, "forwarded request failed"); - record_send_failure(state, &owner, &error); + record_send_failure(state, owner, &error); if prefetch { state.metrics.prefetch_unavailable.inc(); } diff --git a/src/cache/mod.rs b/src/cache/mod.rs index e75ba02..950a9ba 100644 --- a/src/cache/mod.rs +++ b/src/cache/mod.rs @@ -3,6 +3,8 @@ mod service; mod space; mod stripes; +#[cfg(test)] +mod recent_use_test; #[cfg(test)] mod space_test; @@ -17,4 +19,4 @@ pub use service::{ Admission, CacheError, CacheService, LocatedArtifact, Publication, PublicationOutcome, PublicationTarget, PublishRequest, }; -pub use space::{FreeSpace, Mode, SpaceLedger, SpacePolicy, StatvfsFreeSpace}; +pub use space::{FreeSpace, Mode, SpaceLedger, SpacePolicy, SpaceSnapshot, StatvfsFreeSpace}; diff --git a/src/cache/recent_use.rs b/src/cache/recent_use.rs index 1f69b55..89dc3f8 100644 --- a/src/cache/recent_use.rs +++ b/src/cache/recent_use.rs @@ -63,10 +63,15 @@ impl RecentUse { channel.hash(&mut base); artifact.digest().as_bytes().hash(&mut base); let seed = base.finish(); + // Kirsch-Mitzenmacher: two independent halves of the one hash generate every + // probe with the same false-positive behaviour as independent hashes, so a + // mark costs one hash instead of five. Forcing the step odd keeps it non-zero, + // so the probes never all collapse onto one slot. + let start = seed & 0xffff_ffff; + let step = (seed >> 32) | 1; + let bits = self.bits as u64; std::array::from_fn(|probe| { - let mut hasher = std::collections::hash_map::DefaultHasher::new(); - (seed, probe as u64).hash(&mut hasher); - (hasher.finish() % self.bits as u64) as usize + (start.wrapping_add(step.wrapping_mul(probe as u64)) % bits) as usize }) } } diff --git a/src/cache/recent_use_test.rs b/src/cache/recent_use_test.rs new file mode 100644 index 0000000..b72b530 --- /dev/null +++ b/src/cache/recent_use_test.rs @@ -0,0 +1,52 @@ +use super::recent_use::RecentUse; +use crate::{ + artifact::{ArtifactId, Digest}, + channel::ChannelId, +}; + +fn artifact(index: u32) -> ArtifactId { + let mut bytes = [0_u8; 32]; + bytes[..4].copy_from_slice(&index.to_le_bytes()); + ArtifactId::from_digest(Digest::from_bytes(bytes)) +} + +#[test] +fn a_mark_survives_exactly_one_rotation() { + let filter = RecentUse::new(4096); + let channel = ChannelId::DEFAULT; + filter.mark(channel, artifact(1)); + assert!(filter.seen(channel, artifact(1))); + filter.rotate(); + assert!(filter.seen(channel, artifact(1))); + filter.rotate(); + assert!(!filter.seen(channel, artifact(1))); +} + +#[test] +fn the_same_channel_and_artifact_probe_the_same_slots() { + let filter = RecentUse::new(4096); + filter.mark(ChannelId::DEFAULT, artifact(7)); + // The mark is keyed on both halves of the identity, so neither a different + // artifact in the same channel nor the same artifact elsewhere reads as seen. + assert!(!filter.seen(ChannelId::DEFAULT, artifact(8))); + assert!(!filter.seen(ChannelId::new(), artifact(7))); +} + +#[test] +fn derived_probes_keep_false_positives_rare() { + let filter = RecentUse::new(4096); + let channel = ChannelId::DEFAULT; + for index in 0..100 { + filter.mark(channel, artifact(index)); + } + // Four probes over a 2.4%-full filter should almost never all collide. A + // degenerate derivation (probes collapsing onto one slot) lands near 2.4%, + // an order of magnitude above this bound. + let false_positives = (1_000..2_000) + .filter(|&index| filter.seen(channel, artifact(index))) + .count(); + assert!( + false_positives < 5, + "{false_positives} false positives in 1000 probes" + ); +} diff --git a/src/cache/service.rs b/src/cache/service.rs index a1f400f..9e49a8c 100644 --- a/src/cache/service.rs +++ b/src/cache/service.rs @@ -222,8 +222,15 @@ impl CacheService { reference: String, artifact: ArtifactId, ) -> Result<(), CacheError> { - self.bind_reference_with_validators(channel, reference, artifact, None, None) - .await + self.bind_reference_with_validators( + channel, + reference, + artifact, + None, + None, + Durability::Durable, + ) + .await } pub(crate) async fn bind_reference_with_validators( @@ -233,8 +240,9 @@ impl CacheService { artifact: ArtifactId, etag: Option, last_modified: Option, + durability: Durability, ) -> Result<(), CacheError> { - let _gate = self.channel_fence(channel).await?; + let (_gate, _) = self.channel_fence(channel).await?; let _stripe = self.stripes.reference(channel, &reference).await; self.metadata .bind_reference( @@ -246,6 +254,7 @@ impl CacheService { etag, last_modified, }, + durability, ) .await?; Ok(()) @@ -264,7 +273,7 @@ impl CacheService { channel: ChannelId, reference: &str, ) -> Result<(), CacheError> { - let _gate = self.channel_fence(channel).await?; + let (_gate, _) = self.channel_fence(channel).await?; let _stripe = self.stripes.reference(channel, reference).await; self.metadata.delete_reference(channel, reference).await?; Ok(()) @@ -312,11 +321,7 @@ impl CacheService { /// channel rotates each pass so an early-exhausted budget does not starve the tail. pub async fn run_maintenance_once(&self, limit: usize) -> Result { self.space.refresh(); - self.metrics.record_space( - self.space.free_observed(), - self.space.reserved(), - self.space.committed_since(), - ); + self.metrics.record_space(self.space.snapshot()); let mode = self.space.mode(); let now = self.clock.now(); let mut channels = self.channels().await?; @@ -439,15 +444,8 @@ impl CacheService { // Fence the final mutation against channel deletion: acquire the shared channel // gate and recheck `active` so a body staged before deletion cannot be published // back into a channel whose ranges are being (or have been) removed. - let _gate = match self.channel_fence(channel).await { - Ok(gate) => gate, - Err(error) => { - staged.discard().await; - return Err(error); - } - }; - let expiry_seconds = match self.expiry_seconds(channel).await { - Ok(expiry_seconds) => expiry_seconds, + let (_gate, expiry_seconds) = match self.channel_fence(channel).await { + Ok(fence) => fence, Err(error) => { staged.discard().await; return Err(error); @@ -524,17 +522,18 @@ impl CacheService { } /// Acquires the shared channel lifecycle gate (read side) and rechecks that the - /// channel is still active, returning the guard to hold across a final mutation. + /// channel is still active, returning the guard to hold across a final mutation + /// along with the expiry carried by the very record the recheck just read. /// A deletion takes the write side, so holding this read guard blocks deletion from /// wiping the channel mid-commit, and the active recheck rejects a mutation into a /// channel whose deletion has already begun to tear it down. async fn channel_fence( &self, channel: ChannelId, - ) -> Result, CacheError> { + ) -> Result<(OwnedRwLockReadGuard<()>, u64), CacheError> { let guard = self.channel_gates.gate(channel).read_owned().await; match self.metadata.channel(channel).await? { - Some(record) if record.state == Lifecycle::Active => Ok(guard), + Some(record) if record.state == Lifecycle::Active => Ok((guard, record.expiry_seconds)), Some(_) => Err(CacheError::ChannelDeleting), None => Err(CacheError::MissingChannel), } diff --git a/src/cache/space.rs b/src/cache/space.rs index 68d2df1..8506be9 100644 --- a/src/cache/space.rs +++ b/src/cache/space.rs @@ -49,6 +49,14 @@ pub enum Mode { Reclaiming, } +/// One consistent reading of the ledger's exported counters. +#[derive(Clone, Copy, Debug)] +pub struct SpaceSnapshot { + pub free_observed: u64, + pub reserved: u64, + pub committed_since: u64, +} + /// The mutable ledger state guarded by a single mutex. Guarding the snapshot as one /// unit keeps a `refresh` from erasing bytes committed after its filesystem sample and /// keeps `degraded` observation handling consistent with reservation accounting. The @@ -174,6 +182,17 @@ impl SpaceLedger { .expect("space ledger poisoned") .committed_since } + + /// The three published counters read together, so the exported gauges describe one + /// consistent ledger state instead of three independently locked samples. + pub fn snapshot(&self) -> SpaceSnapshot { + let state = self.state.lock().expect("space ledger poisoned"); + SpaceSnapshot { + free_observed: state.free_observed, + reserved: state.reserved, + committed_since: state.committed_since, + } + } } impl Reserver for SpaceLedger { diff --git a/src/cacheprog/mod.rs b/src/cacheprog/mod.rs index 150db64..cfbe6ca 100644 --- a/src/cacheprog/mod.rs +++ b/src/cacheprog/mod.rs @@ -434,12 +434,12 @@ async fn get( // of downloading the same bytes again: Go's own build parallelism bounds // the waiters, and the signal fires however the download ends, so a // failed one simply falls through to the ordinary request below. - if !file_has_size(&path, entry.size).await - && let Some(mut pending) = session.download_in_progress(&entry.output) - { + let mut present = file_has_size(&path, entry.size).await; + if !present && let Some(mut pending) = session.download_in_progress(&entry.output) { let _ = pending.changed().await; + present = file_has_size(&path, entry.size).await; } - if file_has_size(&path, entry.size).await { + if present { 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 @@ -466,7 +466,7 @@ async fn get( .get(reqwest::header::ETAG) .and_then(|value| value.to_str().ok()) .map(str::to_owned); - let body = response.bytes().await?.to_vec(); + let body = response.bytes().await?; // The put-side check guarantees the stored body's hash IS the output ID. let output = hex::encode(Sha256::digest(&body)); if let Some(expected) = etag diff --git a/src/channel/identity.rs b/src/channel/identity.rs index e9ae1a8..b7519e2 100644 --- a/src/channel/identity.rs +++ b/src/channel/identity.rs @@ -17,11 +17,13 @@ impl ChannelId { } } - pub fn as_key(self) -> [u8; 26] { - self.to_string() - .as_bytes() - .try_into() - .expect("a ULID is always 26 bytes") + /// The channel's durable key prefix: the canonical ULID text, encoded straight + /// into a stack buffer. `Display` itself calls `array_to_str`, so the bytes are + /// identical to the string form without the intermediate allocation. + pub fn as_key(self) -> [u8; ulid::ULID_LEN] { + let mut key = [0_u8; ulid::ULID_LEN]; + self.0.array_to_str(&mut key); + key } } diff --git a/src/proxy/mod.rs b/src/proxy/mod.rs index 1bf178c..601d25c 100644 --- a/src/proxy/mod.rs +++ b/src/proxy/mod.rs @@ -509,6 +509,7 @@ impl ProxyService { record.artifact, record.etag, record.last_modified, + Durability::BestEffort, ) .await?; return self @@ -566,6 +567,7 @@ impl ProxyService { publication.artifact, etag, last_modified, + Durability::BestEffort, ) .await?; Ok(ProxyOutcome::Artifact(publication.artifact)) @@ -757,6 +759,7 @@ impl ProxyService { publication.artifact, etag, last_modified, + Durability::BestEffort, ) .await?; Ok(ProxyOutcome::CachedMetadata { diff --git a/src/reference.rs b/src/reference.rs index 2f2a476..94fddae 100644 --- a/src/reference.rs +++ b/src/reference.rs @@ -9,20 +9,29 @@ pub struct Reference(String); impl Reference { pub fn parse(value: impl Into) -> Result { let value = value.into(); - let valid = !value.is_empty() - && value.len() <= MAX_REFERENCE_LEN - && value.bytes().all(|byte| { - byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') - }); - if !valid { + if !Self::is_valid(&value) { return Err(ReferenceError); } Ok(Self(value)) } + /// The same rule `parse` applies, for callers that only need the verdict and + /// would otherwise allocate an owned reference just to drop it. + pub fn is_valid(value: &str) -> bool { + !value.is_empty() + && value.len() <= MAX_REFERENCE_LEN + && value.bytes().all(|byte| { + byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') + }) + } + pub fn as_str(&self) -> &str { &self.0 } + + pub fn into_string(self) -> String { + self.0 + } } impl fmt::Display for Reference { diff --git a/src/storage/local/artifact_files.rs b/src/storage/local/artifact_files.rs index d4d866e..6fc1bfe 100644 --- a/src/storage/local/artifact_files.rs +++ b/src/storage/local/artifact_files.rs @@ -256,7 +256,7 @@ impl StageWriter { /// Flushes buffered data through to the file and returns it. The flush matters /// even for identity bodies: `tokio::fs::File` completes writes on a background - /// task, and the caller stats this file for the stored length. + /// task, and the caller publishes this file as soon as it returns. async fn finish(self) -> std::io::Result { match self { Self::Identity(mut file) => { @@ -391,7 +391,13 @@ impl ArtifactFiles { writer.write_all(&chunk).await?; } let file = writer.finish().await?; - let stored_len = file.metadata().await?.len(); + // An identity body was written verbatim into a freshly created file, so its + // stored length is the byte count already counted above. Only a zstd frame has + // an encoded size that has to be read back from the filesystem. + let stored_len = match encoding { + StoredEncoding::Identity => len, + StoredEncoding::Zstd => file.metadata().await?.len(), + }; if stored_len > reservation.outstanding && !reservation.reserve(stored_len - reservation.outstanding) { diff --git a/src/storage/metadata/mod.rs b/src/storage/metadata/mod.rs index 9e14283..79daf19 100644 --- a/src/storage/metadata/mod.rs +++ b/src/storage/metadata/mod.rs @@ -13,10 +13,11 @@ pub struct ReferenceRecord { pub last_modified: Option, } -/// Whether a publication batch must be synchronously flushed before it is +/// Whether a metadata batch must be synchronously flushed before it is /// acknowledged. Raw artifact and Bazel CAS publications are `Durable`; build-cache -/// and proxy publications are `BestEffort` because the body is already complete and -/// any crash outcome is a hit, a self-healing miss, or an orphan file. +/// and proxy publications — and the reference bindings that accompany them, including +/// the rebind on an upstream 304 — are `BestEffort` because the body is already +/// complete and any crash outcome is a hit, a self-healing miss, or an orphan file. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum Durability { Durable, diff --git a/src/storage/metadata/rocksdb.rs b/src/storage/metadata/rocksdb.rs index 0b52bad..dc27594 100644 --- a/src/storage/metadata/rocksdb.rs +++ b/src/storage/metadata/rocksdb.rs @@ -149,7 +149,10 @@ impl RocksMetadata { ); } batch.delete_cf(&artifacts, &key); - write_sync(&database, batch) + // The same delete `evict` performs, and for the same reason it needs no + // fsync: losing the removal leaves a metadata row whose body is already + // gone, which the read path self-heals into a miss. + database.write(batch).map_err(store_error) }) .await } @@ -159,6 +162,7 @@ impl RocksMetadata { channel: ChannelId, reference: String, record: ReferenceRecord, + durability: Durability, ) -> Result<(), MetadataError> { let database = Arc::clone(&self.database); blocking(move || { @@ -173,7 +177,7 @@ impl RocksMetadata { reference_key(channel, &reference), encode_record(&record)?, ); - write_sync(&database, batch) + write(&database, batch, durability) }) .await } @@ -467,21 +471,31 @@ fn write_sync(database: &DB, batch: WriteBatch) -> Result<(), MetadataError> { } fn artifact_key(channel: ChannelId, artifact: ArtifactId) -> Vec { - let mut key = channel.as_key().to_vec(); - key.extend_from_slice(artifact.digest().as_bytes()); + let prefix = channel.as_key(); + let digest = artifact.digest(); + let digest = digest.as_bytes(); + let mut key = Vec::with_capacity(prefix.len() + digest.len()); + key.extend_from_slice(&prefix); + key.extend_from_slice(digest); key } fn reference_key(channel: ChannelId, reference: &str) -> Vec { - let mut key = channel.as_key().to_vec(); + let prefix = channel.as_key(); + let mut key = Vec::with_capacity(prefix.len() + reference.len()); + key.extend_from_slice(&prefix); key.extend_from_slice(reference.as_bytes()); key } fn eviction_key(channel: ChannelId, eligible_at: u64, artifact: ArtifactId) -> Vec { - let mut key = channel.as_key().to_vec(); + let prefix = channel.as_key(); + let digest = artifact.digest(); + let digest = digest.as_bytes(); + let mut key = Vec::with_capacity(prefix.len() + 8 + digest.len()); + key.extend_from_slice(&prefix); key.extend_from_slice(&eligible_at.to_be_bytes()); - key.extend_from_slice(artifact.digest().as_bytes()); + key.extend_from_slice(digest); key } diff --git a/src/telemetry.rs b/src/telemetry.rs index 98f60ac..7dd71bb 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -1,4 +1,5 @@ -use http::{HeaderName, HeaderValue, Request}; +use crate::cache::SpaceSnapshot; +use http::{HeaderName, HeaderValue, Request, StatusCode}; use prometheus::{ Encoder, Histogram, HistogramOpts, HistogramVec, IntCounter, IntCounterVec, IntGauge, Opts, Registry, TextEncoder, core::Collector, @@ -220,11 +221,11 @@ impl Metrics { &self, method: &str, route: &str, - status: u16, + status: StatusCode, duration: Duration, ) { self.http_request_duration - .with_label_values(&[method, route, &status.to_string()]) + .with_label_values(&[method, route, status.as_str()]) .observe(duration.as_secs_f64()); } @@ -300,11 +301,12 @@ impl Metrics { .inc(); } - pub fn record_space(&self, free_observed: u64, reserved: u64, committed_since: u64) { - self.free_observed_bytes.set(saturating_i64(free_observed)); - self.reserved_bytes.set(saturating_i64(reserved)); + pub fn record_space(&self, space: SpaceSnapshot) { + self.free_observed_bytes + .set(saturating_i64(space.free_observed)); + self.reserved_bytes.set(saturating_i64(space.reserved)); self.committed_since_bytes - .set(saturating_i64(committed_since)); + .set(saturating_i64(space.committed_since)); } pub fn encode(&self) -> prometheus::Result> { diff --git a/src/transport/http/mod.rs b/src/transport/http/mod.rs index 2bcddb0..e78cd6e 100644 --- a/src/transport/http/mod.rs +++ b/src/transport/http/mod.rs @@ -27,7 +27,7 @@ use axum::{ use base64::{Engine as _, engine::general_purpose::STANDARD}; use futures_util::StreamExt; use serde::{Deserialize, Serialize}; -use std::{io::SeekFrom, sync::Arc, time::Instant}; +use std::{borrow::Cow, io::SeekFrom, sync::Arc, time::Instant}; use tokio::io::{AsyncReadExt, AsyncSeekExt}; use tokio_util::io::ReaderStream; use tower::ServiceBuilder; @@ -141,7 +141,10 @@ fn server_request_span(request: &Request) -> tracing::Span { ) } -fn server_request_identity(request: &Request) -> (String, String) { +/// The span's operation and key for one request. Both borrow out of the request +/// wherever the path already holds the text; only the multi-segment proxy key is +/// assembled. +fn server_request_identity(request: &Request) -> (&str, Cow<'_, str>) { let path = request.uri().path(); let raw_segments: Vec<&str> = path.trim_start_matches('/').split('/').collect(); let segments = match raw_segments.as_slice() { @@ -149,21 +152,20 @@ fn server_request_identity(request: &Request) -> (String, String) { rest => rest, }; match segments { - ["artifacts", _algorithm, digest] => ("artifact".to_owned(), (*digest).to_owned()), - ["build-cache", "bazel", "cas", hash] => ("artifact".to_owned(), (*hash).to_owned()), - ["build-cache", "bazel", "ac", hash] => ("bazel-action".to_owned(), (*hash).to_owned()), - ["build-cache", "http", key] => ("http-cache".to_owned(), (*key).to_owned()), - ["references", reference] => ("reference".to_owned(), (*reference).to_owned()), + ["artifacts", _algorithm, digest] => ("artifact", Cow::Borrowed(*digest)), + ["build-cache", "bazel", "cas", hash] => ("artifact", Cow::Borrowed(*hash)), + ["build-cache", "bazel", "ac", hash] => ("bazel-action", Cow::Borrowed(*hash)), + ["build-cache", "http", key] => ("http-cache", Cow::Borrowed(*key)), + ["references", reference] => ("reference", Cow::Borrowed(*reference)), ["proxy", protocol, rest @ ..] if !rest.is_empty() => { - ((*protocol).to_owned(), rest.join("/")) + (*protocol, Cow::Owned(rest.join("/"))) } _ => ( request .extensions() .get::() - .map_or("unmatched", MatchedPath::as_str) - .to_owned(), - String::new(), + .map_or("unmatched", MatchedPath::as_str), + Cow::Borrowed(""), ), } } @@ -241,7 +243,9 @@ struct HashPath { pub(super) struct ChannelContext { pub channel: ChannelId, - pub route_prefix: String, + /// Set only when the request came in under an explicit `/channels/{id}` prefix. + /// The bare routes resolve to the default channel and carry no prefix. + scope: Option, pub access_control: bool, } @@ -254,17 +258,25 @@ impl ChannelContext { let Some(channel) = channel else { return Ok(Self { channel: ChannelId::DEFAULT, - route_prefix: String::new(), + scope: None, access_control: false, }); }; let record = authorize_channel(state, channel, headers).await?; Ok(Self { channel: record.id, - route_prefix: format!("/channels/{}", record.id), + scope: Some(record.id), access_control: record.access.is_protected(), }) } + + /// The path prefix rewritten package URLs must carry to route back to this + /// channel. Only the package-proxy handlers rewrite URLs, so it is built on + /// demand rather than on every request. + pub(super) fn route_prefix(&self) -> String { + self.scope + .map_or_else(String::new, |channel| format!("/channels/{channel}")) + } } async fn live() -> StatusCode { @@ -308,21 +320,23 @@ async fn count_request( .headers() .get(REQUEST_PURPOSE_HEADER) .is_some_and(|value| value.as_bytes() == REQUEST_PURPOSE_PREFETCH.as_bytes()); - let session = request - .headers() - .get(REQUEST_SESSION_HEADER) - .and_then(|value| value.to_str().ok()) - .map(str::to_owned); - let configured_concurrency = request - .headers() - .get(REQUEST_PREFETCH_CONCURRENCY_HEADER) - .and_then(|value| value.to_str().ok()) - .and_then(|value| value.parse::().ok()); + // The session and concurrency headers are read only to describe a prefetch, so + // foreground traffic never parses them. let observation = prefetch.then(|| { + let session = request + .headers() + .get(REQUEST_SESSION_HEADER) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let configured_concurrency = request + .headers() + .get(REQUEST_PREFETCH_CONCURRENCY_HEADER) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()); PrefetchObservation::start( Arc::clone(&metrics), route.clone(), - session.clone(), + session, configured_concurrency, ) }); @@ -332,7 +346,7 @@ async fn count_request( metrics.observe_http_request( method.as_str(), &route, - response.status().as_u16(), + response.status(), response_headers_duration, ); match observation { @@ -675,7 +689,7 @@ async fn put_reference( }; match state .cache - .bind_reference(context.channel, reference.to_string(), artifact) + .bind_reference(context.channel, reference.into_string(), artifact) .await { Ok(()) => StatusCode::NO_CONTENT.into_response(), @@ -1113,7 +1127,7 @@ async fn put_http_cache( Ok(context) => context, Err(response) => return response, }; - if Reference::parse(&path.key).is_err() { + if !Reference::is_valid(&path.key) { return StatusCode::BAD_REQUEST.into_response(); } let Some(_permit) = acquire_foreground(&state) else { @@ -1307,9 +1321,9 @@ async fn get_bazel_ac( Ok(context) => context, Err(response) => return response, }; - if ArtifactId::parse("sha256", &path.hash).is_err() { - return StatusCode::BAD_REQUEST.into_response(); - } + // No hex validation on the read: an unparseable hash cannot name a stored action + // result either, so it takes the same reference lookup and misses, exactly as the + // HTTP build-cache read does. serve_reference_artifact( state, context.channel, diff --git a/src/transport/http/packages.rs b/src/transport/http/packages.rs index 1025236..218cb03 100644 --- a/src/transport/http/packages.rs +++ b/src/transport/http/packages.rs @@ -69,7 +69,7 @@ async fn fetch_python_simple( let accept = headers .get(header::ACCEPT) .and_then(|value| value.to_str().ok()); - let Some(transform) = Transform::python_simple(context.route_prefix, path, accept) else { + let Some(transform) = Transform::python_simple(context.route_prefix(), path, accept) else { return StatusCode::NOT_ACCEPTABLE.into_response(); }; fetch(state, context.channel, Protocol::Python, path, transform).await @@ -118,7 +118,7 @@ pub(super) async fn npm( &state, context.channel, &path.path, - context.route_prefix, + context.route_prefix(), accept, ) .await @@ -153,7 +153,7 @@ pub(super) async fn cargo_config( Err(response) => return response, }; Json(serde_json::json!({ - "dl": format!("{}/proxy/cargo/crates", context.route_prefix), + "dl": format!("{}/proxy/cargo/crates", context.route_prefix()), "api": null, "auth-required": context.access_control, })) @@ -246,22 +246,17 @@ async fn outcome_response( ) .await } + // The rewritten document is already materialized in memory, so there is no + // cache work left to admit and no reason to hold a foreground slot for the + // client's download. ProxyOutcome::CachedMetadata { body, content_type } => { - let Some(permit) = acquire_foreground(state) else { - return busy_response(); - }; let mut builder = Response::builder() .status(StatusCode::OK) .header(header::CONTENT_LENGTH, body.len()); if let Some(content_type) = content_type { builder = builder.header(header::CONTENT_TYPE, content_type); } - builder - .body(body_with_permit( - futures_util::stream::iter([Ok::<_, std::io::Error>(body)]), - permit, - )) - .unwrap() + builder.body(Body::from(body)).unwrap() } ProxyOutcome::Upstream { status,