Allow plugin installations to be cancelled

src-tauri/src/lib.rs

@@ -614,6 +614,7 @@ pub fn run() {
             // Plugin Registry
             plugins::commands::fetch_plugin_registry,
             plugins::commands::install_plugin,
+            plugins::commands::cancel_plugin_install,
             plugins::commands::uninstall_plugin,
             plugins::commands::get_installed_plugins,
             plugins::commands::disable_plugin,

src-tauri/src/plugins/commands.rs

@@ -1,12 +1,10 @@
 use std::fs;
-use std::time::Duration;
 
 use crate::drivers::driver_trait::PluginManifest;
 use crate::plugins::installer::{self, InstalledPluginInfo};
 use crate::plugins::manager::ConfigManifest;
 use crate::plugins::registry::{self, RegistryPlugin, RegistryPluginWithStatus, RegistryReleaseWithStatus};
 use tauri::AppHandle;
-use tokio::time::sleep;
 
 /// Resolves which Tabularium registry to talk to. Operators pin a
 /// URL via `tabularium_registry_url` in `config.json`; otherwise the
@@ -157,11 +155,9 @@ pub async fn install_plugin(
     plugin_id: String,
     version: Option<String>,
 ) -> Result<(), String> {
-    // Updating an installed plugin must stop the existing process first,
-    // otherwise the OS may keep files locked while we replace the directory.
-    crate::drivers::registry::unregister_driver(&plugin_id).await;
-    crate::drivers::registry::unregister_manifest(&plugin_id).await;
-    sleep(Duration::from_millis(500)).await;
+    let install_guard = crate::plugins::install_cancellation::begin(&plugin_id)?;
+    let cancellation = install_guard.cancellation();
+    cancellation.check()?;
 
     let config = crate::config::load_config_internal(&app);
     let platform = registry::get_current_platform();
@@ -206,11 +202,13 @@ pub async fn install_plugin(
     // happens inside download_and_install, while the bundle is still in its
     // temp dir — a mismatching archive is discarded without ever touching an
     // existing installation.
+    cancellation.check()?;
     installer::download_and_install(
         &plugin_id,
         &download_url,
         expected_sha256.as_deref(),
         Some(&target_version),
+        cancellation,
     )
     .await?;
 
@@ -227,6 +225,11 @@ pub async fn install_plugin(
     Ok(())
 }
 
+#[tauri::command]
+pub fn cancel_plugin_install(plugin_id: String) -> bool {
+    crate::plugins::install_cancellation::cancel(&plugin_id)
+}
+
 #[tauri::command]
 pub async fn uninstall_plugin(plugin_id: String) -> Result<(), String> {
     // Unregister from in-memory driver registry first

src-tauri/src/plugins/install_cancellation.rs

@@ -0,0 +1,116 @@
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+
+use once_cell::sync::Lazy;
+use tokio::sync::Notify;
+
+pub const INSTALL_CANCELLED_ERROR: &str = "PLUGIN_INSTALL_CANCELLED";
+
+struct CancellationInner {
+    cancelled: AtomicBool,
+    notify: Notify,
+}
+
+#[derive(Clone)]
+pub struct InstallCancellation {
+    inner: Arc<CancellationInner>,
+}
+
+impl InstallCancellation {
+    fn new() -> Self {
+        Self {
+            inner: Arc::new(CancellationInner {
+                cancelled: AtomicBool::new(false),
+                notify: Notify::new(),
+            }),
+        }
+    }
+
+    pub fn is_cancelled(&self) -> bool {
+        self.inner.cancelled.load(Ordering::Acquire)
+    }
+
+    pub fn check(&self) -> Result<(), String> {
+        if self.is_cancelled() {
+            Err(INSTALL_CANCELLED_ERROR.to_string())
+        } else {
+            Ok(())
+        }
+    }
+
+    pub async fn cancelled(&self) {
+        if self.is_cancelled() {
+            return;
+        }
+
+        let notified = self.inner.notify.notified();
+        if self.is_cancelled() {
+            return;
+        }
+        notified.await;
+    }
+
+    fn cancel(&self) {
+        if !self.inner.cancelled.swap(true, Ordering::AcqRel) {
+            self.inner.notify.notify_waiters();
+        }
+    }
+}
+
+static ACTIVE_INSTALLS: Lazy<Mutex<HashMap<String, InstallCancellation>>> =
+    Lazy::new(|| Mutex::new(HashMap::new()));
+
+pub struct InstallGuard {
+    plugin_id: String,
+    cancellation: InstallCancellation,
+}
+
+impl InstallGuard {
+    pub fn cancellation(&self) -> &InstallCancellation {
+        &self.cancellation
+    }
+}
+
+impl Drop for InstallGuard {
+    fn drop(&mut self) {
+        let mut installs = ACTIVE_INSTALLS
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        installs.remove(&self.plugin_id);
+    }
+}
+
+pub fn begin(plugin_id: &str) -> Result<InstallGuard, String> {
+    let mut installs = ACTIVE_INSTALLS
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    if installs.contains_key(plugin_id) {
+        return Err(format!(
+            "An installation is already running for plugin '{}'",
+            plugin_id
+        ));
+    }
+
+    let cancellation = InstallCancellation::new();
+    installs.insert(plugin_id.to_string(), cancellation.clone());
+    Ok(InstallGuard {
+        plugin_id: plugin_id.to_string(),
+        cancellation,
+    })
+}
+
+pub fn cancel(plugin_id: &str) -> bool {
+    let cancellation = ACTIVE_INSTALLS
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+        .get(plugin_id)
+        .cloned();
+
+    if let Some(cancellation) = cancellation {
+        cancellation.cancel();
+        true
+    } else {
+        false
+    }
+}

src-tauri/src/plugins/installer.rs

@@ -1,10 +1,14 @@
 use std::fs;
 use std::io::Read;
 use std::path::{Path, PathBuf};
+use std::time::Duration;
 
 use directories::ProjectDirs;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
+use tokio::time::sleep;
+
+use super::install_cancellation::{InstallCancellation, INSTALL_CANCELLED_ERROR};
 
 #[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct InstalledPluginInfo {
@@ -158,7 +162,9 @@ pub async fn download_and_install(
     download_url: &str,
     expected_sha256: Option<&str>,
     expected_version: Option<&str>,
+    cancellation: &InstallCancellation,
 ) -> Result<(), String> {
+    cancellation.check()?;
     let plugins_dir = get_plugins_dir()?;
     let tmp_dir = plugins_dir.join(format!(".tmp-{}", plugin_id));
     let final_dir = plugins_dir.join(plugin_id);
@@ -171,9 +177,13 @@ pub async fn download_and_install(
 
     // Download ZIP to memory
     log::info!("Downloading plugin '{}' from: {}", plugin_id, download_url);
-    let response = reqwest::get(download_url)
-        .await
-        .map_err(|e| format!("Failed to download plugin: {}", e))?;
+    let response = tokio::select! {
+        biased;
+        _ = cancellation.cancelled() => return Err(INSTALL_CANCELLED_ERROR.to_string()),
+        result = reqwest::get(download_url) => {
+            result.map_err(|e| format!("Failed to download plugin: {}", e))?
+        }
+    };
 
     let status = response.status();
     let content_type = response
@@ -204,10 +214,13 @@ pub async fn download_and_install(
         ));
     }
 
-    let bytes = response
-        .bytes()
-        .await
-        .map_err(|e| format!("Failed to read plugin download: {}", e))?;
+    let bytes = tokio::select! {
+        biased;
+        _ = cancellation.cancelled() => return Err(INSTALL_CANCELLED_ERROR.to_string()),
+        result = response.bytes() => {
+            result.map_err(|e| format!("Failed to read plugin download: {}", e))?
+        }
+    };
 
     log::info!(
         "Plugin '{}' downloaded {} bytes (content-type: {})",
@@ -241,6 +254,8 @@ pub async fn download_and_install(
         log::info!("Plugin '{}' SHA-256 verified ({})", plugin_id, actual);
     }
 
+    cancellation.check()?;
+
     // Extract to temp dir
     fs::create_dir_all(&tmp_dir).map_err(|e| format!("Failed to create temp directory: {}", e))?;
 
@@ -262,6 +277,12 @@ pub async fn download_and_install(
     })?;
 
     for i in 0..archive.len() {
+        if cancellation.is_cancelled() {
+            drop(archive);
+            fs::remove_dir_all(&tmp_dir).ok();
+            return Err(INSTALL_CANCELLED_ERROR.to_string());
+        }
+
         let mut file = archive
             .by_index(i)
             .map_err(|e| format!("Failed to read ZIP entry: {}", e))?;
@@ -282,8 +303,22 @@ pub async fn download_and_install(
                 }
             }
             let mut buf = Vec::new();
-            file.read_to_end(&mut buf)
-                .map_err(|e| format!("Failed to read ZIP file content: {}", e))?;
+            let mut chunk = [0_u8; 64 * 1024];
+            while !cancellation.is_cancelled() {
+                let read = file
+                    .read(&mut chunk)
+                    .map_err(|e| format!("Failed to read ZIP file content: {}", e))?;
+                if read == 0 {
+                    break;
+                }
+                buf.extend_from_slice(&chunk[..read]);
+            }
+            if cancellation.is_cancelled() {
+                drop(file);
+                drop(archive);
+                fs::remove_dir_all(&tmp_dir).ok();
+                return Err(INSTALL_CANCELLED_ERROR.to_string());
+            }
             fs::write(&out_path, &buf).map_err(|e| format!("Failed to write file: {}", e))?;
 
             // Set executable permissions on Unix
@@ -334,8 +369,21 @@ pub async fn download_and_install(
         }
     }
 
-    // Remove existing plugin dir if present
+    // Cancellation remains safe until this commit point: the existing plugin
+    // is still registered and its files have not been touched.
+    if cancellation.is_cancelled() {
+        fs::remove_dir_all(&tmp_dir).ok();
+        return Err(INSTALL_CANCELLED_ERROR.to_string());
+    }
+
+    // Updating an installed plugin must stop its process immediately before
+    // replacing files, otherwise the OS may keep them locked. Once this short
+    // commit phase starts, installation is completed atomically rather than
+    // leaving the existing plugin disabled.
     if final_dir.exists() {
+        crate::drivers::registry::unregister_driver(plugin_id).await;
+        crate::drivers::registry::unregister_manifest(plugin_id).await;
+        sleep(Duration::from_millis(500)).await;
         fs::remove_dir_all(&final_dir)
             .map_err(|e| format!("Failed to remove existing plugin: {}", e))?;
     }

src-tauri/src/plugins/mod.rs

@@ -2,6 +2,7 @@ pub mod commands;
 pub mod compat; // COMPAT(registry-ga): remove with the BC layer
 pub mod deep_link;
 pub mod driver;
+pub mod install_cancellation;
 pub mod installer;
 pub mod integrity;
 pub mod manager;

src-tauri/src/plugins/tests.rs

@@ -2,9 +2,37 @@ use std::fs;
 
 use tempfile::tempdir;
 
+use super::install_cancellation::{begin, cancel, INSTALL_CANCELLED_ERROR};
 use super::installer::{has_manifest, migrate_plugins_between, read_plugin_info_from_dir};
 use super::manager::ConfigManifest;
 
+#[test]
+fn cancellation_marks_an_active_install() {
+    let guard = begin("cancellation-test-plugin").expect("begin install");
+
+    assert!(!guard.cancellation().is_cancelled());
+    assert!(cancel("cancellation-test-plugin"));
+    assert_eq!(
+        guard.cancellation().check().expect_err("cancelled install"),
+        INSTALL_CANCELLED_ERROR
+    );
+
+    drop(guard);
+    assert!(!cancel("cancellation-test-plugin"));
+}
+
+#[test]
+fn duplicate_install_is_rejected_until_guard_is_dropped() {
+    let guard = begin("duplicate-install-test-plugin").expect("begin install");
+    let error = begin("duplicate-install-test-plugin")
+        .err()
+        .expect("duplicate install must fail");
+    assert!(error.contains("already running"));
+
+    drop(guard);
+    assert!(begin("duplicate-install-test-plugin").is_ok());
+}
+
 #[test]
 fn reads_canonical_tabularium_manifest() {
     // The canonical bundle ships `.tabularium` (JSON content). It drops `id`