Merge pull request #3340 from itowlson/watchexec-aargh

Upgrade watchexec to 8.x and other audit adventures

Author: itowlson

Cargo.lock

@@ -22,7 +22,7 @@ bytes = { workspace = true }
 clap = { workspace = true, features = ["deprecated", "derive", "env"] }
 clearscreen = "4"
 comfy-table = "7"
-command-group = "2"
+command-group = { version = "5", features = ["with-tokio"] }
 ctrlc = { workspace = true }
 dialoguer = { workspace = true }
 futures = { workspace = true }
@@ -49,8 +49,9 @@ toml = { workspace = true }
 tracing = { workspace = true }
 url = { workspace = true }
 uuid = { version = "1.0", features = ["v4"] }
-watchexec = { git = "https://github.com/watchexec/watchexec.git", rev = "8e91d26ef6400c1e60b32a8314cbb144fa33f288" }
-watchexec-filterer-globset = { git = "https://github.com/watchexec/watchexec.git", rev = "8e91d26ef6400c1e60b32a8314cbb144fa33f288" }
+watchexec = "8.0"
+watchexec-events = "6.0"
+watchexec-filterer-globset = "8.0"
 
 spin-app = { path = "crates/app" }
 spin-build = { path = "crates/build" }
@@ -170,7 +171,7 @@ toml = "0.8"
 toml_edit = "0.22"
 tower-service = "0.3.3"
 tracing = { version = "0.1.41", features = ["log"] }
-url = "2"
+url = "2.5.7"
 walkdir = "2"
 wasm-encoder = "0.239.0"
 wasm-metadata = "0.239.0"

Cargo.toml

@@ -7,14 +7,14 @@ edition = { workspace = true }
 [dependencies]
 anyhow = { workspace = true }
 async-compression = { version = "0.4", features = ["gzip", "tokio"] }
-async-tar = "0.5"
 base64 = { workspace = true }
 chrono = { workspace = true }
 # Fork with updated auth to support ACR login
 # Ref https://github.com/camallo/dkregistry-rs/pull/263
 dirs = { workspace = true }
 dkregistry = { git = "https://github.com/fermyon/dkregistry-rs", rev = "161cf2b66996ed97c7abaf046e38244484814de3" }
 docker_credential = "1"
+flate2 = { workspace = true }
 futures-util = { workspace = true }
 itertools = { workspace = true }
 oci-distribution = { git = "https://github.com/fermyon/oci-distribution", rev = "7b291a39f74d1a3c9499d934a56cae6580fc8e37" }
@@ -25,6 +25,7 @@ spin-common = { path = "../common" }
 spin-compose = { path = "../compose" }
 spin-loader = { path = "../loader" }
 spin-locked-app = { path = "../locked-app" }
+tar = "0.4"
 tempfile = { workspace = true }
 tokio = { workspace = true, features = ["fs"] }
 tokio-util = { version = "0.7", features = ["compat"] }

crates/oci/Cargo.toml

