From 492cb0687c5cfcd1c4aeb182869f565019c7b8bf Mon Sep 17 00:00:00 2001 From: Rob Lyon Date: Tue, 21 Jul 2026 09:27:53 -0700 Subject: [PATCH] refactor: extract the channel context, share Accept parsing, and fold duplicated blocks ChannelContext is now a FromRequestParts extractor, so the four-line resolve prologue leaves 18 handlers and the channel field leaves all seven per-route path structs. put_reference takes its context as a Result and resolves the rejection in the body: extractors run before the body, so rejecting early would answer a syntactically broken request with a channel error instead of a 400. negotiate_python and negotiate_npm shared roughly 25 byte-identical lines, and the two tie-breaks turn out to be the same function once the variables are renamed. Both now share media_range, offer, acceptable, and outranks, leaving each negotiator its media-type table and one call. The per-protocol default for a wildcard covering both representations -- Python serves JSON, npm abbreviated -- now lives visibly in the argument order. Also folds the OutOfSpace arm into cache_write_failure so all three PUT paths share it, replaces unix_time with the injected Clock, extracts fresh_outcome and revalidated from fetch_url, collapses the fetch_npm chain and the three select! blocks, shares publish_temporary, upstream_response, content_type, and content_length, merges the NotFound and Deleting response arms, and has SpaceLedger::new call refresh instead of repeating its match. --- src/cache/space.rs | 16 +- src/cacheprog/mod.rs | 44 +++--- src/cacheprog/session.rs | 7 +- src/channel/service.rs | 51 +++--- src/lib.rs | 1 + src/proxy/mod.rs | 274 ++++++++++++++++----------------- src/transport/http/mod.rs | 212 +++++++++---------------- src/transport/http/packages.rs | 121 +++++---------- src/transport/mod.rs | 14 ++ 9 files changed, 311 insertions(+), 429 deletions(-) diff --git a/src/cache/space.rs b/src/cache/space.rs index dac9f82..d365704 100644 --- a/src/cache/space.rs +++ b/src/cache/space.rs @@ -86,22 +86,22 @@ pub struct SpaceLedger { } impl SpaceLedger { + /// Starts degraded with no observation, then takes the first one through `refresh` + /// so a ledger whose very first `statvfs` fails stays degraded and admits nothing. pub fn new(source: Arc, policy: SpacePolicy) -> Self { - let (free_observed, degraded) = match source.free_bytes() { - Some(free) => (free, false), - None => (0, true), - }; - Self { + let ledger = Self { source, policy, state: Mutex::new(State { - free_observed, + free_observed: 0, reserved: 0, committed_since: 0, - degraded, + degraded: true, reclaiming: false, }), - } + }; + ledger.refresh(); + ledger } /// Re-observes filesystem free space and resets the committed-since counter, since diff --git a/src/cacheprog/mod.rs b/src/cacheprog/mod.rs index d3beec3..4d0cab3 100644 --- a/src/cacheprog/mod.rs +++ b/src/cacheprog/mod.rs @@ -222,29 +222,16 @@ where break; } - let event = if close_id.is_some() { - tokio::select! { - biased; - () = &mut shutdown => Event::Shutdown, - response = in_flight.next() => Event::Response( - response.expect("in-flight request was checked") - ), - } - } else if in_flight.is_empty() { - tokio::select! { - biased; - () = &mut shutdown => Event::Shutdown, - message = reader.next_message() => Event::Input(message?), - } - } else { - tokio::select! { - biased; - () = &mut shutdown => Event::Shutdown, - message = reader.next_message() => Event::Input(message?), - response = in_flight.next() => { - Event::Response(response.expect("in-flight request was checked")) - }, - } + // Shutdown first, then input, then completions: a `close` stops the helper + // reading further requests, and an empty `FuturesUnordered` would otherwise + // report completion immediately and spin. + let event = tokio::select! { + biased; + () = &mut shutdown => Event::Shutdown, + message = reader.next_message(), if close_id.is_none() => Event::Input(message?), + response = in_flight.next(), if !in_flight.is_empty() => { + Event::Response(response.expect("in-flight request was checked")) + }, }; match event { @@ -556,8 +543,15 @@ async fn write_atomic(path: &Path, body: &[u8]) -> anyhow::Result<()> { return Err(error.into()); } drop(file); - if let Err(error) = tokio::fs::rename(&temporary, path).await { - let _ = tokio::fs::remove_file(&temporary).await; + publish_temporary(&temporary, path).await +} + +/// Moves a fully written temporary file into its final place. Losing the rename to a +/// concurrent writer of the same content-addressed object is not a failure: the +/// temporary is dropped and the file that won stands. +async fn publish_temporary(temporary: &Path, path: &Path) -> anyhow::Result<()> { + if let Err(error) = tokio::fs::rename(temporary, path).await { + let _ = tokio::fs::remove_file(temporary).await; if !tokio::fs::try_exists(path).await? { return Err(error.into()); } diff --git a/src/cacheprog/session.rs b/src/cacheprog/session.rs index 406d41f..2338df9 100644 --- a/src/cacheprog/session.rs +++ b/src/cacheprog/session.rs @@ -414,12 +414,7 @@ async fn download_object( let _ = tokio::fs::remove_file(&temporary).await; return Err(error); } - if let Err(error) = tokio::fs::rename(&temporary, &path).await { - let _ = tokio::fs::remove_file(&temporary).await; - if !tokio::fs::try_exists(&path).await? { - return Err(error.into()); - } - } + super::publish_temporary(&temporary, &path).await?; Ok(Some(entry.size)) } diff --git a/src/channel/service.rs b/src/channel/service.rs index 1f80d71..7eeadcc 100644 --- a/src/channel/service.rs +++ b/src/channel/service.rs @@ -1,13 +1,13 @@ use super::{Access, ChannelId, ChannelRecord, ChannelToken, Lifecycle}; -use crate::storage::{ - local::{ArtifactFiles, LocalError}, - metadata::{MetadataError, RocksMetadata}, +use crate::{ + clock::Clock, + storage::{ + local::{ArtifactFiles, LocalError}, + metadata::{MetadataError, RocksMetadata}, + }, }; use dashmap::DashMap; -use std::{ - sync::Arc, - time::{SystemTime, UNIX_EPOCH}, -}; +use std::sync::Arc; use tokio::sync::{OwnedRwLockReadGuard, RwLock}; /// The per-channel lifecycle gates, shared between the channel service (which takes the @@ -42,6 +42,7 @@ pub struct ChannelService { store: Arc, files: Arc, gates: Arc, + clock: Arc, } pub struct IssuedChannel { @@ -59,11 +60,13 @@ impl ChannelService { store: Arc, files: Arc, gates: Arc, + clock: Arc, ) -> Self { Self { store, files, gates, + clock, } } @@ -76,7 +79,7 @@ impl ChannelService { access: Access::Open, expiry_seconds, state: Lifecycle::Active, - created_at: unix_time(), + created_at: self.clock.now(), }; match self.store.create_channel(record.clone()).await { Ok(()) => record, @@ -109,7 +112,7 @@ impl ChannelService { .map_or(Access::Open, |token| Access::Token(token.digest())), expiry_seconds, state: Lifecycle::Active, - created_at: unix_time(), + created_at: self.clock.now(), }; self.store.create_channel(record.clone()).await?; self.gates.gate(record.id); @@ -121,23 +124,16 @@ impl ChannelService { id: ChannelId, credential: Option<&str>, ) -> Result { - let Some(record) = self.store.channel(id).await? else { - return Err(ChannelError::NotFound); - }; - authorize_record(&record, credential)?; - if record.state != Lifecycle::Active { - return Err(ChannelError::Deleting); - } + // The pre-gate check is load-bearing, not a fast path: `ChannelGates::gate` + // inserts an entry for whatever id it is handed, so taking the gate before the + // channel is known to exist would leave a permanent `DashMap` entry behind for + // every unknown id a client asks about. + self.authorize(id, credential).await?; let guard = self.gates.gate(id).read_owned().await; - let Some(current) = self.store.channel(id).await? else { - return Err(ChannelError::NotFound); - }; - authorize_record(¤t, credential)?; - if current.state != Lifecycle::Active { - return Err(ChannelError::Deleting); - } + // Recheck under the gate: a deletion may have landed between the two. + let record = self.authorize(id, credential).await?; Ok(ChannelLease { - record: current, + record, _guard: guard, }) } @@ -218,13 +214,6 @@ fn authorize_record(record: &ChannelRecord, credential: Option<&str>) -> Result< Ok(()) } -fn unix_time() -> u64 { - SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_secs() -} - #[derive(Debug, thiserror::Error)] pub enum ChannelError { #[error("channel does not exist")] diff --git a/src/lib.rs b/src/lib.rs index 97dcd57..e5c2f26 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -65,6 +65,7 @@ impl Flywheel { Arc::clone(&metadata), Arc::clone(&files), Arc::clone(&channel_gates), + Arc::clone(&clock), )); channels .ensure_default(config.default_expiry_seconds) diff --git a/src/proxy/mod.rs b/src/proxy/mod.rs index 601d25c..d756f2c 100644 --- a/src/proxy/mod.rs +++ b/src/proxy/mod.rs @@ -17,7 +17,7 @@ use bytes::Bytes; use dashmap::DashMap; use futures_util::{Stream, stream}; use html_escape::decode_html_entities; -use http::{HeaderMap, HeaderValue, StatusCode, header}; +use http::{HeaderValue, StatusCode, header}; use lol_html::{RewriteStrSettings, element, rewrite_str}; use serde_json::Value; use sha2::{Digest as _, Sha256}; @@ -123,6 +123,54 @@ enum PythonRepresentation { Json, } +/// Splits one `Accept` media range into its lowercased media type and its `q` value, +/// defaulting to 1.0 and clamping to the range the grammar allows. +fn media_range(range: &str) -> (String, f32) { + let mut fields = range.split(';'); + let media_type = fields + .next() + .unwrap_or_default() + .trim() + .to_ascii_lowercase(); + let quality = fields + .filter_map(|field| field.trim().split_once('=')) + .find_map(|(name, value)| { + name.trim() + .eq_ignore_ascii_case("q") + .then(|| value.trim().parse::().ok()) + .flatten() + }) + .unwrap_or(1.0) + .clamp(0.0, 1.0); + (media_type, quality) +} + +/// Records one media range against a candidate representation as `(specificity, +/// quality)`, keeping the most specific range that named it — 2 for an exact media +/// type, 1 for a subtype wildcard, 0 for `*/*` — and the best quality among equally +/// specific ones. +fn offer(candidate: &mut Option<(u8, f32)>, specificity: u8, quality: f32) { + if candidate.is_none_or(|(current_specificity, current_quality)| { + specificity > current_specificity + || (specificity == current_specificity && quality > current_quality) + }) { + *candidate = Some((specificity, quality)); + } +} + +/// Whether `preferred` beats `other`. Equal qualities go to `preferred` whenever it was +/// named at least as specifically, so a wildcard range covering both — `application/*` — +/// resolves to whichever representation the caller passes first. That choice is the +/// protocol default and differs per protocol: Python serves JSON, npm serves abbreviated. +fn outranks(preferred: (u8, f32), other: (u8, f32)) -> bool { + preferred.1 > other.1 || (preferred.1 == other.1 && preferred.0 > 0 && preferred.0 >= other.0) +} + +/// Drops a candidate the client explicitly refused with `q=0`. +fn acceptable(candidate: Option<(u8, f32)>) -> Option<(u8, f32)> { + candidate.filter(|(_, quality)| *quality > 0.0) +} + fn negotiate_python(accept: Option<&str>) -> Option { let Some(accept) = accept.filter(|value| !value.trim().is_empty()) else { return Some(PythonRepresentation::Html); @@ -130,65 +178,33 @@ fn negotiate_python(accept: Option<&str>) -> Option { let mut html = None; let mut json = None; for range in accept.split(',') { - let mut fields = range.split(';'); - let media_type = fields - .next() - .unwrap_or_default() - .trim() - .to_ascii_lowercase(); - let quality = fields - .filter_map(|field| field.trim().split_once('=')) - .find_map(|(name, value)| { - name.trim() - .eq_ignore_ascii_case("q") - .then(|| value.trim().parse::().ok()) - .flatten() - }) - .unwrap_or(1.0) - .clamp(0.0, 1.0); - let update = |candidate: &mut Option<(u8, f32)>, specificity| { - if candidate.is_none_or(|current| { - specificity > current.0 || (specificity == current.0 && quality > current.1) - }) { - *candidate = Some((specificity, quality)); - } - }; + let (media_type, quality) = media_range(range); match media_type.as_str() { "application/vnd.pypi.simple.v1+json" | "application/vnd.pypi.simple.latest+json" => { - update(&mut json, 2) + offer(&mut json, 2, quality); } "application/vnd.pypi.simple.v1+html" | "application/vnd.pypi.simple.latest+html" - | "text/html" => update(&mut html, 2), + | "text/html" => offer(&mut html, 2, quality), "application/*" => { - update(&mut html, 1); - update(&mut json, 1); + offer(&mut html, 1, quality); + offer(&mut json, 1, quality); } - "text/*" => update(&mut html, 1), + "text/*" => offer(&mut html, 1, quality), "*/*" => { - update(&mut html, 0); - update(&mut json, 0); + offer(&mut html, 0, quality); + offer(&mut json, 0, quality); } _ => {} } } - let html = html.filter(|(_, quality)| *quality > 0.0); - let json = json.filter(|(_, quality)| *quality > 0.0); - match (html, json) { + match (acceptable(html), acceptable(json)) { (None, None) => None, (Some(_), None) => Some(PythonRepresentation::Html), (None, Some(_)) => Some(PythonRepresentation::Json), - (Some((html_specificity, html_quality)), Some((json_specificity, json_quality))) => { - if json_quality > html_quality - || (json_quality == html_quality - && json_specificity > 0 - && json_specificity >= html_specificity) - { - Some(PythonRepresentation::Json) - } else { - Some(PythonRepresentation::Html) - } - } + // JSON is Python's default, so it takes the ties `application/*` produces. + (Some(html), Some(json)) if outranks(json, html) => Some(PythonRepresentation::Json), + (Some(_), Some(_)) => Some(PythonRepresentation::Html), } } @@ -199,53 +215,27 @@ fn negotiate_npm(accept: Option<&str>) -> Option { let mut abbreviated = None; let mut full = None; for range in accept.split(',') { - let mut fields = range.split(';'); - let media_type = fields - .next() - .unwrap_or_default() - .trim() - .to_ascii_lowercase(); - let quality = fields - .filter_map(|field| field.trim().split_once('=')) - .find_map(|(name, value)| { - name.trim() - .eq_ignore_ascii_case("q") - .then(|| value.trim().parse::().ok()) - .flatten() - }) - .unwrap_or(1.0) - .clamp(0.0, 1.0); - let update = |candidate: &mut Option<(u8, f32)>, specificity| { - if candidate.is_none_or(|current| { - specificity > current.0 || (specificity == current.0 && quality > current.1) - }) { - *candidate = Some((specificity, quality)); - } - }; + let (media_type, quality) = media_range(range); match media_type.as_str() { - "application/vnd.npm.install-v1+json" => update(&mut abbreviated, 2), - "application/json" => update(&mut full, 2), + "application/vnd.npm.install-v1+json" => offer(&mut abbreviated, 2, quality), + "application/json" => offer(&mut full, 2, quality), "application/*" => { - update(&mut abbreviated, 1); - update(&mut full, 1); + offer(&mut abbreviated, 1, quality); + offer(&mut full, 1, quality); } "*/*" => { - update(&mut abbreviated, 0); - update(&mut full, 0); + offer(&mut abbreviated, 0, quality); + offer(&mut full, 0, quality); } _ => {} } } - let abbreviated = abbreviated.filter(|(_, quality)| *quality > 0.0); - let full = full.filter(|(_, quality)| *quality > 0.0); - match (abbreviated, full) { + match (acceptable(abbreviated), acceptable(full)) { (None, None) => None, (Some(_), None) => Some(true), (None, Some(_)) => Some(false), - (Some((specificity, quality)), Some((full_specificity, full_quality))) => Some( - quality > full_quality - || (quality == full_quality && specificity > 0 && specificity >= full_specificity), - ), + // Abbreviated is npm's default, so it takes the ties `application/*` produces. + (Some(abbreviated), Some(full)) => Some(outranks(abbreviated, full)), } } @@ -388,7 +378,7 @@ impl ProxyService { } } - pub fn url(&self, protocol: Protocol, path: &str) -> Result { + fn url(&self, protocol: Protocol, path: &str) -> Result { let base = match protocol { Protocol::Go => &self.bases.go, Protocol::Python => &self.bases.python, @@ -414,21 +404,13 @@ impl ProxyService { self.fetch_url(channel, protocol, url, transform).await } + /// Fetches a base64url-encoded upstream URL, optionally with `suffix` appended to + /// its path — how PyPI's `.metadata` and `.asc` sidecars are addressed. pub async fn fetch_encoded_url( &self, channel: ChannelId, protocol: Protocol, encoded: &str, - ) -> Result { - self.fetch_encoded_url_with_suffix(channel, protocol, encoded, "") - .await - } - - pub async fn fetch_encoded_url_with_suffix( - &self, - channel: ChannelId, - protocol: Protocol, - encoded: &str, suffix: &str, ) -> Result { let bytes = URL_SAFE_NO_PAD @@ -459,11 +441,10 @@ impl ProxyService { transform: Transform, ) -> Result { let reference = proxy_reference(protocol, &url, transform.cache_variant()); - if let Some(record) = self.cache.resolve_reference(channel, &reference).await? - && self.clock.now().saturating_sub(record.updated_at) < self.ttl - && let Some(outcome) = self - .cached_outcome(channel, record.artifact, &url, &transform) - .await? + let cached = self.cache.resolve_reference(channel, &reference).await?; + if let Some(outcome) = self + .fresh_outcome(channel, cached.as_ref(), &url, &transform) + .await? { return Ok(outcome); } @@ -480,12 +461,12 @@ impl ProxyService { key, }; let _coalesced = lock.lock().await; + // Re-read behind the coalescing lock: whichever request held it may already have + // refreshed this reference. let current = self.cache.resolve_reference(channel, &reference).await?; - if let Some(record) = ¤t - && self.clock.now().saturating_sub(record.updated_at) < self.ttl - && let Some(outcome) = self - .cached_outcome(channel, record.artifact, &url, &transform) - .await? + if let Some(outcome) = self + .fresh_outcome(channel, current.as_ref(), &url, &transform) + .await? { return Ok(outcome); } @@ -499,27 +480,9 @@ impl ProxyService { .send_validated(protocol, url.clone(), ¤t, transform.upstream_accept()) .await?; if response.status() == StatusCode::NOT_MODIFIED { - let record = current.ok_or_else(|| { - ProxyError::Unavailable("upstream returned 304 without a cached value".to_owned()) - })?; - self.cache - .bind_reference_with_validators( - channel, - reference, - record.artifact, - record.etag, - record.last_modified, - Durability::BestEffort, - ) - .await?; return self - .cached_outcome(channel, record.artifact, &url, &transform) - .await? - .ok_or_else(|| { - ProxyError::Unavailable( - "upstream returned 304 for a missing cached body".to_owned(), - ) - }); + .revalidated(channel, reference, current, &url, &transform) + .await; } let status = response.status(); @@ -544,7 +507,7 @@ impl ProxyService { // near-full steady state), the body is handed back untouched and streamed // straight to the client — no buffering and no second upstream fetch. Transform::None => { - let upstream_length = header_length(response.headers()); + let upstream_length = crate::transport::content_length(response.headers()); let bypass_content_type = content_type.clone(); match self .cache @@ -606,6 +569,57 @@ impl ProxyService { } } + /// The cached outcome when a stored binding is still inside the revalidation TTL and + /// its body is still on disk. `None` means the upstream has to be consulted. + async fn fresh_outcome( + &self, + channel: ChannelId, + record: Option<&ReferenceRecord>, + upstream: &Url, + transform: &Transform, + ) -> Result, ProxyError> { + let Some(record) = + record.filter(|record| self.clock.now().saturating_sub(record.updated_at) < self.ttl) + else { + return Ok(None); + }; + self.cached_outcome(channel, record.artifact, upstream, transform) + .await + } + + /// Completes a revalidation: rebinds the reference so its TTL restarts, then serves + /// the cached body. A 304 with nothing cached, or whose body has since been + /// reclaimed, leaves nothing to answer with. + async fn revalidated( + &self, + channel: ChannelId, + reference: String, + current: Option, + upstream: &Url, + transform: &Transform, + ) -> Result { + let record = current.ok_or_else(|| { + ProxyError::Unavailable("upstream returned 304 without a cached value".to_owned()) + })?; + self.cache + .bind_reference_with_validators( + channel, + reference, + record.artifact, + record.etag, + record.last_modified, + Durability::BestEffort, + ) + .await?; + self.cached_outcome(channel, record.artifact, upstream, transform) + .await? + .ok_or_else(|| { + ProxyError::Unavailable( + "upstream returned 304 for a missing cached body".to_owned(), + ) + }) + } + async fn cached_outcome( &self, channel: ChannelId, @@ -878,20 +892,6 @@ fn is_redirect(status: StatusCode) -> bool { ) } -/// Publishes proxied bytes best-effort, returning the cached artifact id — or `None` -/// when the reservation failed under disk pressure so the caller can bypass the cache -/// and serve the upstream response directly instead of surfacing an error. -/// Parses an upstream `Content-Length` so the cache can size its reservation up front -/// and make an immediate admit/bypass decision before the body is streamed. -fn header_length(headers: &HeaderMap) -> Option { - headers - .get(header::CONTENT_LENGTH)? - .to_str() - .ok()? - .parse() - .ok() -} - fn rewrite_npm( upstream: &Url, route_prefix: &str, diff --git a/src/transport/http/mod.rs b/src/transport/http/mod.rs index 47e8a57..414a03d 100644 --- a/src/transport/http/mod.rs +++ b/src/transport/http/mod.rs @@ -15,13 +15,14 @@ use crate::{ telemetry::{ MakeRequestUlid, Metrics, PrefetchObservation, REQUEST_ID_HEADER, observe_prefetch_body, }, + transport::content_length, }; use async_compression::tokio::bufread::ZstdDecoder; use axum::{ Json, Router, body::Body, - extract::{MatchedPath, Path, Request, State, rejection::JsonRejection}, - http::{HeaderMap, HeaderValue, Method, StatusCode, header}, + extract::{FromRequestParts, MatchedPath, Path, Request, State, rejection::JsonRejection}, + http::{HeaderMap, HeaderValue, Method, StatusCode, header, request::Parts}, middleware::{self, Next}, response::{IntoResponse, Response}, routing::{get, post}, @@ -213,6 +214,8 @@ fn data_router() -> Router> { ) } +/// The `{channel}` capture every data route may carry. Present on the `/channels/{id}` +/// prefixed forms and absent on the bare ones, which resolve to the default channel. #[derive(Deserialize)] struct ChannelPath { channel: Option, @@ -220,26 +223,22 @@ struct ChannelPath { #[derive(Deserialize)] struct ArtifactPath { - channel: Option, algorithm: String, digest: String, } #[derive(Deserialize)] struct ReferencePath { - channel: Option, reference: String, } #[derive(Deserialize)] struct KeyPath { - channel: Option, key: String, } #[derive(Deserialize)] struct HashPath { - channel: Option, hash: String, } @@ -251,12 +250,22 @@ pub(super) struct ChannelContext { pub access_control: bool, } -impl ChannelContext { - async fn resolve( +/// Resolves and authorizes the channel a data request addresses, so every data handler +/// gets it by taking a `ChannelContext` argument instead of repeating the lookup. +/// +/// The `{channel}` capture is read through `Path`, which borrows the captured +/// parameters out of the request extensions rather than taking them, so a handler can +/// still extract its own `Path` for the rest of the route. +impl FromRequestParts> for ChannelContext { + type Rejection = Response; + + async fn from_request_parts( + parts: &mut Parts, state: &Arc, - channel: Option<&str>, - headers: &HeaderMap, - ) -> Result { + ) -> Result { + let Path(ChannelPath { channel }) = Path::::from_request_parts(parts, state) + .await + .map_err(|_| invalid_channel())?; let Some(channel) = channel else { return Ok(Self { channel: ChannelId::DEFAULT, @@ -264,14 +273,16 @@ impl ChannelContext { access_control: false, }); }; - let record = authorize_channel(state, channel, headers).await?; + let record = authorize_channel(state, &channel, &parts.headers).await?; Ok(Self { channel: record.id, scope: Some(record.id), access_control: record.access.is_protected(), }) } +} +impl ChannelContext { /// 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. @@ -359,31 +370,24 @@ async fn count_request( async fn put_artifact( State(state): State>, + context: ChannelContext, Path(path): Path, headers: HeaderMap, body: Body, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; let Ok(artifact) = ArtifactId::parse(&path.algorithm, &path.digest) else { return StatusCode::BAD_REQUEST.into_response(); }; let Some(_permit) = acquire_foreground(&state) else { return busy_response(); }; - let content_type = headers - .get(header::CONTENT_TYPE) - .and_then(|value| value.to_str().ok()) - .map(str::to_owned); let content_length = content_length(&headers); match state .cache .publish(PublishRequest { channel: context.channel, target: PublicationTarget::ById(artifact), - content_type, + content_type: content_type(&headers), stream: body.into_data_stream(), content_length, durability: Durability::Durable, @@ -395,36 +399,17 @@ async fn put_artifact( StatusCode::CREATED.into_response() } Ok(_) => StatusCode::NO_CONTENT.into_response(), - Err(CacheError::Local(crate::storage::local::LocalError::OutOfSpace)) => { - state.metrics.raw_pressure_error(); - insufficient_storage() - } - Err(CacheError::Local(crate::storage::local::LocalError::DigestMismatch)) => { - StatusCode::CONFLICT.into_response() - } - Err(CacheError::Local(crate::storage::local::LocalError::TooLarge)) => { - StatusCode::PAYLOAD_TOO_LARGE.into_response() - } - Err(CacheError::ChannelDeleting | CacheError::MissingChannel) => { - StatusCode::NOT_FOUND.into_response() - } - Err(error) => { - tracing::error!(error = %error, "artifact publication failed"); - StatusCode::INTERNAL_SERVER_ERROR.into_response() - } + Err(error) => cache_write_failure(&state, error), } } async fn get_artifact( State(state): State>, + context: ChannelContext, Path(path): Path, method: Method, headers: HeaderMap, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; let Ok(artifact) = ArtifactId::parse(&path.algorithm, &path.digest) else { return StatusCode::BAD_REQUEST.into_response(); }; @@ -562,13 +547,17 @@ struct ArtifactBinding { digest: String, } +/// The one data handler that takes its `ChannelContext` as a `Result`: extractors run +/// before the body, so rejecting here would answer a syntactically broken request with +/// a channel error. Holding the rejection until the body has parsed keeps a malformed +/// payload a 400 whatever the channel's access control says. async fn put_reference( State(state): State>, + context: Result, Path(path): Path, - headers: HeaderMap, Json(binding): Json, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { + let context = match context { Ok(context) => context, Err(response) => return response, }; @@ -613,13 +602,9 @@ async fn put_reference( async fn get_reference( State(state): State>, + context: ChannelContext, Path(path): Path, - headers: HeaderMap, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; let Ok(reference) = Reference::parse(path.reference) else { return api_error( StatusCode::BAD_REQUEST, @@ -658,13 +643,9 @@ async fn get_reference( async fn delete_reference( State(state): State>, + context: ChannelContext, Path(path): Path, - headers: HeaderMap, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; let Ok(reference) = Reference::parse(path.reference) else { return api_error( StatusCode::BAD_REQUEST, @@ -908,20 +889,11 @@ async fn channel_lease( let channel = channel .parse::() .map_err(|_| invalid_channel())?; - let credential = credential(headers); - match state + state .channels - .authorize_with_lease(channel, credential.as_deref()) + .authorize_with_lease(channel, credential(headers).as_deref()) .await - { - Ok(lease) => Ok(lease), - Err(error) => { - if matches!(error, ChannelError::Unauthorized) { - state.metrics.authorization_denial(); - } - Err(channel_failure(error)) - } - } + .map_err(|error| channel_denied(state, error)) } /// Validates channel credentials and `active` state without taking the lifecycle gate. @@ -934,20 +906,20 @@ async fn authorize_channel( let channel = channel .parse::() .map_err(|_| invalid_channel())?; - let credential = credential(headers); - match state + state .channels - .authorize(channel, credential.as_deref()) + .authorize(channel, credential(headers).as_deref()) .await - { - Ok(record) => Ok(record), - Err(error) => { - if matches!(error, ChannelError::Unauthorized) { - state.metrics.authorization_denial(); - } - Err(channel_failure(error)) - } + .map_err(|error| channel_denied(state, error)) +} + +/// The response for a rejected channel authorization, counting the denials that were +/// credential failures rather than missing or deleting channels. +fn channel_denied(state: &Arc, error: ChannelError) -> Response { + if matches!(error, ChannelError::Unauthorized) { + state.metrics.authorization_denial(); } + channel_failure(error) } fn invalid_channel() -> Response { @@ -977,7 +949,9 @@ fn credential(headers: &HeaderMap) -> Option { fn channel_failure(error: ChannelError) -> Response { match error { - ChannelError::NotFound => api_error( + // A channel being deleted is indistinguishable from an absent one to a client: + // its data is already unreachable and its id will never be reused. + ChannelError::NotFound | ChannelError::Deleting => api_error( StatusCode::NOT_FOUND, "channel_not_found", "channel does not exist", @@ -994,11 +968,6 @@ fn channel_failure(error: ChannelError) -> Response { ); response } - ChannelError::Deleting => api_error( - StatusCode::NOT_FOUND, - "channel_not_found", - "channel does not exist", - ), ChannelError::DefaultChannel => api_error( StatusCode::CONFLICT, "default_channel", @@ -1021,24 +990,17 @@ fn build_reference(kind: &str, key: &str) -> String { async fn put_http_cache( State(state): State>, + context: ChannelContext, Path(path): Path, headers: HeaderMap, body: Body, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; if !Reference::is_valid(&path.key) { return StatusCode::BAD_REQUEST.into_response(); } let Some(_permit) = acquire_foreground(&state) else { return busy_response(); }; - let content_type = headers - .get(header::CONTENT_TYPE) - .and_then(|value| value.to_str().ok()) - .map(str::to_owned); let content_length = content_length(&headers); match state .cache @@ -1047,7 +1009,7 @@ async fn put_http_cache( target: PublicationTarget::ContentAddressed { reference: Some(build_reference("http", &path.key)), }, - content_type, + content_type: content_type(&headers), stream: body.into_data_stream(), content_length, durability: Durability::BestEffort, @@ -1062,14 +1024,11 @@ async fn put_http_cache( async fn get_http_cache( State(state): State>, + context: ChannelContext, Path(path): Path, method: Method, headers: HeaderMap, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; serve_reference_artifact( state, context.channel, @@ -1108,31 +1067,24 @@ async fn serve_reference_artifact( async fn put_bazel_cas( State(state): State>, + context: ChannelContext, Path(path): Path, headers: HeaderMap, body: Body, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; let Ok(artifact) = ArtifactId::parse("sha256", &path.hash) else { return StatusCode::BAD_REQUEST.into_response(); }; let Some(_permit) = acquire_foreground(&state) else { return busy_response(); }; - let content_type = headers - .get(header::CONTENT_TYPE) - .and_then(|value| value.to_str().ok()) - .map(str::to_owned); let content_length = content_length(&headers); match state .cache .publish(PublishRequest { channel: context.channel, target: PublicationTarget::ById(artifact), - content_type, + content_type: content_type(&headers), stream: body.into_data_stream(), content_length, durability: Durability::Durable, @@ -1141,24 +1093,17 @@ async fn put_bazel_cas( .await { Ok(_) => StatusCode::OK.into_response(), - Err(CacheError::Local(crate::storage::local::LocalError::OutOfSpace)) => { - state.metrics.raw_pressure_error(); - insufficient_storage() - } - Err(error) => cache_write_failure(error), + Err(error) => cache_write_failure(&state, error), } } async fn get_bazel_cas( State(state): State>, + context: ChannelContext, Path(path): Path, method: Method, headers: HeaderMap, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; let Ok(artifact) = ArtifactId::parse("sha256", &path.hash) else { return StatusCode::BAD_REQUEST.into_response(); }; @@ -1174,24 +1119,17 @@ async fn get_bazel_cas( async fn put_bazel_ac( State(state): State>, + context: ChannelContext, Path(path): Path, headers: HeaderMap, body: Body, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; if ArtifactId::parse("sha256", &path.hash).is_err() { return StatusCode::BAD_REQUEST.into_response(); } let Some(_permit) = acquire_foreground(&state) else { return busy_response(); }; - let content_type = headers - .get(header::CONTENT_TYPE) - .and_then(|value| value.to_str().ok()) - .map(str::to_owned); let content_length = content_length(&headers); match state .cache @@ -1200,7 +1138,7 @@ async fn put_bazel_ac( target: PublicationTarget::ContentAddressed { reference: Some(build_reference("bazel-ac", &path.hash)), }, - content_type, + content_type: content_type(&headers), stream: body.into_data_stream(), content_length, durability: Durability::BestEffort, @@ -1215,14 +1153,11 @@ async fn put_bazel_ac( async fn get_bazel_ac( State(state): State>, + context: ChannelContext, Path(path): Path, method: Method, headers: HeaderMap, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return 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. @@ -1236,8 +1171,15 @@ async fn get_bazel_ac( .await } -fn cache_write_failure(error: CacheError) -> Response { +/// The response for a failed upload. Disk pressure is reported to the client as 507 +/// with a retry hint; the artifact and Bazel CAS routes have no protocol-level way to +/// say "stored elsewhere", so they cannot bypass the way the build-cache writes do. +fn cache_write_failure(state: &Arc, error: CacheError) -> Response { match error { + CacheError::Local(crate::storage::local::LocalError::OutOfSpace) => { + state.metrics.raw_pressure_error(); + insufficient_storage() + } CacheError::Local(crate::storage::local::LocalError::DigestMismatch) => { StatusCode::CONFLICT.into_response() } @@ -1248,7 +1190,7 @@ fn cache_write_failure(error: CacheError) -> Response { StatusCode::NOT_FOUND.into_response() } error => { - tracing::error!(error = %error, "build-cache write failed"); + tracing::error!(error = %error, "cache write failed"); StatusCode::INTERNAL_SERVER_ERROR.into_response() } } @@ -1264,16 +1206,14 @@ fn cache_write_failure_with_bypass(state: &Arc, error: CacheError) -> state.metrics.build_cache_bypass(); return StatusCode::OK.into_response(); } - cache_write_failure(error) + cache_write_failure(state, error) } -fn content_length(headers: &HeaderMap) -> Option { +fn content_type(headers: &HeaderMap) -> Option { headers - .get(header::CONTENT_LENGTH)? - .to_str() - .ok()? - .parse() - .ok() + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map(str::to_owned) } fn insufficient_storage() -> Response { diff --git a/src/transport/http/packages.rs b/src/transport/http/packages.rs index 218cb03..c72c29e 100644 --- a/src/transport/http/packages.rs +++ b/src/transport/http/packages.rs @@ -3,19 +3,16 @@ use crate::proxy::{Protocol, ProxyError, ProxyOutcome, Transform}; #[derive(Deserialize)] pub(super) struct ProxyPath { - channel: Option, path: String, } #[derive(Deserialize)] pub(super) struct EncodedPath { - channel: Option, encoded: String, } #[derive(Deserialize)] pub(super) struct CargoCratePath { - channel: Option, #[serde(rename = "crate")] name: String, version: String, @@ -23,13 +20,9 @@ pub(super) struct CargoCratePath { pub(super) async fn go( State(state): State>, + context: ChannelContext, Path(path): Path, - headers: HeaderMap, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; fetch( &state, context.channel, @@ -42,30 +35,27 @@ pub(super) async fn go( pub(super) async fn python_simple( State(state): State>, + context: ChannelContext, Path(path): Path, headers: HeaderMap, ) -> Response { - fetch_python_simple(&state, path.channel.as_deref(), &path.path, &headers).await + fetch_python_simple(&state, &context, &path.path, &headers).await } pub(super) async fn python_simple_root( State(state): State>, - Path(path): Path, + context: ChannelContext, headers: HeaderMap, ) -> Response { - fetch_python_simple(&state, path.channel.as_deref(), "", &headers).await + fetch_python_simple(&state, &context, "", &headers).await } async fn fetch_python_simple( state: &Arc, - channel: Option<&str>, + context: &ChannelContext, path: &str, headers: &HeaderMap, ) -> Response { - let context = match ChannelContext::resolve(state, channel, headers).await { - Ok(context) => context, - Err(response) => return response, - }; let accept = headers .get(header::ACCEPT) .and_then(|value| value.to_str().ok()); @@ -77,13 +67,9 @@ async fn fetch_python_simple( pub(super) async fn python_file( State(state): State>, + context: ChannelContext, Path(path): Path, - headers: HeaderMap, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; let (encoded, suffix) = [".metadata", ".asc"] .into_iter() .find_map(|suffix| { @@ -92,47 +78,39 @@ pub(super) async fn python_file( .map(|encoded| (encoded, suffix)) }) .unwrap_or((&path.encoded, "")); - match state - .proxy - .fetch_encoded_url_with_suffix(context.channel, Protocol::Python, encoded, suffix) - .await - { - Ok(outcome) => outcome_response(&state, context.channel, outcome).await, - Err(error) => proxy_failure(error), - } + fetch_encoded(&state, context.channel, Protocol::Python, encoded, suffix).await } pub(super) async fn npm( State(state): State>, + context: ChannelContext, Path(path): Path, headers: HeaderMap, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; + if let Some(encoded) = path.path.strip_prefix("-/tarball/") { + return fetch_encoded(&state, context.channel, Protocol::Npm, encoded, "").await; + } let accept = headers .get(header::ACCEPT) .and_then(|value| value.to_str().ok()); - fetch_npm( + let Some(transform) = Transform::npm_metadata(context.route_prefix(), accept) else { + return StatusCode::NOT_ACCEPTABLE.into_response(); + }; + fetch( &state, context.channel, + Protocol::Npm, &path.path, - context.route_prefix(), - accept, + transform, ) .await } pub(super) async fn cargo_index( State(state): State>, + context: ChannelContext, Path(path): Path, - headers: HeaderMap, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; fetch( &state, context.channel, @@ -143,15 +121,7 @@ pub(super) async fn cargo_index( .await } -pub(super) async fn cargo_config( - State(state): State>, - Path(path): Path, - headers: HeaderMap, -) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; +pub(super) async fn cargo_config(context: ChannelContext) -> Response { Json(serde_json::json!({ "dl": format!("{}/proxy/cargo/crates", context.route_prefix()), "api": null, @@ -162,32 +132,12 @@ pub(super) async fn cargo_config( pub(super) async fn cargo_crate( State(state): State>, + context: ChannelContext, Path(path): Path, - headers: HeaderMap, ) -> Response { - let context = match ChannelContext::resolve(&state, path.channel.as_deref(), &headers).await { - Ok(context) => context, - Err(response) => return response, - }; fetch_cargo_crate(&state, context.channel, &path.name, &path.version).await } -async fn fetch_npm( - state: &Arc, - channel: ChannelId, - path: &str, - prefix: String, - accept: Option<&str>, -) -> Response { - if let Some(encoded) = path.strip_prefix("-/tarball/") { - return fetch_encoded(state, channel, Protocol::Npm, encoded).await; - } - let Some(transform) = Transform::npm_metadata(prefix, accept) else { - return StatusCode::NOT_ACCEPTABLE.into_response(); - }; - fetch(state, channel, Protocol::Npm, path, transform).await -} - async fn fetch_cargo_crate( state: &Arc, channel: ChannelId, @@ -219,10 +169,11 @@ async fn fetch_encoded( channel: ChannelId, protocol: Protocol, encoded: &str, + suffix: &str, ) -> Response { match state .proxy - .fetch_encoded_url(channel, protocol, encoded) + .fetch_encoded_url(channel, protocol, encoded, suffix) .await { Ok(outcome) => outcome_response(state, channel, outcome).await, @@ -262,25 +213,23 @@ async fn outcome_response( status, body, content_type, - } => { - let mut builder = Response::builder().status(status); - if let Some(content_type) = content_type { - builder = builder.header(header::CONTENT_TYPE, content_type); - } - builder.body(Body::from(body)).unwrap() - } + } => upstream_response(status, content_type, Body::from(body)), ProxyOutcome::UpstreamStream { status, body, content_type, - } => { - let mut builder = Response::builder().status(status); - if let Some(content_type) = content_type { - builder = builder.header(header::CONTENT_TYPE, content_type); - } - builder.body(Body::from_stream(body)).unwrap() - } + } => upstream_response(status, content_type, Body::from_stream(body)), + } +} + +/// Passes an upstream response through unchanged apart from its content type, whether +/// the body was buffered or is being streamed past the cache under disk pressure. +fn upstream_response(status: StatusCode, content_type: Option, body: Body) -> Response { + let mut builder = Response::builder().status(status); + if let Some(content_type) = content_type { + builder = builder.header(header::CONTENT_TYPE, content_type); } + builder.body(body).unwrap() } fn proxy_failure(error: ProxyError) -> Response { diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 3883215..e070d48 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -1 +1,15 @@ pub mod http; + +use ::http::{HeaderMap, header}; + +/// Parses a `Content-Length` so a body can be sized before it is read: uploads size +/// their reservation from the request header, proxied downloads from the upstream +/// response header. +pub(crate) fn content_length(headers: &HeaderMap) -> Option { + headers + .get(header::CONTENT_LENGTH)? + .to_str() + .ok()? + .parse() + .ok() +}