From a6d444a5d1fb469c7a6c5208e91bf8c1270bd6dc Mon Sep 17 00:00:00 2001 From: Rob Lyon Date: Tue, 21 Jul 2026 08:26:41 -0700 Subject: [PATCH] refactor: delete dead code and guards against unreachable states Removes the http-body-util dependency, the references stripe array (512 mutexes guarding two single atomic batch writes with no read-modify-write), four unreferenced SpaceLedger accessors, FromStr for Digest, ArtifactId::algorithm, ChannelService::gate, the publish_or_reject passthrough, the prefetch pool's active/peak atomics, empty_output's OnceLock, AwaitedRemoval, deleting_channels, the PredictedObject/PublishedBody/Fetched wrappers, and session_finished. The eviction column family's stored_len value was written at two sites, discarded by its only reader, and little-endian while its key is big-endian; it is now empty. Old stores stay readable because the reader already ignored it. Also drops guards against states that cannot occur: Ring::owner's checked_rem after the empty case returns, the u32 member index that saved no memory but cost a fallible conversion, ChannelId::new's nil-ULID retry loop, RecentUse::new's max(1) behind a validated config, ensure_default's id check, prefix_end's carry loop over a base32 prefix whose last byte is never 0xFF, decode_eviction_key's repeated starts_with and two unreachable map_err arms, and serve_artifact's unsatisfiable pre-check along with the unreachable! and the Eq/PartialEq derives that existed only to serve it. --- Cargo.lock | 1 - Cargo.toml | 1 - src/agent/ring.rs | 12 ++-- src/artifact/identity.rs | 24 ++----- src/cache/mod.rs | 3 +- src/cache/recent_use.rs | 5 +- src/cache/service.rs | 26 ++----- src/cache/space.rs | 25 ------- src/cache/space_test.rs | 8 +-- src/cache/stripes.rs | 21 +----- src/cacheprog/mod.rs | 14 ++-- src/cacheprog/session.rs | 103 +++++++++------------------- src/channel/identity.rs | 7 +- src/channel/service.rs | 23 +++---- src/storage/local/artifact_files.rs | 65 +++++------------- src/storage/metadata/rocksdb.rs | 78 ++++++++------------- src/transport/http/mod.rs | 23 +++---- tests/integration/domain.rs | 2 +- 18 files changed, 131 insertions(+), 310 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 477d513..f7aeb4e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -630,7 +630,6 @@ dependencies = [ "hickory-resolver", "html-escape", "http", - "http-body-util", "lol_html", "postcard", "prometheus", diff --git a/Cargo.toml b/Cargo.toml index d4679c6..8b12783 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,7 +19,6 @@ hex = "0.4.3" hickory-resolver = "0.26.1" html-escape = "0.2.13" http = "1.4.0" -http-body-util = "0.1.3" lol_html = "3.0.0" postcard = { version = "1.1.3", features = ["use-std"] } prometheus = "0.14.0" diff --git a/src/agent/ring.rs b/src/agent/ring.rs index 3a90816..688b96e 100644 --- a/src/agent/ring.rs +++ b/src/agent/ring.rs @@ -51,7 +51,7 @@ pub struct RingMember { #[derive(Debug)] pub struct Ring { members: Vec, - points: Vec<(u64, u32)>, + points: Vec<(u64, usize)>, fingerprint: String, } @@ -69,7 +69,7 @@ impl Ring { hasher.update(vnode.to_be_bytes()); let digest = hasher.finalize(); let point = u64::from_be_bytes(digest[..8].try_into().expect("sha256 slice")); - points.push((point, u32::try_from(index).expect("member count fits u32"))); + points.push((point, index)); } } points.sort_unstable(); @@ -100,13 +100,9 @@ impl Ring { if self.points.is_empty() { return None; } - let index = self - .points - .partition_point(|(point, _)| *point < position) - .checked_rem(self.points.len()) - .expect("points is non-empty"); + let index = self.points.partition_point(|(point, _)| *point < position) % self.points.len(); let (_, member) = self.points[index]; - Some(&self.members[member as usize]) + Some(&self.members[member]) } /// Hex SHA-256 of the sorted member IDs joined with `\n`. Diagnostic only. diff --git a/src/artifact/identity.rs b/src/artifact/identity.rs index 46c33d4..462c383 100644 --- a/src/artifact/identity.rs +++ b/src/artifact/identity.rs @@ -1,19 +1,17 @@ use serde::{Deserialize, Serialize}; -use std::{fmt, str::FromStr}; +use std::fmt; const SHA256_BYTES: usize = 32; -const SHA256_HEX_LEN: usize = SHA256_BYTES * 2; #[derive(Clone, Copy, Eq, Hash, PartialEq, Serialize, Deserialize)] pub struct Digest([u8; SHA256_BYTES]); impl Digest { pub fn parse(value: &str) -> Result { - if value.len() != SHA256_HEX_LEN - || value - .bytes() - .any(|byte| !byte.is_ascii_hexdigit() || byte.is_ascii_uppercase()) - { + // `decode_to_slice` already rejects a wrong length and any non-hex byte; only + // the canonical lowercase form is a valid identity, so uppercase hex — which + // decodes fine — has to be rejected separately. + if value.bytes().any(|byte| byte.is_ascii_uppercase()) { return Err(IdentityError::InvalidDigest); } @@ -46,14 +44,6 @@ impl fmt::Display for Digest { } } -impl FromStr for Digest { - type Err = IdentityError; - - fn from_str(value: &str) -> Result { - Self::parse(value) - } -} - #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)] pub struct ArtifactId { digest: Digest, @@ -75,10 +65,6 @@ impl ArtifactId { Self { digest } } - pub fn algorithm(&self) -> &'static str { - Self::ALGORITHM - } - pub fn digest(&self) -> Digest { self.digest } diff --git a/src/cache/mod.rs b/src/cache/mod.rs index 950a9ba..e3a6ee9 100644 --- a/src/cache/mod.rs +++ b/src/cache/mod.rs @@ -19,4 +19,5 @@ pub use service::{ Admission, CacheError, CacheService, LocatedArtifact, Publication, PublicationOutcome, PublicationTarget, PublishRequest, }; -pub use space::{FreeSpace, Mode, SpaceLedger, SpacePolicy, SpaceSnapshot, StatvfsFreeSpace}; +pub(crate) use space::Mode; +pub use space::{FreeSpace, SpaceLedger, SpacePolicy, SpaceSnapshot, StatvfsFreeSpace}; diff --git a/src/cache/recent_use.rs b/src/cache/recent_use.rs index 89dc3f8..c56eba8 100644 --- a/src/cache/recent_use.rs +++ b/src/cache/recent_use.rs @@ -20,8 +20,9 @@ pub(crate) struct RecentUse { impl RecentUse { pub(crate) fn new(bits: usize) -> Self { - // Round up to whole 64-bit words, with a floor so the filter is never empty. - let words = bits.div_ceil(64).max(1); + // Round up to whole 64-bit words. `Config::validate` rejects a zero width, so + // the filter is never empty. + let words = bits.div_ceil(64); Self { filters: [new_words(words), new_words(words)], bits: words * 64, diff --git a/src/cache/service.rs b/src/cache/service.rs index 9e49a8c..ce3e2cc 100644 --- a/src/cache/service.rs +++ b/src/cache/service.rs @@ -1,7 +1,8 @@ use crate::{ artifact::{ArtifactId, ArtifactMetadata, StoredEncoding}, + cache::Mode, cache::recent_use::RecentUse, - cache::space::{Mode, SpaceLedger}, + cache::space::SpaceLedger, cache::stripes::Stripes, channel::{ChannelGates, ChannelId, ChannelStoreError, Lifecycle}, clock::Clock, @@ -114,31 +115,20 @@ impl CacheService { S: Stream> + Unpin, E: std::fmt::Display, { - match self.publish_impl(request).await? { + match self.publish_or_reject(request).await? { Admission::Published(publication) => Ok(publication), Admission::Rejected(_) => Err(CacheError::Local(LocalError::OutOfSpace)), } } - /// Content-addressed publication that hands the body back instead of failing when - /// the up-front reservation is rejected under disk pressure. The package proxy uses - /// this so a bypassed download streams the untouched body straight to the client — - /// never buffered, never re-fetched. + /// Publication that hands the body back instead of failing when the up-front + /// reservation is rejected under disk pressure. The package proxy uses this so a + /// bypassed download streams the untouched body straight to the client — never + /// buffered, never re-fetched. pub async fn publish_or_reject( &self, request: PublishRequest, ) -> Result, CacheError> - where - S: Stream> + Unpin, - E: std::fmt::Display, - { - self.publish_impl(request).await - } - - async fn publish_impl( - &self, - request: PublishRequest, - ) -> Result, CacheError> where S: Stream> + Unpin, E: std::fmt::Display, @@ -243,7 +233,6 @@ impl CacheService { durability: Durability, ) -> Result<(), CacheError> { let (_gate, _) = self.channel_fence(channel).await?; - let _stripe = self.stripes.reference(channel, &reference).await; self.metadata .bind_reference( channel, @@ -274,7 +263,6 @@ impl CacheService { reference: &str, ) -> Result<(), CacheError> { let (_gate, _) = self.channel_fence(channel).await?; - let _stripe = self.stripes.reference(channel, reference).await; self.metadata.delete_reference(channel, reference).await?; Ok(()) } diff --git a/src/cache/space.rs b/src/cache/space.rs index 8506be9..dac9f82 100644 --- a/src/cache/space.rs +++ b/src/cache/space.rs @@ -133,13 +133,6 @@ impl SpaceLedger { state.free_observed.saturating_sub(spoken_for) } - /// Bytes available for new staging reservations after subtracting outstanding - /// reservations, bytes committed since the last observation, and headroom. - pub fn available(&self) -> u64 { - let state = self.state.lock().expect("space ledger poisoned"); - self.available_locked(&state) - } - /// The current maintenance mode using low/high watermark hysteresis. pub fn mode(&self) -> Mode { let mut state = self.state.lock().expect("space ledger poisoned"); @@ -165,24 +158,6 @@ impl SpaceLedger { self.state.lock().expect("space ledger poisoned").degraded } - pub fn free_observed(&self) -> u64 { - self.state - .lock() - .expect("space ledger poisoned") - .free_observed - } - - pub fn reserved(&self) -> u64 { - self.state.lock().expect("space ledger poisoned").reserved - } - - pub fn committed_since(&self) -> u64 { - self.state - .lock() - .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 { diff --git a/src/cache/space_test.rs b/src/cache/space_test.rs index 3cc4009..bd0f209 100644 --- a/src/cache/space_test.rs +++ b/src/cache/space_test.rs @@ -31,14 +31,14 @@ fn concurrent_reservations_never_double_spend_the_same_capacity() { assert!(ledger.try_reserve(60)); assert!(ledger.try_reserve(40)); assert!(!ledger.try_reserve(1)); - assert_eq!(ledger.reserved(), 100); + assert_eq!(ledger.snapshot().reserved, 100); // Committing frees the reservation slot but keeps the bytes accounted until the // next filesystem observation. ledger.commit(60); - assert_eq!(ledger.reserved(), 40); + assert_eq!(ledger.snapshot().reserved, 40); assert!(!ledger.try_reserve(1)); - assert_eq!(ledger.committed_since(), 60); + assert_eq!(ledger.snapshot().committed_since, 60); } #[test] @@ -132,7 +132,7 @@ fn refresh_failure_retains_last_observation_but_stops_reservations() { fail.store(true, Ordering::SeqCst); ledger.refresh(); assert!(ledger.degraded()); - assert_eq!(ledger.free_observed(), 100); + assert_eq!(ledger.snapshot().free_observed, 100); assert!(!ledger.try_reserve(1)); // Recovery clears the degraded state. diff --git a/src/cache/stripes.rs b/src/cache/stripes.rs index 2e6e3a8..795b3e1 100644 --- a/src/cache/stripes.rs +++ b/src/cache/stripes.rs @@ -6,19 +6,17 @@ use tokio::sync::{Mutex, MutexGuard}; /// so mutation-serialization memory is bounded. const STRIPES: usize = 512; -/// Fixed arrays of asynchronous mutexes that serialize only the final metadata and -/// file transition for one `( channel, digest)` or `( channel, reference)`. -/// Staging, hashing, reads, and physical unlink all run outside these locks. +/// A fixed array of asynchronous mutexes that serializes only the final metadata and +/// file transition for one `( channel, digest)`. Staging, hashing, reads, and physical +/// unlink all run outside these locks. pub(crate) struct Stripes { artifacts: Box<[Mutex<()>]>, - references: Box<[Mutex<()>]>, } impl Stripes { pub(crate) fn new() -> Self { Self { artifacts: (0..STRIPES).map(|_| Mutex::new(())).collect(), - references: (0..STRIPES).map(|_| Mutex::new(())).collect(), } } @@ -34,17 +32,4 @@ impl Stripes { .lock() .await } - - pub(crate) async fn reference( - &self, - channel: ChannelId, - reference: &str, - ) -> MutexGuard<'_, ()> { - let mut hasher = std::collections::hash_map::DefaultHasher::new(); - channel.hash(&mut hasher); - reference.hash(&mut hasher); - self.references[(hasher.finish() as usize) % self.references.len()] - .lock() - .await - } } diff --git a/src/cacheprog/mod.rs b/src/cacheprog/mod.rs index cfbe6ca..d3beec3 100644 --- a/src/cacheprog/mod.rs +++ b/src/cacheprog/mod.rs @@ -179,8 +179,6 @@ where args.prefetch_concurrency, Arc::clone(&session_state), ))); - let mut session_finished = false; - write_response( &mut writer, &Response { @@ -206,7 +204,6 @@ where { finish_session( &mut prefetch_task, - &mut session_finished, &client, &base, args.token.as_deref(), @@ -288,7 +285,6 @@ where } finish_session( &mut prefetch_task, - &mut session_finished, &client, &base, args.token.as_deref(), @@ -306,20 +302,18 @@ where Ok(()) } -/// Ends the session exactly once: stop predicting, then persist what this build -/// used so the next one can prefetch it. +/// Ends the session: stop predicting, then persist what this build used so the next +/// one can prefetch it. Both halves are idempotent — the join handle is taken out of +/// the option and `finalize` drains the usage map — so the close path calling this and +/// then falling through to the same call after the loop costs nothing. async fn finish_session( task: &mut Option>, - finished: &mut bool, client: &reqwest::Client, base: &reqwest::Url, token: Option<&str>, manifest_key: &str, state: &SessionState, ) { - if std::mem::replace(finished, true) { - return; - } if let Some(task) = task.take() { task.abort(); let _ = task.await; diff --git a/src/cacheprog/session.rs b/src/cacheprog/session.rs index 788778e..406d41f 100644 --- a/src/cacheprog/session.rs +++ b/src/cacheprog/session.rs @@ -19,10 +19,7 @@ use sha2::{Digest as _, Sha256}; use std::{ collections::{HashMap, HashSet}, path::{Path, PathBuf}, - sync::{ - Arc, Mutex, OnceLock, - atomic::{AtomicUsize, Ordering}, - }, + sync::{Arc, Mutex, OnceLock}, time::Duration, }; use tokio::{io::AsyncWriteExt, sync::watch}; @@ -224,10 +221,7 @@ pub fn unix_now() -> u64 { /// The hex digest naming the empty body: every zero-size action resolves to this /// one output, so a single synthesized empty file answers all of them locally. -fn empty_output() -> &'static str { - static EMPTY: OnceLock = OnceLock::new(); - EMPTY.get_or_init(|| hex::encode(Sha256::digest([]))) -} +const EMPTY_OUTPUT: &str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"; /// The startup task: fetch the stored manifest by its cache key — the same plain /// GET `finalize` issues — then warm the object directory with bounded parallel @@ -289,7 +283,7 @@ async fn prefetch( // file locally instead of downloading nothing over the network. A // zero-size entry under any other digest can never verify, so it is // dropped from the work list entirely. - if entry.output == empty_output() { + if entry.output == EMPTY_OUTPUT { super::write_disk_file(directory, &entry.output, &[]).await?; local += 1; } @@ -320,14 +314,10 @@ async fn prefetch( } let limit = Arc::new(tokio::sync::Semaphore::new(concurrency)); - let active = AtomicUsize::new(0); - let peak = AtomicUsize::new(0); let mut downloads: FuturesUnordered<_> = wanted .iter() .map(|&(action, entry)| { let limit = Arc::clone(&limit); - let active = &active; - let peak = &peak; // Register before the download queues behind the concurrency bound so // a foreground get for any wanted output — queued or actively // downloading — waits for the pool instead of duplicating the fetch. @@ -338,24 +328,18 @@ async fn prefetch( .acquire_owned() .await .expect("prefetch semaphore is never closed"); - let running = active.fetch_add(1, Ordering::Relaxed) + 1; - peak.fetch_max(running, Ordering::Relaxed); - let outcome = download_object( + match download_object( client, base, token, directory, - PredictedObject { - session_label: manifest_key, - concurrency, - action, - entry, - }, + manifest_key, + concurrency, + (action, entry), ) - .await; - active.fetch_sub(1, Ordering::Relaxed); - match outcome { - Ok(fetched) => Some(fetched), + .await + { + Ok(published) => published, Err(error) => { // Every per-object failure is just a future foreground miss. tracing::debug!(%error, "prefetch download skipped"); @@ -368,76 +352,57 @@ async fn prefetch( let attempted = downloads.len(); let mut downloaded = 0usize; let mut bytes = 0u64; - let mut published_meanwhile = 0usize; - while let Some(result) = downloads.next().await { - match result { - Some(Fetched::Published(size)) => { - downloaded += 1; - bytes += size; - } - Some(Fetched::AlreadyLocal) => published_meanwhile += 1, - Some(Fetched::Miss) | None => {} + while let Some(published) = downloads.next().await { + if let Some(size) = published { + downloaded += 1; + bytes += size; } } drop(downloads); tracing::debug!( advertised, - local = local + published_meanwhile, + local, attempted, downloaded, - missed = attempted - downloaded - published_meanwhile, + missed = attempted - downloaded, bytes, duration_ms = started.elapsed().as_millis() as u64, - peak_concurrency = peak.load(Ordering::Relaxed), "build-cache prefetch complete" ); Ok(()) } -enum Fetched { - /// Downloaded, verified, and published, carrying the object size. - Published(u64), - /// A foreground get published the object while this download was queued. - AlreadyLocal, - /// The server does not hold the object. - Miss, -} - -struct PredictedObject<'a> { - session_label: &'a str, - concurrency: usize, - action: &'a str, - entry: &'a ManifestEntry, -} - -/// Downloads one predicted object through the ordinary cache route — the exact GET -/// a foreground miss would issue, plus the telemetry purpose header — verifying -/// size and digest against the manifest entry before publishing it into the local -/// object directory. +/// Downloads one predicted `(action, entry)` through the ordinary cache route — the +/// exact GET a foreground miss would issue, plus the telemetry purpose header — +/// verifying size and digest against the manifest entry before publishing it into +/// the local object directory. Returns the object's size when this call published +/// it, and `None` when the server does not hold it or a foreground get won the race. async fn download_object( client: &reqwest::Client, base: &reqwest::Url, token: Option<&str>, directory: &Path, - object: PredictedObject<'_>, -) -> anyhow::Result { - let path = directory.join(&object.entry.output); + manifest_key: &str, + concurrency: usize, + (action, entry): (&str, &ManifestEntry), +) -> anyhow::Result> { + let path = directory.join(&entry.output); // The work list was computed at startup; a foreground get may have published // this object while the download waited behind the concurrency bound. - if super::file_has_size(&path, object.entry.size).await { - return Ok(Fetched::AlreadyLocal); + if super::file_has_size(&path, entry.size).await { + return Ok(None); } let mut request = client - .get(base.join(&format!("go-{}", object.action))?) + .get(base.join(&format!("go-{action}"))?) .header(REQUEST_PURPOSE_HEADER, REQUEST_PURPOSE_PREFETCH) - .header(REQUEST_PREFETCH_CONCURRENCY_HEADER, object.concurrency) - .header(REQUEST_SESSION_HEADER, object.session_label); + .header(REQUEST_PREFETCH_CONCURRENCY_HEADER, concurrency) + .header(REQUEST_SESSION_HEADER, manifest_key); if let Some(token) = token { request = request.bearer_auth(token); } let response = request.send().await?; if response.status() == reqwest::StatusCode::NOT_FOUND { - return Ok(Fetched::Miss); + return Ok(None); } anyhow::ensure!( response.status().is_success(), @@ -445,7 +410,7 @@ async fn download_object( response.status() ); let temporary = super::temporary_path(directory); - if let Err(error) = spool_verified(response, &temporary, object.entry).await { + if let Err(error) = spool_verified(response, &temporary, entry).await { let _ = tokio::fs::remove_file(&temporary).await; return Err(error); } @@ -455,7 +420,7 @@ async fn download_object( return Err(error.into()); } } - Ok(Fetched::Published(object.entry.size)) + Ok(Some(entry.size)) } /// Streams the response body into the temporary file while hashing incrementally, diff --git a/src/channel/identity.rs b/src/channel/identity.rs index b7519e2..fa9b067 100644 --- a/src/channel/identity.rs +++ b/src/channel/identity.rs @@ -9,12 +9,7 @@ impl ChannelId { pub const DEFAULT: Self = Self(Ulid::nil()); pub fn new() -> Self { - loop { - let id = Self(Ulid::new()); - if id != Self::DEFAULT { - return id; - } - } + Self(Ulid::new()) } /// The channel's durable key prefix: the canonical ULID text, encoded straight diff --git a/src/channel/service.rs b/src/channel/service.rs index a5bb6f6..71da6fc 100644 --- a/src/channel/service.rs +++ b/src/channel/service.rs @@ -89,13 +89,10 @@ impl ChannelService { } } }; - if record.id != ChannelId::DEFAULT - || record.access != Access::Open - || record.state != Lifecycle::Active - { + if record.access != Access::Open || record.state != Lifecycle::Active { return Err(ChannelError::InvalidDefault); } - self.gate(ChannelId::DEFAULT); + self.gates.gate(ChannelId::DEFAULT); Ok(()) } @@ -115,7 +112,7 @@ impl ChannelService { created_at: unix_time(), }; self.store.create_channel(record.clone()).await?; - self.gate(record.id); + self.gates.gate(record.id); Ok(IssuedChannel { record, token }) } @@ -131,7 +128,7 @@ impl ChannelService { if record.state != Lifecycle::Active { return Err(ChannelError::Deleting); } - let guard = self.gate(id).read_owned().await; + let guard = self.gates.gate(id).read_owned().await; let Some(current) = self.store.channel(id).await? else { return Err(ChannelError::NotFound); }; @@ -187,14 +184,16 @@ impl ChannelService { self.store.store_channel(deleting).await?; drop(lease); - let gate = self.gate(id); + let gate = self.gates.gate(id); let _exclusive = gate.write().await; self.finish_deletion(id).await } pub async fn resume_deletions(&self) -> Result<(), ChannelError> { - for channel in self.store.deleting_channels().await? { - let gate = self.gate(channel.id); + let mut deleting = self.store.channels().await?; + deleting.retain(|channel| channel.state == Lifecycle::Deleting); + for channel in deleting { + let gate = self.gates.gate(channel.id); let _exclusive = gate.write().await; self.finish_deletion(channel.id).await?; } @@ -208,10 +207,6 @@ impl ChannelService { self.gates.forget(id); Ok(()) } - - fn gate(&self, id: ChannelId) -> Arc> { - self.gates.gate(id) - } } fn authorize_record(record: &ChannelRecord, credential: Option<&str>) -> Result<(), ChannelError> { diff --git a/src/storage/local/artifact_files.rs b/src/storage/local/artifact_files.rs index 6fc1bfe..cba2728 100644 --- a/src/storage/local/artifact_files.rs +++ b/src/storage/local/artifact_files.rs @@ -88,31 +88,6 @@ fn schedule_removal(path: PathBuf, settlement: PendingSettlement) { }); } -/// Makes an explicitly awaited discard cancellation-safe. If its caller is dropped -/// while the async unlink is pending, this guard schedules one best-effort retry using -/// the same conservative accounting fallback as `Reservation::drop`. -struct AwaitedRemoval { - path: Option, - settlement: Option, -} - -impl AwaitedRemoval { - async fn run(mut self) { - let result = tokio::fs::remove_file(self.path.as_ref().expect("path is present")).await; - self.path.take(); - let settlement = self.settlement.take().expect("settlement is present"); - settle_removal(result, settlement); - } -} - -impl Drop for AwaitedRemoval { - fn drop(&mut self) { - if let (Some(path), Some(settlement)) = (self.path.take(), self.settlement.take()) { - schedule_removal(path, settlement); - } - } -} - impl Reservation { fn new(reserver: Arc) -> Self { Self { @@ -164,21 +139,16 @@ impl Reservation { self.state = ReservationState::Published; } - /// Settles a failed operation before returning to its caller. Temporary files are - /// removed asynchronously; successful deletion and `NotFound` release the capacity, - /// while deletion failure or a published file commits it until refresh. + /// Settles a failed operation before returning to its caller. Successful deletion + /// of the temporary file and `NotFound` release the capacity, while deletion + /// failure or a published file commits it until refresh. pub(crate) async fn discard(mut self) { let state = std::mem::replace(&mut self.state, ReservationState::Vacant); let settlement = self.settlement(); match state { ReservationState::Vacant => settlement.release(), ReservationState::Temporary(path) => { - AwaitedRemoval { - path: Some(path), - settlement: Some(settlement), - } - .run() - .await; + settle_removal(tokio::fs::remove_file(path).await, settlement); } ReservationState::Published => settlement.commit(), } @@ -234,11 +204,6 @@ enum StageWriter { Zstd(ZstdEncoder), } -enum PublishedBody { - Created, - Existing, -} - impl StageWriter { fn new(file: tokio::fs::File, encoding: StoredEncoding) -> Self { match encoding { @@ -433,7 +398,7 @@ impl ArtifactFiles { mut reservation, .. } = staged; - let published = match self + let created = match self .publish_reserved( channel, artifact, @@ -444,20 +409,22 @@ impl ArtifactFiles { ) .await { - Ok(published) => published, + Ok(created) => created, Err(error) => { reservation.discard().await; return Err(error); } }; - Ok(match published { - PublishedBody::Created => FilePublication::Created(reservation), - PublishedBody::Existing => FilePublication::Existing(reservation), + Ok(if created { + FilePublication::Created(reservation) + } else { + FilePublication::Existing(reservation) }) } /// Performs the fallible pre-publication and rename work, leaving one awaited - /// discard site in `publish` for every ordinary failure. + /// discard site in `publish` for every ordinary failure. Returns whether this call + /// installed the body, as opposed to finding it already published. async fn publish_reserved( &self, channel: ChannelId, @@ -466,7 +433,7 @@ impl ArtifactFiles { digest: Digest, reservation: &mut Reservation, durability: Durability, - ) -> Result { + ) -> Result { if digest != artifact.digest() { return Err(LocalError::DigestMismatch); } @@ -476,14 +443,14 @@ impl ArtifactFiles { if tokio::fs::try_exists(&final_path).await? { tokio::fs::remove_file(staged_path).await?; reservation.mark_vacant(); - return Ok(PublishedBody::Existing); + return Ok(false); } match tokio::fs::rename(staged_path, &final_path).await { Ok(()) => {} Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { tokio::fs::remove_file(staged_path).await?; reservation.mark_vacant(); - return Ok(PublishedBody::Existing); + return Ok(false); } Err(error) => return Err(error.into()), } @@ -502,7 +469,7 @@ impl ArtifactFiles { .and_then(|result| result.map_err(LocalError::from)); synced?; } - Ok(PublishedBody::Created) + Ok(true) } pub fn path(&self, channel: ChannelId, artifact: ArtifactId) -> PathBuf { diff --git a/src/storage/metadata/rocksdb.rs b/src/storage/metadata/rocksdb.rs index dc27594..b5b57f1 100644 --- a/src/storage/metadata/rocksdb.rs +++ b/src/storage/metadata/rocksdb.rs @@ -1,7 +1,7 @@ use super::{Candidate, Durability, Evicted, MetadataError, ReferenceRecord}; use crate::{ artifact::{ArtifactId, ArtifactMetadata, Digest}, - channel::{ChannelId, ChannelRecord, ChannelStoreError, Lifecycle}, + channel::{ChannelId, ChannelRecord, ChannelStoreError}, storage::records::{decode_record, encode_record}, }; use rocksdb::{ColumnFamilyDescriptor, DB, IteratorMode, Options, WriteBatch, WriteOptions}; @@ -97,7 +97,9 @@ impl RocksMetadata { batch.put_cf( &eviction, eviction_key(channel, metadata.eligible_at, artifact), - metadata.stored_len.to_le_bytes(), + // The queue row is its key: deadline and artifact are both encoded + // there and the artifact record holds everything else. + b"", ); } if let Some((name, record)) = reference { @@ -235,7 +237,7 @@ impl RocksMetadata { if !key.starts_with(&prefix) || candidates.len() >= limit { break; } - let (eligible_at, artifact) = decode_eviction_key(channel, &key)?; + let (eligible_at, artifact) = decode_eviction_key(&key)?; candidates.push(Candidate { eligible_at, artifact, @@ -274,7 +276,7 @@ impl RocksMetadata { batch.put_cf( &eviction, eviction_key(channel, new_eligible_at, artifact), - metadata.stored_len.to_le_bytes(), + b"", ); batch.put_cf(&artifacts, &key, encode_record(&metadata)?); database.write(batch).map_err(store_error) @@ -382,26 +384,6 @@ impl RocksMetadata { .await } - pub(crate) async fn deleting_channels(&self) -> Result, ChannelStoreError> { - let database = Arc::clone(&self.database); - channel_blocking(move || { - let family = database - .cf_handle(CHANNELS) - .ok_or_else(channel_missing_cf)?; - database - .iterator_cf(&family, IteratorMode::Start) - .map(|item| { - let (_, bytes) = item.map_err(channel_store_error)?; - decode_record::(&bytes).map_err(ChannelStoreError::from) - }) - .filter( - |result| !matches!(result, Ok(record) if record.state != Lifecycle::Deleting), - ) - .collect() - }) - .await - } - pub(crate) async fn channels(&self) -> Result, ChannelStoreError> { let database = Arc::clone(&self.database); channel_blocking(move || { @@ -422,12 +404,12 @@ impl RocksMetadata { pub(crate) async fn delete_channel_data(&self, id: ChannelId) -> Result<(), ChannelStoreError> { let database = Arc::clone(&self.database); channel_blocking(move || { - let prefix = id.as_key().to_vec(); - let end = prefix_end(&prefix); + let prefix = id.as_key(); + let end = prefix_end(prefix); let mut batch = WriteBatch::default(); for name in [ARTIFACTS, REFERENCES, EVICTION] { let family = database.cf_handle(name).ok_or_else(channel_missing_cf)?; - batch.delete_range_cf(&family, &prefix, &end); + batch.delete_range_cf(&family, prefix, end); } channel_write_sync(&database, batch) }) @@ -499,21 +481,20 @@ fn eviction_key(channel: ChannelId, eligible_at: u64, artifact: ArtifactId) -> V key } -fn decode_eviction_key(channel: ChannelId, key: &[u8]) -> Result<(u64, ArtifactId), MetadataError> { - let prefix = channel.as_key(); - if !key.starts_with(&prefix) || key.len() != prefix.len() + 8 + 32 { - return Err(MetadataError::Store("invalid eviction key".to_owned())); - } - let eligible_at = u64::from_be_bytes( - key[prefix.len()..prefix.len() + 8] - .try_into() - .map_err(|_| MetadataError::Store("invalid eviction deadline".to_owned()))?, - ); - let digest = key[prefix.len() + 8..] +/// Splits a queue key into its deadline and artifact. The caller has already matched +/// the channel prefix, so the whole-key length is the only thing left to check. +fn decode_eviction_key(key: &[u8]) -> Result<(u64, ArtifactId), MetadataError> { + const DEADLINE_AT: usize = ulid::ULID_LEN; + const DIGEST_AT: usize = DEADLINE_AT + 8; + let key: &[u8; DIGEST_AT + 32] = key .try_into() - .map_err(|_| MetadataError::Store("invalid eviction digest".to_owned()))?; + .map_err(|_| MetadataError::Store("invalid eviction key".to_owned()))?; + let mut deadline = [0_u8; 8]; + deadline.copy_from_slice(&key[DEADLINE_AT..DIGEST_AT]); + let mut digest = [0_u8; 32]; + digest.copy_from_slice(&key[DIGEST_AT..]); Ok(( - eligible_at, + u64::from_be_bytes(deadline), ArtifactId::from_digest(Digest::from_bytes(digest)), )) } @@ -548,15 +529,10 @@ fn channel_store_error(error: impl std::fmt::Display) -> ChannelStoreError { ChannelStoreError::Store(error.to_string()) } -fn prefix_end(prefix: &[u8]) -> Vec { - let mut end = prefix.to_vec(); - for byte in end.iter_mut().rev() { - if *byte != u8::MAX { - *byte += 1; - return end; - } - *byte = 0; - } - end.push(0); - end +/// The exclusive upper bound of a channel's key range. A channel key is canonical +/// Crockford base32 text, so its last byte is an ASCII digit or capital letter and +/// incrementing it can never carry. +fn prefix_end(mut prefix: [u8; ulid::ULID_LEN]) -> [u8; ulid::ULID_LEN] { + prefix[ulid::ULID_LEN - 1] += 1; + prefix } diff --git a/src/transport/http/mod.rs b/src/transport/http/mod.rs index e78cd6e..087f4c6 100644 --- a/src/transport/http/mod.rs +++ b/src/transport/http/mod.rs @@ -561,19 +561,18 @@ async fn serve_artifact( } else { RangeSelection::Full }; - if range == RangeSelection::Unsatisfiable { - return Response::builder() - .status(StatusCode::RANGE_NOT_SATISFIABLE) - .header(header::ACCEPT_RANGES, "bytes") - .header(header::CONTENT_RANGE, format!("bytes */{content_len}")) - .header(header::ETAG, format!("\"{}\"", artifact)) - .body(Body::empty()) - .unwrap(); - } let (start, end, length) = match range { RangeSelection::Partial { start, end } => (start, end, end - start + 1), RangeSelection::Full => (0, content_len.saturating_sub(1), content_len), - RangeSelection::Unsatisfiable => unreachable!("handled above"), + RangeSelection::Unsatisfiable => { + return Response::builder() + .status(StatusCode::RANGE_NOT_SATISFIABLE) + .header(header::ACCEPT_RANGES, "bytes") + .header(header::CONTENT_RANGE, format!("bytes */{content_len}")) + .header(header::ETAG, format!("\"{}\"", artifact)) + .body(Body::empty()) + .unwrap(); + } }; // A zstd-stored response is passed through untouched when the client accepts zstd; // otherwise it serves the complete logical bytes through the decoder. @@ -734,7 +733,7 @@ async fn get_reference( .await { Ok(Some(record)) => Json(ArtifactBinding { - algorithm: record.artifact.algorithm().to_owned(), + algorithm: ArtifactId::ALGORITHM.to_owned(), digest: record.artifact.digest().to_string(), }) .into_response(), @@ -795,7 +794,7 @@ async fn delete_reference( } } -#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[derive(Clone, Copy, Debug)] enum RangeSelection { Full, Partial { start: u64, end: u64 }, diff --git a/tests/integration/domain.rs b/tests/integration/domain.rs index fe345c4..91c5ccc 100644 --- a/tests/integration/domain.rs +++ b/tests/integration/domain.rs @@ -14,7 +14,7 @@ fn artifact_identity_accepts_only_canonical_sha256() { let hex = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; let id = ArtifactId::parse("sha256", hex).expect("valid identity"); - assert_eq!(id.algorithm(), "sha256"); + assert_eq!(id.to_string(), format!("sha256:{hex}")); assert_eq!(id.digest().to_string(), hex); assert!(ArtifactId::parse("sha512", hex).is_err()); assert!(ArtifactId::parse("sha256", &hex.to_uppercase()).is_err());