perf: remove hot-path allocation, syscalls, and redundant fsyncs

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::<MatchedPath>() {
-        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::<MatchedPath>() {
+        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<Arc<AgentState>>, 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::<usize>().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::<usize>().ok());
         AgentPrefetchObservation::start(Arc::clone(&state), session, configured_concurrency)
     });
     let started = Instant::now();
@@ -358,14 +362,19 @@ async fn forward(State(state): State<Arc<AgentState>>, request: Request) -> Resp
 }
 
 async fn forward_request(state: &Arc<AgentState>, 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<AgentState>, 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<AgentState>, 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();
             }

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

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

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"
+    );
+}

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<String>,
         last_modified: Option<String>,
+        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<usize, CacheError> {
         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<OwnedRwLockReadGuard<()>, 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),
         }

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 {

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

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