refactor: delete dead code and guards against unreachable states

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"

src/agent/ring.rs

@@ -51,7 +51,7 @@ pub struct RingMember {
 #[derive(Debug)]
 pub struct Ring {
     members: Vec<RingMember>,
-    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.

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<Self, IdentityError> {
-        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, Self::Err> {
-        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
     }

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};

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,

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<Item = Result<Bytes, E>> + 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<S, E>(
         &self,
         request: PublishRequest<S>,
     ) -> Result<Admission<S>, CacheError>
-    where
-        S: Stream<Item = Result<Bytes, E>> + Unpin,
-        E: std::fmt::Display,
-    {
-        self.publish_impl(request).await
-    }
-
-    async fn publish_impl<S, E>(
-        &self,
-        request: PublishRequest<S>,
-    ) -> Result<Admission<S>, CacheError>
     where
         S: Stream<Item = Result<Bytes, E>> + 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(())
     }

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 {

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.

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
-    }
 }

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<tokio::task::JoinHandle<()>>,
-    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;