Move things around.

Author: milesj

Cargo.lock

@@ -18,12 +18,12 @@ use bazel_remote_apis::google::bytestream::{
 };
 use moon_blob::{BlobContent, BlobInput, BlobOutput, Bytes};
 use moon_cache_storage::{
-    CacheCapabilities, CacheContext, Compressor, DigestFunction, ExternalDigestExt,
-    InternalDigestExt, Manifest, StorageBackend, check_blob_integrity,
+    CacheCapabilities, CacheContext, Compressor, DigestFunction, Manifest, StorageBackend,
+    check_blob_integrity,
 };
 use moon_common::{Id, color, is_ci, is_remote};
 use moon_config::{RemoteCompression, RemoteConfig};
-use moon_hash::Digest;
+use moon_hash::{Digest, ExternalDigestExt, InternalDigestExt};
 use reqwest::header::HeaderMap;
 use std::fmt::Debug;
 use std::sync::{Arc, OnceLock};

crates/cache-remote/src/grpc_remote_storage.rs

@@ -13,6 +13,7 @@ moon_blob = { path = "../blob" }
 moon_common = { path = "../common" }
 moon_config = { path = "../config" }
 moon_hash = { path = "../hash" }
+moon_manifest = { path = "../manifest" }
 moon_process = { path = "../process" }
 async-trait = { workspace = true }
 bazel-remote-apis = { workspace = true }

crates/cache-storage/Cargo.toml

@@ -1,32 +1,7 @@
-use bazel_remote_apis::google::protobuf::Timestamp;
-use chrono::NaiveDateTime;
 use moon_common::BLOCKING_THREAD_COUNT;
 use moon_hash::Digest;
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tracing::warn;
 
