fix(plugins): block updates while connections are active

crates/dbx-core/src/connection.rs

@@ -1560,7 +1560,7 @@ impl AppState {
 
     pub async fn external_driver_pool(&self, driver_id: &str, config: &ConnectionConfig) -> Result<PoolKind, String> {
         let env = self.external_driver_runtime_env(driver_id)?;
-        let session = self.plugins.start_driver_session_with_env(driver_id, env).await?;
+        let session = self.plugins.start_driver_session_for_connection(driver_id, env, &config.name).await?;
         let params = serde_json::json!({ "connection": config });
         session
             .invoke_with_timeout::<serde_json::Value>("connect", params, Some(external_driver_connect_timeout(config)))

crates/dbx-core/src/plugins.rs

@@ -10,6 +10,7 @@ mod assets;
 mod filesystem;
 mod host;
 mod installer;
+mod lifecycle;
 mod manifest;
 mod marketplace;
 mod runtime;
@@ -28,6 +29,7 @@ pub use installer::{
     PluginRollbackResult, PluginSignatureStatus, PluginTrustStore, PluginTrustedKey, DBXP_EXTENSION,
     MAX_PLUGIN_PACKAGE_BYTES, PLUGIN_CHECKSUMS_FILE, PLUGIN_SIGNATURE_FILE,
 };
+pub use lifecycle::PluginLifecycle;
 pub use marketplace::{
     url_install_trust_store, PluginMarketplace, PluginMarketplaceArtifact, PluginMarketplaceCatalog,
     PluginMarketplaceInstallRequest, PluginMarketplaceLocalization, PluginMarketplacePlugin,
@@ -139,6 +141,7 @@ impl PluginRuntimeEnv {
 pub struct PluginRegistry {
     root_dir: PathBuf,
     app_version: String,
+    lifecycle: PluginLifecycle,
 }
 
 impl PluginRegistry {
@@ -147,7 +150,11 @@ impl PluginRegistry {
     }
 
     pub fn new_with_app_version(root_dir: PathBuf, app_version: impl Into<String>) -> Self {
-        Self { root_dir, app_version: app_version.into() }
+        Self { root_dir, app_version: app_version.into(), lifecycle: PluginLifecycle::default() }
+    }
+
+    pub fn lifecycle(&self) -> PluginLifecycle {
+        self.lifecycle.clone()
     }
 
     pub fn root_dir(&self) -> &Path {
@@ -282,12 +289,30 @@ impl PluginRegistry {
         &self,
         driver_id: &str,
         env: PluginRuntimeEnv,
+    ) -> Result<Arc<PluginDriverSession>, String> {
+        self.start_driver_session_for_connection(driver_id, env, driver_id).await
+    }
+
+    pub async fn start_driver_session_for_connection(
+        &self,
+        driver_id: &str,
+        env: PluginRuntimeEnv,
+        connection_label: &str,
     ) -> Result<Arc<PluginDriverSession>, String> {
         let plugin =
             self.find_driver(driver_id)?.ok_or_else(|| format!("Plugin driver '{driver_id}' is not installed"))?;
+        let activity = self.lifecycle.begin_connection(&plugin.manifest.id, connection_label)?;
+        let current =
+            self.find_driver(driver_id)?.ok_or_else(|| format!("Plugin driver '{driver_id}' is not installed"))?;
+        if current.manifest.id != plugin.manifest.id {
+            return Err(format!("Plugin driver '{driver_id}' changed. Please reconnect."));
+        }
+        let plugin = current;
         ensure_plugin_compatible(&plugin)?;
         let env = env.with_plugin_data_dir(&self.plugin_data_dir(&plugin.manifest.id));
-        PluginDriverSession::start(plugin, driver_id.to_string(), self.app_version.clone(), env).await.map(Arc::new)
+        PluginDriverSession::start(plugin, driver_id.to_string(), self.app_version.clone(), env, activity)
+            .await
+            .map(Arc::new)
     }
 }
 
@@ -326,6 +351,7 @@ fn ensure_plugin_compatible(plugin: &InstalledPlugin) -> Result<(), String> {
 pub struct PluginDriverSession {
     sidecar: Arc<PluginSidecarSession>,
     driver_id: String,
+    _activity: lifecycle::PluginUsageGuard,
 }
 
 impl PluginDriverSession {
@@ -334,9 +360,10 @@ impl PluginDriverSession {
         driver_id: String,
         app_version: String,
         env: PluginRuntimeEnv,
+        activity: lifecycle::PluginUsageGuard,
     ) -> Result<Self, String> {
         let sidecar = PluginSidecarSession::start(plugin, app_version, env).await?;
-        Ok(Self { sidecar, driver_id })
+        Ok(Self { sidecar, driver_id, _activity: activity })
     }
 
     pub async fn invoke<T>(&self, method: &str, params: serde_json::Value) -> Result<T, String>
@@ -384,7 +411,8 @@ impl PluginDriverSession {
         driver_id: String,
         env: PluginRuntimeEnv,
     ) -> Result<Self, String> {
-        Self::start(plugin, driver_id, env!("CARGO_PKG_VERSION").to_string(), env).await
+        let activity = PluginLifecycle::default().begin_connection(&plugin.manifest.id, &driver_id)?;
+        Self::start(plugin, driver_id, env!("CARGO_PKG_VERSION").to_string(), env, activity).await
     }
 }
 
@@ -472,6 +500,7 @@ sleep 30
         .unwrap();
 
         let registry = PluginRegistry::new_with_app_version(data_dir.path().join("plugins"), "0.5.67");
+        let lifecycle = registry.lifecycle();
         let host = PluginHost::new(registry);
         let reported: String =
             host.invoke("sample.sidecar", "sample/dataDir", serde_json::Value::Null, None, None).await.unwrap();
@@ -480,6 +509,24 @@ sleep 30
             data_dir.path().join("plugin-data").join("sample.sidecar"),
             "sidecar must see <data dir>/plugin-data/<id> in DBX_PLUGIN_DATA_DIR"
         );
+        assert!(lifecycle.begin_update("sample.sidecar").is_ok());
+        let request_host = host.clone();
+        let pending = tokio::spawn(async move {
+            request_host
+                .invoke::<serde_json::Value>("sample.sidecar", "sample/wait", serde_json::Value::Null, None, None)
+                .await
+        });
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            while lifecycle.check_update("sample.sidecar").is_ok() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .unwrap();
+        assert!(lifecycle.begin_update("sample.sidecar").unwrap_err().contains("active operations"));
+        pending.abort();
+        assert!(pending.await.unwrap_err().is_cancelled());
+        assert!(lifecycle.begin_update("sample.sidecar").is_ok());
         host.stop_all().await;
     }
 
@@ -518,14 +565,9 @@ sleep 30
             env!("CARGO_PKG_VERSION"),
         );
 
-        let session = PluginDriverSession::start(
-            plugin,
-            "jdbc".to_string(),
-            env!("CARGO_PKG_VERSION").to_string(),
-            PluginRuntimeEnv::default(),
-        )
-        .await
-        .expect("session should start");
+        let session = PluginDriverSession::start_for_test(plugin, "jdbc".to_string(), PluginRuntimeEnv::default())
+            .await
+            .expect("session should start");
         let pid = session.pid().await.expect("child should have a pid");
 
         session.shutdown().await;

crates/dbx-core/src/plugins/host.rs

@@ -45,6 +45,7 @@ pub struct PluginConnectionHandle {
     session: Option<Arc<PluginSidecarSession>>,
     disconnect: bool,
     params: serde_json::Value,
+    _activity: super::lifecycle::PluginUsageGuard,
 }
 
 impl PluginConnectionHandle {
@@ -115,6 +116,7 @@ impl PluginHost {
         plugin_id: &str,
         env: PluginRuntimeEnv,
     ) -> Result<Arc<PluginSidecarSession>, String> {
+        let _activity = self.inner.registry.lifecycle.begin_operation(plugin_id)?;
         if let Some(session) = self.running_session(plugin_id).await {
             return Ok(session);
         }
@@ -151,6 +153,7 @@ impl PluginHost {
     where
         T: DeserializeOwned,
     {
+        let _activity = self.inner.registry.lifecycle.begin_operation(plugin_id)?;
         let session = self.activate(plugin_id).await?;
         ensure_permission(session.plugin(), required_permission)?;
         session.invoke_with_timeout(method, params, None, timeout).await
@@ -163,6 +166,7 @@ impl PluginHost {
         params: serde_json::Value,
         required_permission: Option<&str>,
     ) -> Result<(), String> {
+        let _activity = self.inner.registry.lifecycle.begin_operation(plugin_id)?;
         let session = self.activate(plugin_id).await?;
         ensure_permission(session.plugin(), required_permission)?;
         session.notify(method, params, None).await
@@ -175,6 +179,7 @@ impl PluginHost {
         data: &[u8],
         required_permission: Option<&str>,
     ) -> Result<(), String> {
+        let _activity = self.inner.registry.lifecycle.begin_operation(plugin_id)?;
         let session = self.activate(plugin_id).await?;
         ensure_permission(session.plugin(), required_permission)?;
         session.send_binary(channel, data).await
@@ -195,6 +200,11 @@ impl PluginHost {
         runtime_host: &str,
         runtime_port: u16,
     ) -> Result<ConnectionTestResult, String> {
+        let _activity = self
+            .inner
+            .registry
+            .lifecycle
+            .begin_connection(config.plugin_id.as_deref().unwrap_or_default(), &config.name)?;
         let (plugin, provider) = self.resolve_connection_provider(config)?;
         validate_plugin_connection_values(config, &provider)?;
         let provider_label = provider.label.as_deref().unwrap_or(&plugin.manifest.name);
@@ -219,6 +229,11 @@ impl PluginHost {
         runtime_host: &str,
         runtime_port: u16,
     ) -> Result<PluginConnectionHandle, String> {
+        let activity = self
+            .inner
+            .registry
+            .lifecycle
+            .begin_connection(config.plugin_id.as_deref().unwrap_or_default(), &config.name)?;
         let (_, provider) = self.resolve_connection_provider(config)?;
         validate_plugin_connection_values(config, &provider)?;
         let params = plugin_connection_params(config, &provider, runtime_host, runtime_port)?;
@@ -250,6 +265,7 @@ impl PluginHost {
             session,
             disconnect,
             params,
+            _activity: activity,
         })
     }
 
@@ -260,6 +276,11 @@ impl PluginHost {
         runtime_host: &str,
         runtime_port: u16,
     ) -> Result<PluginConnectionActionResult, String> {
+        let _activity = self
+            .inner
+            .registry
+            .lifecycle
+            .begin_connection(config.plugin_id.as_deref().unwrap_or_default(), &config.name)?;
         let (_, provider) = self.resolve_connection_provider(config)?;
         let action = plugin_invoke_connection_action(&provider, action_id)?;
         validate_plugin_connection_values_for_action(config, &provider, action.requires_valid_form)?;
@@ -738,6 +759,57 @@ mod tests {
     use crate::models::connection::ConnectionConfig;
     use crate::plugins::PluginConnectionProviderContribution;
 
+    #[tokio::test]
+    async fn ui_only_connections_hold_update_guards_but_saved_configs_do_not() {
+        let root = tempfile::tempdir().unwrap();
+        let plugin_dir = root.path().join("sample.ui");
+        std::fs::create_dir(&plugin_dir).unwrap();
+        std::fs::write(plugin_dir.join("index.html"), "<html></html>").unwrap();
+        std::fs::write(
+            plugin_dir.join("manifest.json"),
+            serde_json::json!({
+                "manifest_version": 1,
+                "id": "sample.ui",
+                "name": "Sample UI",
+                "version": "1.0.0",
+                "publisher": "sample",
+                "engines": { "dbx": ">=0.5.0", "host_api": "^1.0" },
+                "entrypoints": { "ui": { "entry": "index.html" } },
+                "contributions": [{
+                    "type": "connection-provider",
+                    "id": "sample.connection",
+                    "database_type": "sample"
+                }]
+            })
+            .to_string(),
+        )
+        .unwrap();
+        let config: ConnectionConfig = serde_json::from_value(serde_json::json!({
+            "id": "saved-connection",
+            "name": "Saved UI connection",
+            "db_type": "plugin",
+            "host": "localhost",
+            "port": 0,
+            "username": "",
+            "password": "",
+            "plugin_id": "sample.ui",
+            "plugin_connection_provider": "sample.connection",
+            "plugin_connection_type": "sample"
+        }))
+        .unwrap();
+        let registry = crate::plugins::PluginRegistry::new_with_app_version(root.path().to_path_buf(), "0.5.67");
+        let lifecycle = registry.lifecycle();
+        let host = super::PluginHost::new(registry);
+        let update = lifecycle.begin_update("sample.ui").unwrap();
+        assert!(host.connect_connection(&config, "localhost", 0).await.is_err());
+        drop(update);
+        let connection = host.connect_connection(&config, "localhost", 0).await.unwrap();
+        assert!(lifecycle.begin_update("sample.ui").unwrap_err().contains("Saved UI connection"));
+        connection.disconnect().await.unwrap();
+        drop(connection);
+        assert!(lifecycle.begin_update("sample.ui").is_ok());
+    }
+
     #[test]
     fn rejects_zero_port_for_port_bound_connection_field() {
         let config: ConnectionConfig = serde_json::from_value(serde_json::json!({

crates/dbx-core/src/plugins/installer.rs

@@ -12,7 +12,8 @@ use semver::Version;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 
-use super::{InstalledPlugin, InstalledPluginInfo, PluginManifest};
+use super::lifecycle::PluginUsageGuard;
+use super::{InstalledPlugin, InstalledPluginInfo, PluginLifecycle, PluginManifest};
 
 pub const DBXP_EXTENSION: &str = "dbxp";
 pub const PLUGIN_CHECKSUMS_FILE: &str = "checksums.json";
@@ -48,12 +49,14 @@ pub struct PluginInstallResult {
     pub previous_version: Option<String>,
     pub package_sha256: String,
     pub signature: PluginSignatureStatus,
+    _update_guard: Option<PluginUsageGuard>,
 }
 
 #[derive(Debug, Clone)]
 pub struct PluginRollbackResult {
     pub plugin: InstalledPlugin,
     pub previous_version: String,
+    _update_guard: Option<PluginUsageGuard>,
 }
 
 #[derive(Debug, Clone, Serialize)]
@@ -135,6 +138,7 @@ pub struct PluginPackageInstaller {
     root_dir: PathBuf,
     app_version: String,
     trust_store: Arc<PluginTrustStore>,
+    lifecycle: Option<PluginLifecycle>,
 }
 
 #[derive(Debug, Clone)]
@@ -256,11 +260,16 @@ fn read_trust_document(root_dir: &Path) -> Result<PluginTrustDocument, String> {
 impl PluginPackageInstaller {
     pub fn new(root_dir: PathBuf, app_version: impl Into<String>) -> Result<Self, String> {
         let trust_store = super::marketplace::package_install_trust_store(&root_dir)?;
-        Ok(Self { root_dir, app_version: app_version.into(), trust_store: Arc::new(trust_store) })
+        Ok(Self { root_dir, app_version: app_version.into(), trust_store: Arc::new(trust_store), lifecycle: None })
     }
 
     pub fn with_trust_store(root_dir: PathBuf, app_version: impl Into<String>, trust_store: PluginTrustStore) -> Self {
-        Self { root_dir, app_version: app_version.into(), trust_store: Arc::new(trust_store) }
+        Self { root_dir, app_version: app_version.into(), trust_store: Arc::new(trust_store), lifecycle: None }
+    }
+
+    pub fn with_lifecycle(mut self, lifecycle: PluginLifecycle) -> Self {
+        self.lifecycle = Some(lifecycle);
+        self
     }
 
     pub fn install_file(
@@ -374,6 +383,7 @@ impl PluginPackageInstaller {
         }
         make_backend_executable(&compatibility.backend_executable)?;
 
+        let update_guard = self.lifecycle.as_ref().map(|lifecycle| lifecycle.begin_update(&manifest.id)).transpose()?;
         let container_dir = self.root_dir.join(&manifest.id);
         migrate_legacy_container(&container_dir)?;
         let versions_dir = container_dir.join(VERSIONS_DIR);
@@ -460,10 +470,11 @@ impl PluginPackageInstaller {
             log::warn!("Failed to prune plugin '{}' install history: {error}", manifest.id);
         }
         let plugin = InstalledPlugin::new(manifest, version_dir, &self.app_version);
-        Ok(PluginInstallResult { plugin, previous_version, package_sha256, signature })
+        Ok(PluginInstallResult { plugin, previous_version, package_sha256, signature, _update_guard: update_guard })
     }
 
     fn rollback_locked(&self, plugin_id: &str) -> Result<PluginRollbackResult, String> {
+        let update_guard = self.lifecycle.as_ref().map(|lifecycle| lifecycle.begin_update(plugin_id)).transpose()?;
         let container_dir = self.root_dir.join(plugin_id);
         let current = read_latest_activation(&container_dir)?
             .ok_or_else(|| format!("Plugin '{plugin_id}' does not have an active version"))?;
@@ -501,7 +512,7 @@ impl PluginPackageInstaller {
         if let Err(error) = prune_plugin_history(&container_dir, &activation) {
             log::warn!("Failed to prune plugin '{plugin_id}' rollback history: {error}");
         }
-        Ok(PluginRollbackResult { plugin, previous_version })
+        Ok(PluginRollbackResult { plugin, previous_version, _update_guard: update_guard })
     }
 }
 
@@ -1019,6 +1030,94 @@ mod tests {
     };
     use crate::plugins::{PluginManifest, PluginRegistry};
 
+    #[test]
+    fn runtime_updates_and_rollbacks_preserve_active_connections_and_versions() {
+        let root = tempfile::tempdir().unwrap();
+        let registry = PluginRegistry::new_with_app_version(root.path().to_path_buf(), "0.5.67");
+        let lifecycle = registry.lifecycle();
+        let installer =
+            PluginPackageInstaller::with_trust_store(root.path().to_path_buf(), "0.5.67", PluginTrustStore::default())
+                .with_lifecycle(lifecycle.clone());
+        let first = package("1.0.0", None, false);
+        let second = package("1.1.0", None, false);
+        installer.install_bytes(&first, PluginInstallPolicy::LocalDevelopment).unwrap();
+        let connection = lifecycle.begin_connection("sample.hello", "My connection").unwrap();
+        for bytes in [&first, &second] {
+            let error = installer.install_bytes(bytes, PluginInstallPolicy::LocalDevelopment).unwrap_err();
+            assert!(error.contains("My connection"));
+            assert_eq!(registry.find_plugin("sample.hello").unwrap().unwrap().manifest.version, "1.0.0");
+        }
+        assert!(!root.path().join("sample.hello/versions/1.1.0").exists());
+        drop(connection);
+
+        let unrelated = lifecycle.begin_connection("other.plugin", "Unrelated connection").unwrap();
+        let installed = installer.install_bytes(&second, PluginInstallPolicy::LocalDevelopment).unwrap();
+        assert!(lifecycle.begin_connection("sample.hello", "New connection").is_err());
+        assert!(lifecycle.begin_operation("sample.hello").is_err());
+        drop(installed);
+        let connection = lifecycle.begin_connection("sample.hello", "My connection").unwrap();
+        assert!(installer.rollback("sample.hello").unwrap_err().contains("My connection"));
+        assert_eq!(registry.find_plugin("sample.hello").unwrap().unwrap().manifest.version, "1.1.0");
+        drop(connection);
+        let rollback = installer.rollback("sample.hello").unwrap();
+        assert_eq!(registry.find_plugin("sample.hello").unwrap().unwrap().manifest.version, "1.0.0");
+        assert!(lifecycle.begin_operation("sample.hello").is_err());
+        drop(rollback);
+        assert!(lifecycle.begin_connection("sample.hello", "Reconnected").is_ok());
+        drop(unrelated);
+    }
+
+    #[test]
+    fn runtime_install_file_and_signed_packages_share_the_update_guard() {
+        let root = tempfile::tempdir().unwrap();
+        let registry = PluginRegistry::new_with_app_version(root.path().to_path_buf(), "0.5.67");
+        let lifecycle = registry.lifecycle();
+        let signer = SigningKey::from_bytes(&[23; 32]);
+        let trust = PluginTrustStore::from_base64_keys(BTreeMap::from([(
+            "test".to_string(),
+            base64::engine::general_purpose::STANDARD.encode(signer.verifying_key().as_bytes()),
+        )]))
+        .unwrap();
+        let installer = PluginPackageInstaller::with_trust_store(root.path().to_path_buf(), "0.5.67", trust)
+            .with_lifecycle(lifecycle.clone());
+        installer
+            .install_bytes(&package("1.0.0", Some((&signer, "test")), false), PluginInstallPolicy::LocalSigned)
+            .unwrap();
+        let bytes = package("1.1.0", Some((&signer, "test")), false);
+        let package_path = root.path().join("update.dbxp");
+        std::fs::write(&package_path, &bytes).unwrap();
+        let operation = lifecycle.begin_operation("sample.hello").unwrap();
+        assert!(installer
+            .install_file(&package_path, PluginInstallPolicy::LocalSigned)
+            .unwrap_err()
+            .contains("active operations"));
+        let expectation = PluginPackageExpectation {
+            id: "sample.hello".to_string(),
+            version: "1.1.0".to_string(),
+            publisher: "sample".to_string(),
+            permissions: BTreeSet::from(["host.events".to_string()]),
+            signing_key_id: "test".to_string(),
+        };
+        assert!(installer.install_marketplace_bytes(&bytes, &expectation).unwrap_err().contains("active operations"));
+        drop(operation);
+        installer.install_marketplace_bytes(&bytes, &expectation).unwrap();
+        assert_eq!(registry.find_plugin("sample.hello").unwrap().unwrap().manifest.version, "1.1.0");
+    }
+
+    #[test]
+    fn failed_runtime_update_releases_the_admission_guard() {
+        let root = tempfile::tempdir().unwrap();
+        let lifecycle = crate::plugins::PluginLifecycle::default();
+        let installer =
+            PluginPackageInstaller::with_trust_store(root.path().to_path_buf(), "0.5.67", PluginTrustStore::default())
+                .with_lifecycle(lifecycle.clone());
+        installer.install_bytes(&package("1.0.0", None, false), PluginInstallPolicy::LocalDevelopment).unwrap();
+        assert!(installer.rollback("sample.hello").is_err());
+        assert!(lifecycle.begin_connection("sample.hello", "Retry").is_ok());
+        assert!(installer.install_bytes(&package("1.1.0", None, true), PluginInstallPolicy::LocalDevelopment).is_err());
+        assert!(lifecycle.begin_update("sample.hello").is_ok());
+    }
+
     #[test]
     fn installs_versioned_package_and_rolls_back_atomically() {
         let root = tempfile::tempdir().unwrap();

crates/dbx-core/src/plugins/lifecycle.rs

@@ -0,0 +1,183 @@
+use std::collections::{BTreeMap, HashMap};
+use std::sync::{Arc, Mutex};
+
+const UPDATE_IN_PROGRESS: &str = "Plugin update is in progress. Please try again after it finishes.";
+const OPERATIONS_ACTIVE: &str = "Plugin update blocked by active operations. Please wait for them to finish.";
+
+#[derive(Debug, Clone, Default)]
+pub struct PluginLifecycle {
+    usage: Arc<Mutex<HashMap<String, PluginUsage>>>,
+}
+
+#[derive(Debug, Default)]
+struct PluginUsage {
+    connections: BTreeMap<String, usize>,
+    operations: usize,
+    updating: bool,
+}
+
+impl PluginUsage {
+    fn ensure_updatable(&self) -> Result<(), String> {
+        if self.updating {
+            return Err(UPDATE_IN_PROGRESS.to_string());
+        }
+        if !self.connections.is_empty() {
+            return Err(format!(
+                "Plugin update blocked by active connections: {}",
+                self.connections.keys().cloned().collect::<Vec<_>>().join(", ")
+            ));
+        }
+        if self.operations > 0 {
+            return Err(OPERATIONS_ACTIVE.to_string());
+        }
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct PluginUsageGuard {
+    _lease: Arc<PluginUsageLease>,
+}
+
+#[derive(Debug)]
+struct PluginUsageLease {
+    lifecycle: PluginLifecycle,
+    plugin_id: String,
+    kind: PluginUsageKind,
+}
+
+#[derive(Debug)]
+enum PluginUsageKind {
+    Connection(String),
+    Operation,
+    Update,
+}
+
+impl PluginLifecycle {
+    pub(crate) fn check_update(&self, plugin_id: &str) -> Result<(), String> {
+        let usage = self.usage.lock().unwrap_or_else(|error| error.into_inner());
+        usage.get(plugin_id).map_or(Ok(()), PluginUsage::ensure_updatable)
+    }
+
+    pub(crate) fn begin_connection(&self, plugin_id: &str, label: &str) -> Result<PluginUsageGuard, String> {
+        let label = if label.trim().is_empty() { plugin_id } else { label };
+        self.acquire(plugin_id, PluginUsageKind::Connection(label.to_string()))
+    }
+
+    pub(crate) fn begin_operation(&self, plugin_id: &str) -> Result<PluginUsageGuard, String> {
+        self.acquire(plugin_id, PluginUsageKind::Operation)
+    }
+
+    pub(crate) fn begin_update(&self, plugin_id: &str) -> Result<PluginUsageGuard, String> {
+        self.acquire(plugin_id, PluginUsageKind::Update)
+    }
+
+    fn acquire(&self, plugin_id: &str, kind: PluginUsageKind) -> Result<PluginUsageGuard, String> {
+        let mut plugins = self.usage.lock().unwrap_or_else(|error| error.into_inner());
+        let usage = plugins.entry(plugin_id.to_string()).or_default();
+        if usage.updating {
+            return Err(UPDATE_IN_PROGRESS.to_string());
+        }
+        match &kind {
+            PluginUsageKind::Connection(label) => *usage.connections.entry(label.clone()).or_default() += 1,
+            PluginUsageKind::Operation => usage.operations += 1,
+            PluginUsageKind::Update => {
+                usage.ensure_updatable()?;
+                usage.updating = true;
+            }
+        }
+        Ok(PluginUsageGuard {
+            _lease: Arc::new(PluginUsageLease { lifecycle: self.clone(), plugin_id: plugin_id.to_string(), kind }),
+        })
+    }
+}
+
+impl Drop for PluginUsageLease {
+    fn drop(&mut self) {
+        let mut plugins = self.lifecycle.usage.lock().unwrap_or_else(|error| error.into_inner());
+        let Some(usage) = plugins.get_mut(&self.plugin_id) else {
+            return;
+        };
+        match &self.kind {
+            PluginUsageKind::Connection(label) => {
+                if let Some(count) = usage.connections.get_mut(label) {
+                    *count -= 1;
+                    if *count == 0 {
+                        usage.connections.remove(label);
+                    }
+                }
+            }
+            PluginUsageKind::Operation => usage.operations -= 1,
+            PluginUsageKind::Update => usage.updating = false,
+        }
+        if usage.connections.is_empty() && usage.operations == 0 && !usage.updating {
+            plugins.remove(&self.plugin_id);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn connections_block_only_their_plugin_until_all_handles_are_dropped() {
+        let lifecycle = PluginLifecycle::default();
+        let connection = lifecycle.begin_connection("s3", "Production S3").unwrap();
+        let clone = connection.clone();
+        let second = lifecycle.begin_connection("s3", "Production S3").unwrap();
+        assert_eq!(
+            lifecycle.begin_update("s3").unwrap_err(),
+            "Plugin update blocked by active connections: Production S3"
+        );
+        assert!(lifecycle.begin_update("other").is_ok());
+        drop(connection);
+        drop(second);
+        assert!(lifecycle.begin_update("s3").is_err());
+        drop(clone);
+        assert!(lifecycle.begin_update("s3").is_ok());
+    }
+
+    #[test]
+    fn operations_block_updates_without_connections() {
+        let lifecycle = PluginLifecycle::default();
+        let operation = lifecycle.begin_operation("tool").unwrap();
+        assert_eq!(lifecycle.begin_update("tool").unwrap_err(), OPERATIONS_ACTIVE);
+        drop(operation);
+        assert!(lifecycle.begin_update("tool").is_ok());
+    }
+
+    #[test]
+    fn updates_block_new_connections_operations_and_updates() {
+        let lifecycle = PluginLifecycle::default();
+        let update = lifecycle.begin_update("s3").unwrap();
+        assert_eq!(lifecycle.begin_connection("s3", "S3").unwrap_err(), UPDATE_IN_PROGRESS);
+        assert_eq!(lifecycle.begin_operation("s3").unwrap_err(), UPDATE_IN_PROGRESS);
+        assert_eq!(lifecycle.begin_update("s3").unwrap_err(), UPDATE_IN_PROGRESS);
+        assert!(lifecycle.begin_connection("other", "Other").is_ok());
+        drop(update);
+        assert!(lifecycle.begin_connection("s3", "S3").is_ok());
+        assert!(lifecycle.usage.lock().unwrap().is_empty());
+    }
+
+    #[test]
+    fn connection_and_update_admission_are_atomic() {
+        for _ in 0..64 {
+            let lifecycle = PluginLifecycle::default();
+            let barrier = Arc::new(std::sync::Barrier::new(2));
+            let other_lifecycle = lifecycle.clone();
+            let other_barrier = barrier.clone();
+            let connection = std::thread::spawn(move || {
+                other_barrier.wait();
+                let result = other_lifecycle.begin_connection("s3", "S3");
+                other_barrier.wait();
+                result
+            });
+            barrier.wait();
+            let update = lifecycle.begin_update("s3");
+            barrier.wait();
+            let connection = connection.join().unwrap();
+            assert_ne!(connection.is_ok(), update.is_ok());
+        }
+    }
+}

crates/dbx-core/src/plugins/marketplace.rs

@@ -284,6 +284,7 @@ pub struct PluginMarketplace {
     app_version: String,
     repositories: PluginRepositoryStore,
     client: Client,
+    lifecycle: Option<super::PluginLifecycle>,
 }
 
 impl PluginMarketplace {
@@ -300,6 +301,7 @@ impl PluginMarketplace {
             root_dir,
             app_version: app_version.into(),
             client,
+            lifecycle: None,
         })
     }
 
@@ -307,6 +309,18 @@ impl PluginMarketplace {
         &self.repositories
     }
 
+    pub fn with_lifecycle(mut self, lifecycle: super::PluginLifecycle) -> Self {
+        self.lifecycle = Some(lifecycle);
+        self
+    }
+
+    fn guard_installer(&self, installer: PluginPackageInstaller) -> PluginPackageInstaller {
+        match &self.lifecycle {
+            Some(lifecycle) => installer.with_lifecycle(lifecycle.clone()),
+            None => installer,
+        }
+    }
+
     pub async fn fetch_catalogs(&self) -> Vec<PluginRepositoryCatalogResult> {
         let repositories = match self.repositories.list() {
             Ok(repositories) => repositories,
@@ -372,6 +386,9 @@ impl PluginMarketplace {
     }
 
     pub async fn install(&self, request: PluginMarketplaceInstallRequest) -> Result<PluginInstallResult, String> {
+        if let Some(lifecycle) = &self.lifecycle {
+            lifecycle.check_update(&request.plugin_id)?;
+        }
         let repository = self.repositories.find(&request.repository_id)?;
         if !repository.enabled {
             return Err(format!("Plugin repository '{}' is disabled", repository.id));
@@ -403,8 +420,12 @@ impl PluginMarketplace {
             permissions: plugin.permissions.iter().cloned().collect(),
             signing_key_id: artifact.signing_key_id.clone(),
         };
-        PluginPackageInstaller::with_trust_store(self.root_dir.clone(), self.app_version.clone(), trust_store)
-            .install_marketplace_bytes(&package, &expectation)
+        self.guard_installer(PluginPackageInstaller::with_trust_store(
+            self.root_dir.clone(),
+            self.app_version.clone(),
+            trust_store,
+        ))
+        .install_marketplace_bytes(&package, &expectation)
     }
 
     /// Downloads a .dbxp package from a direct http(s) URL and installs it with
@@ -445,7 +466,8 @@ impl PluginMarketplace {
             bytes.extend_from_slice(&chunk);
             on_progress(bytes.len() as u64, total);
         }
-        PluginPackageInstaller::new(self.root_dir.clone(), self.app_version.clone())?.install_bytes(&bytes, policy)
+        self.guard_installer(PluginPackageInstaller::new(self.root_dir.clone(), self.app_version.clone())?)
+            .install_bytes(&bytes, policy)
     }
 
     async fn download_limited(&self, url: Url, max_bytes: usize, label: &str) -> Result<Vec<u8>, String> {
@@ -1269,6 +1291,57 @@ mod tests {
         assert_eq!(last_progress, (package.len() as u64, Some(package.len() as u64)));
     }
 
+    #[tokio::test]
+    async fn direct_url_rechecks_runtime_usage_after_download() {
+        let root = tempfile::tempdir().unwrap();
+        let signing_key = SigningKey::from_bytes(&[7u8; 32]);
+        let key_id = "direct-url-release";
+        PluginTrustStore::save_base64_key(
+            root.path(),
+            key_id,
+            &base64::engine::general_purpose::STANDARD.encode(signing_key.verifying_key().as_bytes()),
+        )
+        .unwrap();
+        let registry = crate::plugins::PluginRegistry::new_with_app_version(root.path().to_path_buf(), "0.5.68");
+        let lifecycle = registry.lifecycle();
+        let marketplace =
+            PluginMarketplace::new(root.path().to_path_buf(), "0.5.68").unwrap().with_lifecycle(lifecycle.clone());
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+        let server = serve_package_once(listener, signed_package(&signing_key, key_id));
+        let mut connection = None;
+        let result = marketplace
+            .install_url_package(&format!("http://{address}/plugin.dbxp"), PluginInstallPolicy::LocalSigned, |_, _| {
+                connection.get_or_insert_with(|| {
+                    lifecycle.begin_connection("marketplace.install", "Started during download").unwrap()
+                });
+            })
+            .await;
+        server.await.unwrap();
+        assert!(result.unwrap_err().contains("Started during download"));
+        assert!(registry.find_plugin("marketplace.install").unwrap().is_none());
+        drop(connection);
+        assert!(lifecycle.begin_update("marketplace.install").is_ok());
+    }
+
+    #[tokio::test]
+    async fn marketplace_reports_active_connections_before_fetching_catalogs() {
+        let root = tempfile::tempdir().unwrap();
+        let lifecycle = crate::plugins::PluginLifecycle::default();
+        let marketplace =
+            PluginMarketplace::new(root.path().to_path_buf(), "0.5.68").unwrap().with_lifecycle(lifecycle.clone());
+        let _connection = lifecycle.begin_connection("marketplace.install", "Production").unwrap();
+        let error = marketplace
+            .install(super::PluginMarketplaceInstallRequest {
+                repository_id: "not-configured".to_string(),
+                plugin_id: "marketplace.install".to_string(),
+                version: None,
+            })
+            .await
+            .unwrap_err();
+        assert!(error.contains("Production"));
+    }
+
     #[tokio::test]
     async fn rejects_url_package_signed_by_an_untrusted_key() {
         let root = tempfile::tempdir().unwrap();

crates/dbx-web/src/routes/plugins.rs

@@ -228,7 +228,8 @@ pub async fn install_marketplace_plugin(
 ) -> Result<Json<PluginInstallResponse>, AppError> {
     let marketplace =
         PluginMarketplace::new(state.app.plugins.root_dir().to_path_buf(), state.app.plugins.app_version().to_string())
-            .map_err(AppError::from)?;
+            .map_err(AppError::from)?
+            .with_lifecycle(state.app.plugins.lifecycle());
     let result = marketplace.install(request).await.map_err(AppError::bad_request)?;
     state.app.remove_plugin_connection_pools(&result.plugin.manifest.id).await;
     stop_external_driver_pools(&state, &result.plugin).await;
@@ -288,6 +289,7 @@ pub async fn install_plugin(
         if query.allow_unsigned { PluginInstallPolicy::LocalDevelopment } else { PluginInstallPolicy::LocalSigned };
     let root_dir = state.app.plugins.root_dir().to_path_buf();
     let app_version = state.app.plugins.app_version().to_string();
+    let lifecycle = state.app.plugins.lifecycle();
     let result = tokio::task::spawn_blocking(move || {
         let installer = if query.from_url {
             let trust_store = dbx_core::plugins::url_install_trust_store(&root_dir)?;
@@ -295,7 +297,7 @@ pub async fn install_plugin(
         } else {
             PluginPackageInstaller::new(root_dir.clone(), app_version)?
         };
-        installer.install_bytes(&package, policy)
+        installer.with_lifecycle(lifecycle).install_bytes(&package, policy)
     })
     .await
     .map_err(|error| AppError::internal(error.to_string()))?
@@ -310,17 +312,18 @@ pub async fn rollback_plugin(
     State(state): State<Arc<WebState>>,
     Json(request): Json<PluginIdRequest>,
 ) -> Result<Json<PluginRollbackResponse>, AppError> {
-    state.app.remove_plugin_connection_pools(&request.plugin_id).await;
-    state.app.plugin_host.stop(&request.plugin_id).await;
     let root_dir = state.app.plugins.root_dir().to_path_buf();
     let app_version = state.app.plugins.app_version().to_string();
+    let lifecycle = state.app.plugins.lifecycle();
     let result = tokio::task::spawn_blocking(move || {
-        PluginPackageInstaller::new(root_dir, app_version)?.rollback(&request.plugin_id)
+        PluginPackageInstaller::new(root_dir, app_version)?.with_lifecycle(lifecycle).rollback(&request.plugin_id)
     })
     .await
     .map_err(|error| AppError::internal(error.to_string()))?
     .map_err(AppError::bad_request)?;
+    state.app.remove_plugin_connection_pools(&result.plugin.manifest.id).await;
     stop_external_driver_pools(&state, &result.plugin).await;
+    state.app.plugin_host.stop(&result.plugin.manifest.id).await;
     Ok(Json(result.response()))
 }