@@ -447,7 +447,7 @@ impl Client {
         // Only add the archive layer to the OCI manifest
         tracing::trace!("Adding archive layer for all files in source {:?}", &source);
         let working_dir = tempfile::tempdir()?;
-        let archive_path = crate::utils::archive(source, &working_dir.into_path())
+        let archive_path = crate::utils::archive(source, &working_dir.keep())
             .await
             .context(format!(
                 "Unable to create compressed archive for source {source:?}"

crates/oci/src/client.rs

@@ -1,61 +1,59 @@
 //! Utilities related to distributing Spin apps via OCI registries
 
 use anyhow::{Context, Result};
-use async_compression::tokio::bufread::GzipDecoder;
-use async_compression::tokio::write::GzipEncoder;
-use async_tar::Archive;
+use flate2::read::GzDecoder;
+use flate2::write::GzEncoder;
 use spin_common::ui::quoted_path;
 use std::path::{Path, PathBuf};
+use tar::Archive;
 
 /// Create a compressed archive of source, returning its path in working_dir
 pub async fn archive(source: &Path, working_dir: &Path) -> Result<PathBuf> {
-    // Create tar archive file
-    let tar_gz_path = working_dir
-        .join(source.file_name().unwrap())
-        .with_extension("tar.gz");
-    let tar_gz = tokio::fs::File::create(tar_gz_path.as_path())
-        .await
-        .context(format!(
+    let source = source.to_owned();
+    let working_dir = working_dir.to_owned();
+
+    tokio::task::spawn_blocking(move || {
+        // Create tar archive file
+        let tar_gz_path = working_dir
+            .join(source.file_name().unwrap())
+            .with_extension("tar.gz");
+        let tar_gz = std::fs::File::create(tar_gz_path.as_path()).context(format!(
             "Unable to create tar archive for source {}",
-            quoted_path(source)
+            quoted_path(&source)
         ))?;
 
-    // Create encoder
-    // TODO: use zstd? May be more performant
-    let tar_gz_enc = GzipEncoder::new(tar_gz);
-
-    // Build tar archive
-    let mut tar_builder = async_tar::Builder::new(
-        tokio_util::compat::TokioAsyncWriteCompatExt::compat_write(tar_gz_enc),
-    );
-    tar_builder
-        .append_dir_all(".", source)
-        .await
-        .context(format!(
+        // Create encoder
+        // TODO: use zstd? May be more performant
+        let tar_gz_enc = GzEncoder::new(tar_gz, flate2::Compression::default());
+
+        // Build tar archive
+        let mut tar_builder = tar::Builder::new(tar_gz_enc);
+        tar_builder.append_dir_all(".", &source).context(format!(
             "Unable to create tar archive for source {}",
-            quoted_path(source)
+            quoted_path(&source)
         ))?;
-    // Finish writing the archive
-    tar_builder.finish().await?;
-    // Shutdown the encoder
-    use tokio::io::AsyncWriteExt;
-    tar_builder
-        .into_inner()
-        .await?
-        .into_inner()
-        .shutdown()
-        .await?;
-    Ok(tar_gz_path)
+
+        // Finish writing the archive and shut down the encoder.
+        let inner_enc = tar_builder.into_inner()?;
+        inner_enc.finish()?;
+
+        Ok(tar_gz_path)
+    })
+    .await?
 }
 
 /// Unpack a compressed archive existing at source into dest
 pub async fn unarchive(source: &Path, dest: &Path) -> Result<()> {
-    let decoder = GzipDecoder::new(tokio::io::BufReader::new(
-        tokio::fs::File::open(source).await?,
-    ));
-    let archive = Archive::new(tokio_util::compat::TokioAsyncReadCompatExt::compat(decoder));
-    if let Err(e) = archive.unpack(dest).await {
-        return Err(e.into());
-    };
-    Ok(())
+    let source = source.to_owned();
+    let dest = dest.to_owned();
+
+    tokio::task::spawn_blocking(move || {
+        let decoder = GzDecoder::new(std::fs::File::open(&source)?);
+        let mut archive = Archive::new(decoder);
+        if let Err(e) = archive.unpack(&dest) {
+            return Err(e.into());
+        };
+        Ok(())
+    })
+    .await?
 }

crates/oci/src/utils.rs

@@ -279,7 +279,7 @@ pub struct RuntimeConfigFactory {
 }
 
 impl RuntimeConfigFactory {
-    async fn build_config(&self) -> anyhow::Result<watchexec::config::RuntimeConfig> {
+    async fn build_config(&self, rt: &watchexec::Config) -> anyhow::Result<()> {
         let manifest_str = tokio::fs::read_to_string(&self.manifest_file).await?;
         let manifest = spin_manifest::manifest_from_str(&manifest_str)?;
         let filterer = self
@@ -289,12 +289,11 @@ impl RuntimeConfigFactory {
 
         let handler = NotifyOnFileChange::new(self.notifier.clone(), self.impact_description);
 
-        let mut rt = watchexec::config::RuntimeConfig::default();
-        rt.pathset([&self.manifest_dir]);
+        rt.pathset([self.manifest_dir.as_path()]);
         rt.filterer(filterer);
-        rt.action_throttle(self.debounce);
-        rt.on_action(handler);
-        Ok(rt)
+        rt.throttle(self.debounce);
+        rt.on_action(move |ah| handler.handle(ah));
+        Ok(())
     }
 }
 
@@ -320,13 +319,8 @@ impl NotifyOnFileChange {
             impact_description,
         }
     }
-}
 
-impl watchexec::handler::Handler<watchexec::action::Action> for NotifyOnFileChange {
-    fn handle(
-        &mut self,
-        action: watchexec::action::Action,
-    ) -> std::result::Result<(), Box<dyn std::error::Error>> {
+    fn handle(&self, action: watchexec::action::ActionHandler) -> watchexec::action::ActionHandler {
         if self.despurifier.all_spurious(&action) {
             tracing::debug!("spin watch ignored spurious changes: {}", paths_of(&action));
         } else {
@@ -337,12 +331,12 @@ impl watchexec::handler::Handler<watchexec::action::Action> for NotifyOnFileChan
             );
             _ = self.notifier.send(Uuid::new_v4());
         }
-        action.outcome(watchexec::action::Outcome::DoNothing);
-        Ok::<(), Box<dyn std::error::Error + 'static>>(())
+
+        action
     }
 }
 
-fn paths_of(action: &watchexec::action::Action) -> String {
+fn paths_of(action: &watchexec::action::ActionHandler) -> String {
     action
         .events
         .iter()
@@ -353,13 +347,13 @@ fn paths_of(action: &watchexec::action::Action) -> String {
         .join(", ")
 }
 
-fn path_of_event(event: &watchexec::event::Event) -> Option<&Path> {
+fn path_of_event(event: &watchexec_events::Event) -> Option<&Path> {
     event.tags.iter().filter_map(path_of_tag).next()
 }
 
-fn path_of_tag(tag: &watchexec::event::Tag) -> Option<&Path> {
+fn path_of_tag(tag: &watchexec_events::Tag) -> Option<&Path> {
     match tag {
-        watchexec::event::Tag::Path { path, .. } => Some(path),
+        watchexec_events::Tag::Path { path, .. } => Some(path),
         _ => None,
     }
 }
@@ -369,7 +363,7 @@ mod despurifier {
     use std::collections::HashMap;
     use std::sync::Mutex;
     use std::time::SystemTime;
-    use watchexec::event::{filekind::FileEventKind, Tag};
+    use watchexec_events::{filekind::FileEventKind, Tag};
 
     pub struct Despurifier {
         process_start_time: SystemTime,
@@ -384,13 +378,13 @@ mod despurifier {
             }
         }
 
-        pub fn all_spurious(&mut self, action: &watchexec::action::Action) -> bool {
+        pub fn all_spurious(&self, action: &watchexec::action::ActionHandler) -> bool {
             action.events.iter().all(|e| self.all_spurious_evt(e))
         }
 
         // This is necessary to check due to a bug on macOS emitting modify events on copies
         // https://github.com/rust-lang/rust/issues/107130
-        fn all_spurious_evt(&mut self, event: &watchexec::event::Event) -> bool {
+        fn all_spurious_evt(&self, event: &watchexec_events::Event) -> bool {
             // Deletions are never spurious
             if event
                 .tags
@@ -438,7 +432,7 @@ mod despurifier {
             Self
         }
 
-        pub fn all_spurious(&mut self, _action: &watchexec::action::Action) -> bool {
+        pub fn all_spurious(&self, _action: &watchexec::action::ActionHandler) -> bool {
             false
         }
     }
@@ -459,8 +453,8 @@ impl ReconfigurableWatcher {
     async fn start(
         rtf: RuntimeConfigFactory,
     ) -> anyhow::Result<(Self, tokio::task::JoinHandle<()>)> {
-        let rt = rtf.build_config().await?;
-        let watcher = Watchexec::new(watchexec::config::InitConfig::default(), rt)?;
+        let watcher = Arc::new(Watchexec::default());
+        rtf.build_config(&watcher.config).await?;
         let watcher_clone = watcher.clone();
         let join_handle = tokio::task::spawn(async move {
             _ = watcher_clone.main().await;
@@ -477,16 +471,9 @@ impl ReconfigurableWatcher {
     pub async fn reconfigure(&self) {
         match self {
             Self::Actual((watchexec, rtf)) => {
-                let rt = match rtf.build_config().await {
-                    Ok(rt) => rt,
-                    Err(e) => {
-                        tracing::error!("Unable to re-configure watcher after manifest change. Changes in files newly added to the application may not be detected. Error: {e}");
-                        return;
-                    }
-                };
-                if let Err(e) = watchexec.reconfigure(rt) {
+                if let Err(e) = rtf.build_config(&watchexec.config).await {
                     tracing::error!("Unable to re-configure watcher after manifest change. Changes in files newly added to the application may not be detected. Error: {e}");
-                }
+                };
             }
             Self::Dummy => (),
         }

src/commands/watch.rs

@@ -1,4 +1,4 @@
-use command_group::AsyncCommandGroup;
+use command_group::tokio::AsyncCommandGroup;
 use std::path::PathBuf;
 use uuid::Uuid;
 
@@ -59,7 +59,7 @@ impl Buildifier {
                 }
                 _ = self.watched_changes.changed() => {
                     tracing::debug!("Cancelling build as there are new changes to process");
-                    child.kill()?;
+                    child.kill().await?;
                     if self.clear_screen {
                         _ = clearscreen::clear();
                     }