-pub fn create_timestamp(time: SystemTime) -> Option<Timestamp> {
-    time.duration_since(UNIX_EPOCH)
-        .ok()
-        .map(|duration| Timestamp {
-            seconds: duration.as_secs() as i64,
-            nanos: duration.subsec_nanos() as i32,
-        })
-}
-
-pub fn create_timestamp_from_naive(time: NaiveDateTime) -> Option<Timestamp> {
-    let utc = time.and_utc();
-
-    Some(Timestamp {
-        seconds: utc.timestamp(),
-        nanos: utc.timestamp_subsec_nanos() as i32,
-    })
-}
-
-pub fn create_from_timestamp(timestamp: Timestamp) -> SystemTime {
-    UNIX_EPOCH + Duration::new(timestamp.seconds as u64, timestamp.nanos as u32)
-}
-
 pub fn check_blob_integrity(expected_digest: &Digest, bytes: &[u8]) -> miette::Result<bool> {
     if bytes.len() != expected_digest.size as usize {
         warn!(
@@ -302,20 +277,4 @@ mod tests {
             assert!(batches.iter().all(|batch| batch.items.len() == 3));
         }
     }
-
-    mod timestamps {
-        use super::*;
-
-        #[test]
-        fn round_trips_through_protobuf() {
-            let now = SystemTime::now();
-
-            let restored = create_from_timestamp(create_timestamp(now).unwrap());
-
-            let original = now.duration_since(UNIX_EPOCH).unwrap();
-            let after = restored.duration_since(UNIX_EPOCH).unwrap();
-            assert_eq!(original.as_secs(), after.as_secs());
-            assert_eq!(original.subsec_nanos(), after.subsec_nanos());
-        }
-    }
 }

crates/cache-storage/src/helpers.rs

@@ -1,13 +1,11 @@
 mod capabilities;
-mod digest_compat;
 mod helpers;
-mod manifest;
 mod storage;
 mod storage_backend;
 
 pub use capabilities::*;
-pub use digest_compat::*;
 pub use helpers::*;
-pub use manifest::*;
 pub use storage::*;
 pub use storage_backend::*;
+
+pub use moon_manifest::*;

crates/cache-storage/src/lib.rs

@@ -1,15 +1,16 @@
-use crate::manifest::{Manifest, ManifestSource};
 use crate::storage_backend::{BoxedStorageBackend, StorageBackend};
+use miette::IntoDiagnostic;
 use moon_blob::{BlobCleanStats, BlobContent, BlobInput, BlobOutput};
 use moon_common::{Id, format_error_chain, is_daemon_env};
 use moon_config::{CacheConfig, RemoteConfig};
 use moon_hash::Digest;
+use moon_manifest::Manifest;
 use rustc_hash::FxHashMap;
 use std::collections::VecDeque;
 use std::path::PathBuf;
 use std::sync::Arc;
-use std::sync::Mutex;
 use std::time::{Duration, SystemTime};
+use tokio::sync::Mutex;
 use tokio::task::{AbortHandle, JoinHandle, JoinSet};
 use tracing::{debug, warn};
 
@@ -19,6 +20,12 @@ use tracing::{debug, warn};
 /// aborted and reported, and simply get re-uploaded on the next run.
 const BACKGROUND_FLUSH_TIMEOUT: Duration = Duration::from_secs(60);
 
+pub struct ManifestSource {
+    pub backend: BoxedStorageBackend,
+    pub manifest: Manifest,
+    pub remote: bool,
+}
+
 #[derive(Clone, Debug)]
 pub struct CacheContext {
     pub cache_dir: PathBuf,
@@ -222,7 +229,7 @@ impl Storage {
     }
 
     pub async fn store_blobs(&self, blobs: Vec<BlobInput>) -> miette::Result<()> {
-        let mut background_tasks = self.background_tasks.lock().unwrap();
+        let mut background_tasks = self.background_tasks.lock().await;
 
         for backend in self.get_backends() {
             if !backend.is_writable() {
@@ -282,7 +289,7 @@ impl Storage {
         digest: &Digest,
         manifest: Manifest,
     ) -> miette::Result<()> {
-        let mut background_tasks = self.background_tasks.lock().unwrap();
+        let mut background_tasks = self.background_tasks.lock().await;
 
         debug!(
             hash = digest.hash.as_str(),
@@ -316,6 +323,10 @@ impl Storage {
             }
         }
 
+        while let Some(result) = set.join_next().await {
+            result.into_diagnostic()??;
+        }
+
         debug!(
             hash = digest.hash.as_str(),
             files = manifest.files.len(),
@@ -403,7 +414,7 @@ impl Storage {
     /// backend, so the next run resolves locally instead of round-tripping to
     /// the remote.
     async fn warm_local_backends(&self, digest: &Digest, manifest: &Manifest) {
-        let mut background_tasks = self.background_tasks.lock().unwrap();
+        let mut background_tasks = self.background_tasks.lock().await;
 
         for backend in self.get_local_backends() {
             if !backend.is_writable() {
@@ -432,7 +443,7 @@ impl Storage {
         let background_tasks = {
             self.background_tasks
                 .lock()
-                .unwrap()
+                .await
                 .drain(0..)
                 .collect::<Vec<_>>()
         };

crates/cache-storage/src/storage.rs

@@ -1,11 +1,11 @@
 use crate::capabilities::CacheCapabilities;
 use crate::helpers::{Batch, create_batches};
-use crate::manifest::Manifest;
 use async_trait::async_trait;
 use miette::IntoDiagnostic;
 use moon_blob::{BlobCleanStats, BlobInput, BlobOutput};
 use moon_common::{Id, format_error_chain};
 use moon_hash::Digest;
+use moon_manifest::Manifest;
 use moon_process::ProcessRegistry;
 use rustc_hash::FxHashSet;
 use std::fmt::Debug;

crates/cache-storage/src/storage_backend.rs

@@ -3,8 +3,8 @@ use bazel_remote_apis::build::bazel::remote::execution::v2::{
 };
 use bazel_remote_apis::google::protobuf::Timestamp;
 use moon_blob::{BlobContent, Bytes};
-use moon_cache_storage::{Manifest, ManifestFile};
 use moon_hash::{ContentHash, Digest};
+use moon_manifest::{Manifest, ManifestFile};
 use rustc_hash::FxHashMap;
 use starbase_sandbox::create_empty_sandbox;
 use starbase_utils::json::serde_json;

crates/cache-storage/tests/manifest_test.rs

@@ -1,12 +1,12 @@
 use async_trait::async_trait;
 use moon_blob::{BlobContent, BlobInput, BlobOutput, Bytes};
 use moon_cache_storage::{
-    CacheCapabilities, CacheContext, Manifest, ManifestFile, Storage, StorageBackend,
-    StorageOptions,
+    CacheCapabilities, CacheContext, Storage, StorageBackend, StorageOptions,
 };
 use moon_common::Id;
 use moon_config::{CacheConfig, RemoteConfig};
 use moon_hash::{ContentHash, Digest};
+use moon_manifest::{Manifest, ManifestFile};
 use rustc_hash::FxHashMap;
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};

crates/cache-storage/tests/storage_test.rs

@@ -5,11 +5,11 @@ edition = "2024"
 publish = false
 
 [dependencies]
-moon_cache_storage = { path = "../cache-storage" }
 moon_common = { path = "../common" }
 moon_daemon_proto = { path = "../daemon-proto" }
 moon_daemon_utils = { path = "../daemon-utils" }
 moon_hash = { path = "../hash" }
+moon_manifest = { path = "../manifest" }
 hyper-util = "0.1.20"
 miette = { workspace = true }
 thiserror = { workspace = true }