internal: Integrate new remote storage. (#2587)

* Add crate.

* Start on http.

* Add cache context.

* Remove old code.

* Move more code over.

* Polish.

* Fix tests.

* Tier 2.

* Tier 3.

* Tier 4.

* Polish.

* Fix format.

* Polish.

* Add tests.

Author: milesj

CHANGELOG.md

@@ -11,6 +11,11 @@
   - Improved our "stream and capture output" child process handling to operate on bytes instead of
     lines, which should resolve some edge cases with output not being written to the console, or
     being written out of order.
+- **Remote cache**
+  - Added compression support for streamed read/writes of blobs (large files).
+  - Added extensive testing to account for edge cases.
+  - When the server doesn't support the configured compression, it will now default to "identity"
+    (uncompressed) instead of disabling the cache entirely.
 - **VCS**
   - Hardened all executed Git commands: revisions are validated against argument injection,
     credential prompts now fail immediately instead of hanging, and the fsmonitor daemon is now

Cargo.lock

@@ -21,24 +21,32 @@ clap = { version = "4.6.1", default-features = false, features = [
   "error-context",
 ] }
 clap_complete = "4.6.5"
-compact_str = { version = "0.9.1", default-features = false, features = ["serde"] }
+compact_str = { version = "0.9.1", default-features = false, features = [
+  "serde",
+] }
 convert_case = "0.11.0"
 criterion = { package = "codspeed-criterion-compat", version = "4.7.0", default-features = false, features = [
   "async_tokio",
 ] }
 daggy = { version = "0.9.0", features = ["serde-1"] }
 dirs = "6.0.0"
 futures = "0.3.31"
+httpmock = "0.8.3"
 indexmap = "2.13.0"
 iocraft = "0.8.3"
 libc = "0.2.186"
 md5 = "0.8.0"
 miette = "7.6.0"
 num_cpus = "1.17.0"
 pathdiff = "0.2.3"
-petgraph = { version = "0.8.3", default-features = false, features = ["serde-1"] }
+petgraph = { version = "0.8.3", default-features = false, features = [
+  "serde-1",
+] }
 relative-path = { version = "2.0.1" }
-regex = { version = "1.12.3", default-features = false, features = ["std", "perf"] }
+regex = { version = "1.12.3", default-features = false, features = [
+  "std",
+  "perf",
+] }
 reqwest = { version = "0.13.4", default-features = false, features = [
   "rustls",
   # We don't use openssl but its required for musl builds
@@ -47,7 +55,9 @@ reqwest = { version = "0.13.4", default-features = false, features = [
 once_cell = "1.21.4"
 rustc-hash = "2.1.1"
 scc = "3.7.1"
-schematic = { version = "0.19.7", default-features = false, features = ["schema"] }
+schematic = { version = "0.19.7", default-features = false, features = [
+  "schema",
+] }
 serial_test = "3.4.0"
 semver = "1.0.27"
 serde = { version = "1.0.228", features = ["derive"] }

Cargo.toml

@@ -27,6 +27,8 @@ pub async fn sync_workspace(
     // Connect to the remote service in this action,
     // as it always runs before tasks, and we don't need it
     // for non-pipeline related features!
+    app_context.cache_engine.storage.connect_backends().await?;
+
     if app_context.workspace_config.remote.is_enabled() {
         RemoteService::connect(
             &app_context.workspace_config.remote,

crates/actions/src/actions/sync_workspace.rs

@@ -15,6 +15,8 @@ moon_api = { path = "../api" }
 moon_app_context = { path = "../app-context" }
 moon_app_macros = { path = "../app-macros" }
 moon_cache = { path = "../cache" }
+moon_cache_local = { path = "../cache-local" }
+moon_cache_remote = { path = "../cache-remote" }
 moon_codegen = { path = "../codegen" }
 moon_common = { path = "../common" }
 moon_config = { path = "../config" }
@@ -70,14 +72,23 @@ proto_core = { workspace = true }
 regex = { workspace = true }
 reqwest = { workspace = true }
 rustc-hash = { workspace = true }
-schematic = { workspace = true, features = ["config", "schema", "renderer_template"] }
+schematic = { workspace = true, features = [
+  "config",
+  "schema",
+  "renderer_template",
+] }
 semver = { workspace = true }
 serde = { workspace = true }
 starbase = { workspace = true }
 starbase_archive = { workspace = true }
 starbase_shell = { workspace = true }
 starbase_styles = { workspace = true }
-starbase_utils = { workspace = true, features = ["editor-config", "json", "toml", "yaml"] }
+starbase_utils = { workspace = true, features = [
+  "editor-config",
+  "json",
+  "toml",
+  "yaml",
+] }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }

crates/app/Cargo.toml

@@ -5,14 +5,19 @@ use async_trait::async_trait;
 use moon_action_graph::{ActionGraphBuilder, ActionGraphBuilderOptions};
 use moon_api::Launchpad;
 use moon_app_context::AppContext;
-use moon_cache::CacheEngine;
+use moon_cache::{CacheContext, CacheEngine};
+use moon_cache_local::LocalStorage;
+use moon_cache_remote::{GrpcRemoteStorage, HttpRemoteStorage};
 use moon_codegen::CodeGenerator;
 use moon_common::{is_docker, is_formatted_output, is_remote, is_test_env};
-use moon_config::{ExtensionsConfig, InheritedTasksManager, ToolchainsConfig, WorkspaceConfig};
+use moon_config::{
+    ExtensionsConfig, InheritedTasksManager, RemoteApi, ToolchainsConfig, WorkspaceConfig,
+};
 use moon_config_loader::ConfigLoader;
 use moon_console::{Console, MoonReporter, create_console_theme};
 use moon_daemon::{DaemonClient, DaemonConnector};
 use moon_env::MoonEnvironment;
+use moon_env_var::GlobalEnvBag;
 use moon_extension_plugin::*;
 use moon_plugin::MoonHostData;
 use moon_process::ProcessRegistry;
@@ -196,10 +201,39 @@ impl MoonSession {
 
     pub fn get_cache_engine(&self) -> miette::Result<Arc<CacheEngine>> {
         if self.cache_engine.get().is_none() {
-            let _ = self.cache_engine.set(Arc::new(CacheEngine::new(
-                &self.config_dir,
-                &self.workspace_config.cache,
-            )?));
+            let context = CacheContext {
+                cache_dir: self.config_dir.join("cache"),
+                cache_config: Arc::new(self.workspace_config.cache.clone()),
+                config_dir: self.config_dir.clone(),
+                remote_config: Arc::new(self.workspace_config.remote.clone()),
+                remote_debug: GlobalEnvBag::instance().should_debug_remote(),
+                workspace_root: self.workspace_root.clone(),
+            };
+
+            let mut engine = CacheEngine::new(context.clone())?;
+
+            engine.storage.add_local_backend(LocalStorage::new(
+                context.clone(),
+                &context.cache_dir,
+                false,
+            )?);
+
+            if context.remote_config.is_enabled() {
+                match context.remote_config.api {
+                    RemoteApi::Grpc => {
+                        engine
+                            .storage
+                            .add_remote_backend(GrpcRemoteStorage::new(context.clone())?);
+                    }
+                    RemoteApi::Http => {
+                        engine
+                            .storage
+                            .add_remote_backend(HttpRemoteStorage::new(context.clone())?);
+                    }
+                };
+            }
+
+            let _ = self.cache_engine.set(Arc::new(engine));
         }
 
         Ok(self.cache_engine.get().map(Arc::clone).unwrap())
@@ -463,6 +497,12 @@ impl AppSession for MoonSession {
             let _ = daemon.stop().await;
         }
 
+        // Ensure all in-flight storage tasks have finished
+        self.get_cache_engine()?
+            .storage
+            .wait_for_background_tasks()
+            .await?;
+
         // Ensure all child processes have finished running
         ProcessRegistry::instance()
             .wait_for_running_to_shutdown()

crates/app/src/session.rs

@@ -51,3 +51,14 @@ impl Debug for Blob {
             .finish()
     }
 }
+
+impl TryFrom<Bytes> for Blob {
+    type Error = miette::Report;
+
+    fn try_from(bytes: Bytes) -> Result<Self, Self::Error> {
+        Ok(Blob {
+            digest: Digest::from_bytes(&bytes)?,
+            bytes,
+        })
+    }
+}