refactor: extract the channel context, share Accept parsing, and fold duplicated blocks

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<dyn FreeSpace>, 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

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

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

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<RocksMetadata>,
     files: Arc<ArtifactFiles>,
     gates: Arc<ChannelGates>,
+    clock: Arc<dyn Clock>,
 }
 
 pub struct IssuedChannel {
@@ -59,11 +60,13 @@ impl ChannelService {
         store: Arc<RocksMetadata>,
         files: Arc<ArtifactFiles>,
         gates: Arc<ChannelGates>,
+        clock: Arc<dyn Clock>,
     ) -> 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<ChannelLease, ChannelError> {
-        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(&current, 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")]

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)