fix(updates): auto-continue client upgrade on update-all
This commit is contained in:
10 files changed
+200
-61
No files matched your search
@@ -4904,15 +4904,17 @@ impl AppState {
|
||||
Ok(closed)
|
||||
}
|
||||
|
||||
pub async fn active_agent_connection_driver_keys(&self) -> HashSet<String> {
|
||||
pub async fn active_agent_connection_driver_connections(&self) -> HashMap<String, Vec<String>> {
|
||||
let configs = self.configs.read().await;
|
||||
let connections = self.connection_pools_snapshot().await;
|
||||
let mut keys = HashSet::new();
|
||||
let mut connections_by_key: HashMap<String, Vec<String>> = HashMap::new();
|
||||
|
||||
for (pool_key, pool) in connections.iter() {
|
||||
#[cfg(feature = "duckdb-sidecar")]
|
||||
if matches!(pool, PoolKind::DuckDbWorker(_)) {
|
||||
keys.insert("duckdb".to_string());
|
||||
if let Some(config) = config_for_pool_key(pool_key, &configs) {
|
||||
connections_by_key.entry("duckdb".to_string()).or_default().push(config.name.clone());
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if !matches!(pool, PoolKind::Agent(_)) {
|
||||
@@ -4925,25 +4927,33 @@ impl AppState {
|
||||
&config.db_type,
|
||||
config.driver_profile.as_deref(),
|
||||
) {
|
||||
keys.insert(agent_key.to_string());
|
||||
connections_by_key.entry(agent_key.to_string()).or_default().push(config.name.clone());
|
||||
}
|
||||
}
|
||||
|
||||
keys
|
||||
for names in connections_by_key.values_mut() {
|
||||
names.sort();
|
||||
names.dedup();
|
||||
}
|
||||
connections_by_key
|
||||
}
|
||||
|
||||
pub async fn prepare_agent_driver_updates(&self, driver_keys: &[String]) -> HashSet<String> {
|
||||
pub async fn active_agent_connection_driver_keys(&self) -> HashSet<String> {
|
||||
self.active_agent_connection_driver_connections().await.into_keys().collect()
|
||||
}
|
||||
|
||||
pub async fn prepare_agent_driver_updates(&self, driver_keys: &[String]) -> HashMap<String, Vec<String>> {
|
||||
let candidates = driver_keys.iter().cloned().collect::<HashSet<_>>();
|
||||
if candidates.is_empty() {
|
||||
return HashSet::new();
|
||||
return HashMap::new();
|
||||
}
|
||||
|
||||
let blockers = self
|
||||
.active_agent_connection_driver_keys()
|
||||
.active_agent_connection_driver_connections()
|
||||
.await
|
||||
.into_iter()
|
||||
.filter(|key| candidates.contains(key))
|
||||
.collect::<HashSet<_>>();
|
||||
.filter(|(key, _)| candidates.contains(key))
|
||||
.collect::<HashMap<_, _>>();
|
||||
if !blockers.is_empty() {
|
||||
return blockers;
|
||||
}
|
||||
@@ -4953,7 +4963,11 @@ impl AppState {
|
||||
}
|
||||
|
||||
// A connection may have started while idle runtimes were stopping.
|
||||
self.active_agent_connection_driver_keys().await.into_iter().filter(|key| candidates.contains(key)).collect()
|
||||
self.active_agent_connection_driver_connections()
|
||||
.await
|
||||
.into_iter()
|
||||
.filter(|(key, _)| candidates.contains(key))
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub async fn connection_identifier_quote(
|
||||
@@ -7842,8 +7856,13 @@ mod tests {
|
||||
let (state, dir) = test_app_state().await;
|
||||
let mut config = mysql_config(None);
|
||||
config.id = "dameng-conn".to_string();
|
||||
config.name = "达梦生产".to_string();
|
||||
config.db_type = DatabaseType::Dameng;
|
||||
let mut replica_config = config.clone();
|
||||
replica_config.id = "dameng-replica".to_string();
|
||||
replica_config.name = "达梦报表".to_string();
|
||||
state.configs.write().await.insert(config.id.clone(), config);
|
||||
state.configs.write().await.insert(replica_config.id.clone(), replica_config);
|
||||
state
|
||||
.agent_manager
|
||||
.daemons
|
||||
@@ -7851,14 +7870,19 @@ mod tests {
|
||||
.await
|
||||
.insert("oracle".to_string(), crate::db::agent_driver::AgentDriverClient::test_stub());
|
||||
state.connections.write().await.insert("dameng-conn".to_string(), agent_pool_stub());
|
||||
state.connections.write().await.insert("dameng-replica".to_string(), agent_pool_stub());
|
||||
|
||||
assert_eq!(
|
||||
state.active_agent_connection_driver_keys().await,
|
||||
std::collections::HashSet::from(["dameng".to_string()])
|
||||
state.active_agent_connection_driver_connections().await,
|
||||
std::collections::HashMap::from([(
|
||||
"dameng".to_string(),
|
||||
vec!["达梦报表".to_string(), "达梦生产".to_string()]
|
||||
)])
|
||||
);
|
||||
|
||||
state.connections.write().await.remove("dameng-conn");
|
||||
assert!(state.active_agent_connection_driver_keys().await.is_empty());
|
||||
state.connections.write().await.remove("dameng-replica");
|
||||
assert!(state.active_agent_connection_driver_connections().await.is_empty());
|
||||
|
||||
let _ = std::fs::remove_dir_all(dir);
|
||||
}
|
||||
@@ -7886,6 +7910,7 @@ mod tests {
|
||||
let (state, dir) = test_app_state().await;
|
||||
let mut config = mysql_config(None);
|
||||
config.id = "dameng-conn".to_string();
|
||||
config.name = "达梦生产".to_string();
|
||||
config.db_type = DatabaseType::Dameng;
|
||||
state.configs.write().await.insert(config.id.clone(), config);
|
||||
state.connections.write().await.insert("dameng-conn".to_string(), agent_pool_stub());
|
||||
@@ -7898,7 +7923,7 @@ mod tests {
|
||||
|
||||
let blockers = state.prepare_agent_driver_updates(&["dameng".to_string()]).await;
|
||||
|
||||
assert_eq!(blockers, std::collections::HashSet::from(["dameng".to_string()]));
|
||||
assert_eq!(blockers, std::collections::HashMap::from([("dameng".to_string(), vec!["达梦生产".to_string()])]));
|
||||
assert_eq!(state.agent_manager.active_daemon_keys().await, vec!["dameng".to_string()]);
|
||||
|
||||
let _ = std::fs::remove_dir_all(dir);
|
||||
|
||||
@@ -53,6 +53,7 @@ pub struct AgentUpdateBlockersRequest {
|
||||
pub struct AgentUpdateBlocker {
|
||||
pub db_type: String,
|
||||
pub label: String,
|
||||
pub connections: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
@@ -407,32 +408,37 @@ async fn ensure_no_agent_update_blockers(
|
||||
state: &dbx_core::connection::AppState,
|
||||
db_types: &[String],
|
||||
) -> Result<(), String> {
|
||||
let blockers = update_blockers_from_keys(state.prepare_agent_driver_updates(db_types).await, db_types);
|
||||
let blockers = update_blockers_from_connections(state.prepare_agent_driver_updates(db_types).await, db_types);
|
||||
if blockers.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
let labels = blockers.into_iter().map(|blocker| blocker.label).collect::<Vec<_>>().join(", ");
|
||||
let labels = blockers
|
||||
.into_iter()
|
||||
.flat_map(|blocker| if blocker.connections.is_empty() { vec![blocker.label] } else { blocker.connections })
|
||||
.collect::<Vec<_>>()
|
||||
.join(", ");
|
||||
Err(format!("Close these database connections before updating drivers: {labels}"))
|
||||
}
|
||||
|
||||
async fn agent_update_blockers(state: &dbx_core::connection::AppState, db_types: &[String]) -> Vec<AgentUpdateBlocker> {
|
||||
update_blockers_from_keys(state.active_agent_connection_driver_keys().await, db_types)
|
||||
update_blockers_from_connections(state.active_agent_connection_driver_connections().await, db_types)
|
||||
}
|
||||
|
||||
fn update_blockers_from_keys(
|
||||
active_keys: std::collections::HashSet<String>,
|
||||
fn update_blockers_from_connections(
|
||||
active_connections: std::collections::HashMap<String, Vec<String>>,
|
||||
db_types: &[String],
|
||||
) -> Vec<AgentUpdateBlocker> {
|
||||
let candidate_keys: std::collections::HashSet<&str> = db_types.iter().map(String::as_str).collect();
|
||||
if candidate_keys.is_empty() {
|
||||
return Vec::new();
|
||||
}
|
||||
let mut blockers = active_keys
|
||||
let mut blockers = active_connections
|
||||
.into_iter()
|
||||
.filter(|key| candidate_keys.contains(key.as_str()))
|
||||
.map(|db_type| AgentUpdateBlocker {
|
||||
.filter(|(key, _)| candidate_keys.contains(key.as_str()))
|
||||
.map(|(db_type, connections)| AgentUpdateBlocker {
|
||||
label: dbx_core::agent_catalog::label_for_key(&db_type).unwrap_or(&db_type).to_string(),
|
||||
db_type,
|
||||
connections,
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
blockers.sort_by(|left, right| left.label.cmp(&right.label));
|
||||
@@ -472,12 +478,16 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn update_blockers_include_active_duckdb_connections() {
|
||||
let active_keys = ["duckdb".to_string(), "oracle".to_string()].into_iter().collect();
|
||||
let blockers = update_blockers_from_keys(active_keys, &["duckdb".to_string()]);
|
||||
let active_connections = std::collections::HashMap::from([
|
||||
("duckdb".to_string(), vec!["本地分析".to_string()]),
|
||||
("oracle".to_string(), vec!["生产 Oracle".to_string()]),
|
||||
]);
|
||||
let blockers = update_blockers_from_connections(active_connections, &["duckdb".to_string()]);
|
||||
|
||||
assert_eq!(blockers.len(), 1);
|
||||
assert_eq!(blockers[0].db_type, "duckdb");
|
||||
assert_eq!(blockers[0].label, "DuckDB");
|
||||
assert_eq!(blockers[0].connections, vec!["本地分析"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
Reference in new issue
Block a user