From c09d11ff3be36070f9cc929bd19c030eb6c1ec5d Mon Sep 17 00:00:00 2001 From: cxymds Date: Sat, 1 Aug 2026 08:33:46 +0800 Subject: [PATCH] fix(config): fence persisted config updates and reloads (#5512) * fix(config): fence persisted config updates and reloads * fix(ci): unblock config and e2e checks Co-Authored-By: heihutu --------- Co-authored-by: houseme Co-authored-by: heihutu --- crates/ecstore/src/api/mod.rs | 13 +- crates/ecstore/src/config/com.rs | 695 ++++++++++++++++-- crates/ecstore/src/set_disk/mod.rs | 225 +++++- crates/notify/src/config_manager.rs | 201 ++--- crates/notify/src/lib.rs | 5 +- crates/notify/src/storage_api.rs | 36 +- .../admin/handlers/audit_runtime_config.rs | 187 ++++- rustfs/src/admin/handlers/config_admin.rs | 343 ++++----- rustfs/src/admin/service/config.rs | 446 +++++++++-- rustfs/src/admin/storage_api.rs | 25 +- scripts/e2e-run.sh | 14 +- 11 files changed, 1711 insertions(+), 479 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 15aebe9d9..cfba93f88 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -267,12 +267,13 @@ pub mod config { pub mod com { pub use crate::config::com::{ COMMA_SEPARATED_LISTS, CONFIG_PREFIX, ENV_CONFIG_RECOVER_ON_CORRUPTION, STORAGE_CLASS_SUB_SYS, - ServerConfigCorruptError, ServerConfigSnapshot, delete_config, is_server_config_corrupt_error, lookup_configs, - read_config, read_config_no_lock, read_config_with_metadata, read_config_without_migrate, - read_config_without_migrate_no_lock, read_existing_server_config_no_lock, read_server_config_snapshot, save_config, - save_config_no_lock, save_config_with_opts, save_server_config, save_server_config_no_lock, - save_server_config_snapshot, server_config_path, try_migrate_server_config, with_config_object_read_lock, - with_config_object_write_lock, with_server_config_read_lock, with_server_config_write_lock, + ServerConfigCorruptError, ServerConfigSaveResult, ServerConfigSnapshot, delete_config, + is_server_config_corrupt_error, lookup_configs, read_config, read_config_no_lock, read_config_with_metadata, + read_config_without_migrate, read_config_without_migrate_no_lock, read_existing_server_config_no_lock, + read_server_config_snapshot, save_config, save_config_no_lock, save_config_with_opts, save_server_config, + save_server_config_no_lock, save_server_config_snapshot, save_server_config_snapshot_with_generation, + server_config_path, try_migrate_server_config, with_config_object_read_lock, with_config_object_write_lock, + with_server_config_read_lock, with_server_config_write_lock, }; } diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index 539464173..c62aa74d8 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -53,12 +53,14 @@ use std::sync::LazyLock; use std::sync::{Arc, RwLock}; use tokio::sync::{OwnedRwLockWriteGuard, RwLock as AsyncRwLock}; use tracing::{debug, error, info, instrument, warn}; +use uuid::Uuid; pub const CONFIG_PREFIX: &str = "config"; const SERVER_CONFIG_OBJECT: &str = "config/config.json"; +const CONFIG_TRANSACTION_LOCK_SUFFIX: &str = ".transaction.lock"; -// Server-config lock order: SERVER_CONFIG_LOCK -> distributed namespace lock -// for SERVER_CONFIG_OBJECT. Readers and writers must never reverse this order. +// Server-config lock order: SERVER_CONFIG_LOCK -> transaction lock -> +// SERVER_CONFIG_OBJECT. Readers and writers must never reverse this order. static SERVER_CONFIG_LOCK: LazyLock>> = LazyLock::new(|| Arc::new(AsyncRwLock::new(()))); fn config_task_join_error(operation: &'static str, error: tokio::task::JoinError) -> Error { @@ -76,8 +78,11 @@ where T: Send + 'static, { tokio::spawn(async move { - // Lock order: SERVER_CONFIG_LOCK -> namespace write lock. + // Lock order: SERVER_CONFIG_LOCK -> transaction lock -> object lock. let _local_guard = SERVER_CONFIG_LOCK.write().await; + let transaction_lock = server_config_transaction_lock_path(); + let transaction_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &transaction_lock).await?; + let _transaction_guard = transaction_lock.get_write_lock(get_lock_acquire_timeout()).await?; let namespace_lock = store.new_ns_lock(RUSTFS_META_BUCKET, SERVER_CONFIG_OBJECT).await?; let _write_guard = namespace_lock.get_write_lock(get_lock_acquire_timeout()).await?; Ok(operation().await) @@ -96,8 +101,11 @@ where T: Send + 'static, { tokio::spawn(async move { - // Lock order: SERVER_CONFIG_LOCK -> namespace read lock. + // Lock order: SERVER_CONFIG_LOCK -> transaction lock -> object lock. let _local_guard = SERVER_CONFIG_LOCK.read().await; + let transaction_lock = server_config_transaction_lock_path(); + let transaction_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &transaction_lock).await?; + let _transaction_guard = transaction_lock.get_read_lock(get_lock_acquire_timeout()).await?; let namespace_lock = store.new_ns_lock(RUSTFS_META_BUCKET, SERVER_CONFIG_OBJECT).await?; let _read_guard = namespace_lock.get_read_lock(get_lock_acquire_timeout()).await?; Ok(operation().await) @@ -567,6 +575,21 @@ where } pub async fn save_config_with_opts(api: Arc, file: &str, data: Vec, opts: &ObjectOptions) -> Result<()> +where + S: ObjectIO< + Error = Error, + RangeSpec = HTTPRangeSpec, + HeaderMap = HeaderMap, + ObjectOptions = ObjectOptions, + ObjectInfo = ObjectInfo, + GetObjectReader = GetObjectReader, + PutObjectReader = PutObjReader, + >, +{ + save_config_with_opts_and_metadata(api, file, data, opts).await.map(|_| ()) +} + +async fn save_config_with_opts_and_metadata(api: Arc, file: &str, data: Vec, opts: &ObjectOptions) -> Result where S: ObjectIO< Error = Error, @@ -579,11 +602,13 @@ where >, { let mut put_data = PutObjReader::from_vec(data); - if let Err(err) = api.put_object(RUSTFS_META_BUCKET, file, &mut put_data, opts).await { - error!("save_config_with_opts: err: {:?}, file: {}", err, file); - return Err(err); + match api.put_object(RUSTFS_META_BUCKET, file, &mut put_data, opts).await { + Ok(object_info) => Ok(object_info), + Err(err) => { + error!("save_config_with_opts: err: {:?}, file: {}", err, file); + Err(err) + } } - Ok(()) } fn new_server_config() -> Config { @@ -594,8 +619,12 @@ async fn new_and_save_server_config(api: Arc) -> Result where S: EcstoreObjectIO + StorageAdminApi + NamespaceLocking, { + let snapshot = read_server_config_snapshot(api.clone()).await?; + if snapshot.object_exists() { + return Ok(snapshot.config.clone()); + } let cfg = new_server_config(); - save_server_config(api, &cfg).await?; + save_server_config_snapshot(api, &cfg, &snapshot).await?; Ok(cfg) } @@ -617,6 +646,10 @@ pub fn server_config_path() -> String { SERVER_CONFIG_OBJECT.to_string() } +fn server_config_transaction_lock_path() -> String { + format!("{}{CONFIG_TRANSACTION_LOCK_SUFFIX}", server_config_path()) +} + fn storage_class_kvs_mut(cfg: &mut Config) -> &mut KVS { let sub_cfg = cfg.0.entry(STORAGE_CLASS_SUB_SYS.to_string()).or_insert_with(|| { let mut section = HashMap::new(); @@ -819,6 +852,9 @@ fn apply_external_scalar_config_map( let Some(config_value) = root.get(descriptor.subsystem_key) else { return Ok(false); }; + if descriptor.subsystem_key == HEAL_SUB_SYS && config_value.is_null() { + return Ok(false); + } let overrides = decode_scalar_config_value(config_value, descriptor)?; if overrides.is_empty() { @@ -1463,21 +1499,142 @@ fn build_audit_object(cfg: &Config) -> Map { build_target_object(cfg, &audit_target_descriptors()) } +fn sync_rendered_target_instance(existing: Value, rendered: Option<&Value>, valid_keys: &[&str]) -> Option { + match existing { + Value::Object(mut instance) => { + for key in valid_keys { + instance.remove(*key); + } + if let Some(Value::Object(rendered)) = rendered { + instance.extend(rendered.clone()); + } + (!instance.is_empty()).then_some(Value::Object(instance)) + } + Value::Array(entries) => { + let mut pending = rendered + .and_then(Value::as_object) + .map(|rendered| { + rendered + .iter() + .filter_map(|(key, value)| parse_target_scalar_value(key, value).map(|value| (key.clone(), value))) + .collect::>() + }) + .unwrap_or_default(); + let mut updated = Vec::with_capacity(entries.len().saturating_add(pending.len())); + for entry in entries { + let Some(entry_obj) = entry.as_object() else { + updated.push(entry); + continue; + }; + let Some(key) = entry_obj.get("key").and_then(Value::as_str) else { + updated.push(entry); + continue; + }; + if !valid_keys.contains(&key) { + updated.push(entry); + continue; + } + let Some(value) = pending.remove(key) else { + continue; + }; + let mut entry_obj = entry_obj.clone(); + entry_obj.insert("value".to_string(), Value::String(value)); + updated.push(Value::Object(entry_obj)); + } + updated.extend(rendered_scalar_config_kvs_entries(&pending)); + (!updated.is_empty()).then_some(Value::Array(updated)) + } + value if rendered.is_none() => Some(value), + _ => rendered.cloned(), + } +} + fn sync_rendered_target_object( target_obj: &mut Map, rendered_target: &Map, descriptors: &[TargetConfigDescriptor], ) { for descriptor in descriptors { - match rendered_target.get(descriptor.external_key) { - Some(Value::Object(v)) => { - target_obj.insert(descriptor.external_key.to_string(), Value::Object(v.clone())); - target_obj.remove(descriptor.subsystem_key); + let existing = target_obj.remove(descriptor.external_key); + let alias = target_obj.remove(descriptor.subsystem_key); + let mut section = existing + .or(alias) + .and_then(|value| value.as_object().cloned()) + .unwrap_or_default(); + let rendered = rendered_target.get(descriptor.external_key).and_then(Value::as_object); + + if is_target_instance_shorthand(§ion, descriptor.valid_keys) { + let has_named_instances = rendered.is_some_and(|instances| instances.keys().any(|name| name != "default")); + if !has_named_instances { + if let Some(section) = sync_rendered_target_instance( + Value::Object(section), + rendered.and_then(|instances| instances.get("default")), + descriptor.valid_keys, + ) { + target_obj.insert(descriptor.external_key.to_string(), section); + } + continue; } - _ => { - target_obj.remove(descriptor.external_key); - target_obj.remove(descriptor.subsystem_key); + + let mut nested = Map::new(); + if let Some(default) = sync_rendered_target_instance( + Value::Object(section), + rendered.and_then(|instances| instances.get("default")), + descriptor.valid_keys, + ) { + nested.insert("default".to_string(), default); } + if let Some(rendered) = rendered { + for (instance_name, instance) in rendered { + if instance_name != "default" { + nested.insert(instance_name.clone(), instance.clone()); + } + } + } + if !nested.is_empty() { + target_obj.insert(descriptor.external_key.to_string(), Value::Object(nested)); + } + continue; + } + + if let Some(default_alias) = section.remove(DEFAULT_DELIMITER) { + if let Some(default) = section.get_mut("default") { + if let Some(alias) = sync_rendered_target_instance(default_alias, None, descriptor.valid_keys) { + match (default, alias) { + (Value::Object(default), Value::Object(alias)) => { + for (key, value) in alias { + default.entry(key).or_insert(value); + } + } + (Value::Array(default), Value::Array(alias)) => default.extend(alias), + _ => {} + } + } + } else { + section.insert("default".to_string(), default_alias); + } + } + + let mut merged = Map::new(); + for (instance_name, instance) in section { + if let Some(instance) = sync_rendered_target_instance( + instance, + rendered.and_then(|instances| instances.get(&instance_name)), + descriptor.valid_keys, + ) { + merged.insert(instance_name, instance); + } + } + if let Some(rendered) = rendered { + for (instance_name, instance) in rendered { + if !merged.contains_key(instance_name) { + merged.insert(instance_name.clone(), instance.clone()); + } + } + } + + if !merged.is_empty() { + target_obj.insert(descriptor.external_key.to_string(), Value::Object(merged)); } } } @@ -1496,6 +1653,14 @@ fn encode_server_config_blob(cfg: &Config, seed: Option<&[u8]>) -> Result v, _ => Map::new(), }; + for key in [ + storageclass::CLASS_STANDARD, + storageclass::CLASS_RRS, + storageclass::OPTIMIZE, + storageclass::INLINE_BLOCK, + ] { + sc_obj.remove(key); + } for (k, v) in build_storageclass_object(cfg) { sc_obj.insert(k, v); } @@ -1503,7 +1668,10 @@ fn encode_server_config_blob(cfg: &Config, seed: Option<&[u8]>) -> Result bool { matches!(root.get("version"), Some(Value::String(v)) if !v.trim().is_empty()) && matches!(root.get("storageclass"), Some(Value::Object(_))) && !root.contains_key("storage_class") + && !matches!(root.get(HEAL_SUB_SYS), Some(Value::Null)) } fn configs_semantically_equal(lhs: &Config, rhs: &Config) -> bool { @@ -1593,7 +1762,7 @@ where FileInfo = FileInfo, ObjectToDelete = ObjectToDelete, DeletedObject = DeletedObject, - >, + > + NamespaceLocking, { if let Some(decrypt) = &decrypt_fn { register_server_config_decrypt_fn(decrypt.clone()); @@ -1601,14 +1770,7 @@ where let config_file = server_config_path(); match api - .get_object_info( - RUSTFS_META_BUCKET, - &config_file, - &ObjectOptions { - no_lock: true, - ..Default::default() - }, - ) + .get_object_info(RUSTFS_META_BUCKET, &config_file, &ObjectOptions::default()) .await { Ok(_) => { @@ -1624,7 +1786,6 @@ where let opts = ObjectOptions { max_parity: true, - no_lock: true, ..Default::default() }; @@ -1677,7 +1838,33 @@ where } }; - match save_config(api, &config_file, normalized).await { + let snapshot = match read_server_config_snapshot(api.clone()).await { + Ok(snapshot) => snapshot, + Err(err) => { + warn!("recheck target server config failed, skip migration: {:?}", err); + return; + } + }; + if snapshot.object_exists() { + debug!("server config was created while legacy migration was preparing, skip migration"); + return; + } + + match save_config_with_opts( + api, + &config_file, + normalized, + &ObjectOptions { + max_parity: true, + http_preconditions: Some(HTTPPreconditions { + if_none_match: Some("*".to_string()), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + { Ok(()) => { info!("Migrated compatible server config from legacy metadata bucket"); } @@ -1769,8 +1956,13 @@ where { let config_file = server_config_path(); - // Try to read the configuration file - match read_config_no_lock(api.clone(), &config_file).await { + // Try to read the configuration file. + let data = if namespace_lock_held { + read_config_no_lock(api.clone(), &config_file).await + } else { + read_config(api.clone(), &config_file).await + }; + match data { Ok(data) => read_server_config(api, &data, namespace_lock_held).await, Err(Error::ConfigNotFound) => handle_missing_config(api, "Read the main configuration", namespace_lock_held).await, Err(err) => handle_config_read_error(err, &config_file), @@ -1787,7 +1979,12 @@ where warn!("Received empty configuration data, try to reread from '{}'", config_file); // Try to read the configuration again - match read_config_no_lock(api.clone(), &config_file).await { + let data = if namespace_lock_held { + read_config_no_lock(api.clone(), &config_file).await + } else { + read_config(api.clone(), &config_file).await + }; + match data { Ok(cfg_data) => { let cfg = decode_persisted_server_config(&cfg_data)?; return Ok(cfg.merge()); @@ -2036,11 +2233,16 @@ pub struct ServerConfigSnapshot { raw: Option>, seed: Option>, etag: Option, + generation: Option, _local_guard: OwnedRwLockWriteGuard<()>, _guard: rustfs_lock::NamespaceLockGuard, } impl ServerConfigSnapshot { + pub fn object_exists(&self) -> bool { + self.raw.is_some() + } + pub fn ensure_lock_held(&self) -> Result<()> { if self._guard.is_lock_lost() { return Err(Error::other("server config transaction lock was lost")); @@ -2051,12 +2253,34 @@ impl ServerConfigSnapshot { pub fn is_lock_lost(&self) -> bool { self._guard.is_lock_lost() } + + pub fn generation(&self) -> Option { + self.generation + } } -/// Read a server config transaction snapshot while holding the same local and -/// distributed write locks used by every other server-config writer. Internal -/// reads and the later conditional write use no-lock object I/O; the guards -/// remain live until the snapshot is dropped. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ServerConfigSaveResult { + persisted: bool, + generation: Option, +} + +impl ServerConfigSaveResult { + pub fn persisted(&self) -> bool { + self.persisted + } + + pub fn generation(&self) -> Option { + self.generation + } +} + +/// Read a server config transaction snapshot while holding a dedicated +/// transaction lock. The config object's normal namespace lock remains +/// available to fence reads and the conditional write at commit time. +/// The transaction guard remains live until the snapshot is dropped, +/// serializing persistence and history ordering across admin nodes. Runtime +/// state is reloaded from the durable object after this guard is released. pub async fn read_server_config_snapshot(api: Arc) -> Result where S: ObjectIO< @@ -2071,12 +2295,10 @@ where { let config_file = server_config_path(); let local_guard = SERVER_CONFIG_LOCK.clone().write_owned().await; - let lock = api.new_ns_lock(RUSTFS_META_BUCKET, &config_file).await?; + let transaction_lock = server_config_transaction_lock_path(); + let lock = api.new_ns_lock(RUSTFS_META_BUCKET, &transaction_lock).await?; let guard = lock.get_write_lock(get_lock_acquire_timeout()).await?; - let read_options = ObjectOptions { - no_lock: true, - ..Default::default() - }; + let read_options = ObjectOptions::default(); match read_config_with_metadata_inner(api, &config_file, &read_options, true).await { Ok((raw, object_info)) => { let (config, seed) = decode_persisted_server_config_with_seed(&raw)?; @@ -2085,6 +2307,7 @@ where raw: Some(raw), seed: Some(seed), etag: object_info.etag, + generation: object_info.data_dir.filter(|generation| !generation.is_nil()), _local_guard: local_guard, _guard: guard, }) @@ -2094,6 +2317,7 @@ where raw: None, seed: None, etag: None, + generation: None, _local_guard: local_guard, _guard: guard, }), @@ -2108,6 +2332,27 @@ where /// lock, so a concurrent update or transaction lease loss cannot commit an /// unfenced overwrite. pub async fn save_server_config_snapshot(api: Arc, cfg: &Config, snapshot: &ServerConfigSnapshot) -> Result +where + S: ObjectIO< + Error = Error, + RangeSpec = HTTPRangeSpec, + HeaderMap = HeaderMap, + ObjectOptions = ObjectOptions, + ObjectInfo = ObjectInfo, + GetObjectReader = GetObjectReader, + PutObjectReader = PutObjReader, + > + NamespaceLocking, +{ + save_server_config_snapshot_with_generation(api, cfg, snapshot) + .await + .map(|result| result.persisted()) +} + +pub async fn save_server_config_snapshot_with_generation( + api: Arc, + cfg: &Config, + snapshot: &ServerConfigSnapshot, +) -> Result where S: ObjectIO< Error = Error, @@ -2126,13 +2371,19 @@ where && configs_semantically_equal(&snapshot.config, cfg) { debug!("server config unchanged and already in standard object shape, skip write"); - return Ok(false); + return Ok(ServerConfigSaveResult { + persisted: false, + generation: snapshot.generation(), + }); } let data = encode_server_config_blob(cfg, snapshot.seed.as_deref())?; if snapshot.raw.as_deref().is_some_and(|current| current == data.as_slice()) { debug!("server config bytes unchanged after encode, skip write"); - return Ok(false); + return Ok(ServerConfigSaveResult { + persisted: false, + generation: snapshot.generation(), + }); } let http_preconditions = if snapshot.raw.is_some() { @@ -2152,19 +2403,22 @@ where } }; - save_config_with_opts( + snapshot.ensure_lock_held()?; + let object_info = save_config_with_opts_and_metadata( api, &config_file, data, &ObjectOptions { max_parity: true, - no_lock: true, http_preconditions: Some(http_preconditions), ..Default::default() }, ) .await?; - Ok(true) + Ok(ServerConfigSaveResult { + persisted: true, + generation: object_info.data_dir.filter(|generation| !generation.is_nil()), + }) } /// Saves the server config while an upper layer holds the namespace write @@ -2301,8 +2555,9 @@ mod tests { use super::{ SERVER_CONFIG_LOCK, ServerConfigSnapshot, apply_dynamic_config_for_sub_sys_with, config_task_join_error, configs_semantically_equal, decode_server_config_blob, encode_server_config_blob, is_standard_object_server_config, - lookup_configs, read_config, read_config_preserve_empty, read_config_with_metadata, read_config_without_migrate, - read_server_config_snapshot, save_server_config, save_server_config_snapshot, server_config_path, storage_class_kvs_mut, + lookup_configs, new_and_save_server_config, read_config, read_config_preserve_empty, read_config_with_metadata, + read_config_without_migrate, read_server_config_snapshot, save_server_config, save_server_config_snapshot, + save_server_config_snapshot_with_generation, server_config_transaction_lock_path, storage_class_kvs_mut, }; use crate::config::{audit, heal, notify, oidc, scanner}; use crate::disk::endpoint::Endpoint; @@ -2311,7 +2566,9 @@ mod tests { use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}; use crate::runtime::sources as runtime_sources; use crate::set_disk::SetDisks; - use crate::storage_api_contracts::{admin::StorageAdminApi, namespace::NamespaceLocking as _, range::HTTPRangeSpec}; + use crate::storage_api_contracts::{ + admin::StorageAdminApi, namespace::NamespaceLocking as _, object::HTTPPreconditions, range::HTTPRangeSpec, + }; use http::HeaderMap; use rustfs_config::audit::{AUDIT_AMQP_SUB_SYS, AUDIT_KAFKA_SUB_SYS, AUDIT_MQTT_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS}; use rustfs_config::notify::{ @@ -3104,6 +3361,85 @@ mod tests { ); } + #[test] + fn root_heal_null_decodes_as_no_override_and_is_canonicalized_on_save() { + let seed = br#"{ + "version":"33", + "storageclass":{"standard":"","rrs":""}, + "heal":null, + "future_root":{"mode":"keep"}, + "openid":{"default":{ + "config_url":"https://issuer.example/.well-known/openid-configuration", + "client_id":"console", + "client_secret":"oidc-secret", + "future_provider_control":"keep" + }}, + "notify":{"webhook":{"primary":{ + "enable":true, + "endpoint":"https://notify.example/hook", + "auth_token":"notify-secret", + "future_notify_control":"keep" + }}}, + "logger":{"webhook":{"primary":{ + "enable":true, + "endpoint":"https://audit.example/hook", + "auth_token":"audit-secret", + "future_audit_control":"keep" + }}} + }"#; + + let cfg = decode_server_config_blob(seed).expect("root heal null should mean no persisted override"); + assert!(cfg.get_value(HEAL_SUB_SYS, DEFAULT_DELIMITER).is_none()); + assert!(!is_standard_object_server_config(seed)); + + let encoded = encode_server_config_blob(&cfg, Some(seed)).expect("legacy seed should canonicalize on an authorized save"); + let value: Value = serde_json::from_slice(&encoded).expect("canonical config should be valid JSON"); + assert!(value.get(HEAL_SUB_SYS).is_none()); + assert_eq!(value["future_root"]["mode"].as_str(), Some("keep")); + assert_eq!(value["openid"]["default"]["client_secret"].as_str(), Some("oidc-secret")); + assert_eq!(value["openid"]["default"]["future_provider_control"].as_str(), Some("keep")); + assert_eq!(value["notify"]["webhook"]["primary"]["auth_token"].as_str(), Some("notify-secret")); + assert_eq!(value["notify"]["webhook"]["primary"]["future_notify_control"].as_str(), Some("keep")); + assert_eq!(value["logger"]["webhook"]["primary"]["auth_token"].as_str(), Some("audit-secret")); + assert_eq!(value["logger"]["webhook"]["primary"]["future_audit_control"].as_str(), Some("keep")); + assert!(is_standard_object_server_config(&encoded)); + } + + #[test] + fn invalid_scalar_and_nested_null_config_shapes_remain_rejected() { + let invalid_sections = [ + r#""scanner":null"#, + r#""heal":"""#, + r#""heal":false"#, + r#""heal":0"#, + r#""heal":{"default":null}"#, + r#""heal":{"_":null}"#, + r#""heal":{"bitrot_cycle":null}"#, + r#""heal":[{"key":"bitrot_cycle","value":null}]"#, + ]; + + for section in invalid_sections { + let input = format!(r#"{{"version":"33","storageclass":{{"standard":"","rrs":""}},{section}}}"#); + let err = decode_server_config_blob(input.as_bytes()).expect_err("invalid scalar shape must remain rejected"); + assert!( + err.to_string().contains("expected"), + "invalid section {section} returned an unrelated error: {err}" + ); + } + } + + #[test] + fn valid_heal_object_and_kvs_array_shapes_remain_accepted() { + let empty_object = br#"{"version":"33","storageclass":{"standard":"","rrs":""},"heal":{}}"#; + let cfg = decode_server_config_blob(empty_object).expect("empty heal object should decode as no override"); + assert!(cfg.get_value(HEAL_SUB_SYS, DEFAULT_DELIMITER).is_none()); + + let kvs_array = + br#"{"version":"33","storageclass":{"standard":"","rrs":""},"heal":[{"key":"bitrot_cycle","value":"off"}]}"#; + let cfg = decode_server_config_blob(kvs_array).expect("heal KVS array should decode"); + assert!(cfg.get_value(HEAL_SUB_SYS, DEFAULT_DELIMITER).is_some()); + } + #[test] fn scanner_update_preserves_unknown_root_and_oidc_provider_fields() { let seed = br#"{ @@ -3131,6 +3467,171 @@ mod tests { assert_eq!(value["openid"]["default"]["client_id"].as_str(), Some("console")); } + #[test] + fn storageclass_reset_removes_stale_inline_block_from_seed() { + let seed = br#"{ + "version":"33", + "storageclass":{ + "standard":"EC:2", + "rrs":"EC:1", + "optimize":"availability", + "inline_block":"64KiB", + "future_storage_control":"keep" + } + }"#; + + let encoded = encode_server_config_blob(&Config::new(), Some(seed)).expect("storageclass reset should encode"); + let value: Value = serde_json::from_slice(&encoded).expect("encoded config should be valid json"); + let storageclass = value["storageclass"].as_object().expect("storageclass object"); + + assert!(storageclass.get(crate::config::storageclass::INLINE_BLOCK).is_none()); + assert_eq!(storageclass["future_storage_control"].as_str(), Some("keep")); + } + + #[test] + fn target_update_preserves_unknown_fields_without_restoring_removed_instances() { + let seed = br#"{ + "version":"33", + "storageclass":{"standard":"","rrs":""}, + "notify":{"webhook":{ + "primary":{ + "enable":true, + "endpoint":"https://notify.example/old", + "auth_token":"notify-secret", + "future_control":"keep" + }, + "removed":{"enable":true,"endpoint":"https://notify.example/removed"}, + "retained":{"enable":true,"endpoint":"https://notify.example/retained","future_control":"keep"}, + "enable":{"enable":true,"endpoint":"https://notify.example/named-enable","future_control":"keep"} + }} + }"#; + let mut cfg = decode_server_config_blob(seed).expect("target seed should decode"); + let webhook = cfg + .0 + .get_mut(NOTIFY_WEBHOOK_SUB_SYS) + .expect("notify webhook subsystem should exist"); + webhook + .get_mut("primary") + .expect("primary target should exist") + .insert(rustfs_config::WEBHOOK_ENDPOINT.to_string(), "https://notify.example/new".to_string()); + webhook.remove("removed"); + webhook.remove("retained"); + + let encoded = encode_server_config_blob(&cfg, Some(seed)).expect("target update should encode"); + let value: Value = serde_json::from_slice(&encoded).expect("encoded config should be valid json"); + let webhook = value["notify"]["webhook"].as_object().expect("webhook section"); + + assert_eq!(webhook["primary"]["endpoint"].as_str(), Some("https://notify.example/new")); + assert_eq!(webhook["primary"]["auth_token"].as_str(), Some("notify-secret")); + assert_eq!(webhook["primary"]["future_control"].as_str(), Some("keep")); + assert!(webhook.get("removed").is_none()); + assert_eq!(webhook["retained"]["future_control"].as_str(), Some("keep")); + assert!(webhook["retained"].get("enable").is_none()); + assert!(webhook["retained"].get("endpoint").is_none()); + assert_eq!(webhook["enable"]["endpoint"].as_str(), Some("https://notify.example/named-enable")); + assert_eq!(webhook["enable"]["future_control"].as_str(), Some("keep")); + } + + #[test] + fn shorthand_target_update_preserves_shape_and_unknown_nested_fields() { + let seed = br#"{ + "version":"33", + "storageclass":{"standard":"","rrs":""}, + "notify":{"webhook":{ + "enable":true, + "endpoint":"https://notify.example/old", + "future_control":{"endpoint":"leave-untouched","mode":"keep"} + }} + }"#; + let mut cfg = decode_server_config_blob(seed).expect("shorthand target should decode"); + cfg.0 + .get_mut(NOTIFY_WEBHOOK_SUB_SYS) + .and_then(|targets| targets.get_mut(DEFAULT_DELIMITER)) + .expect("default webhook target should exist") + .insert(rustfs_config::WEBHOOK_ENDPOINT.to_string(), "https://notify.example/new".to_string()); + + let encoded = encode_server_config_blob(&cfg, Some(seed)).expect("shorthand target update should encode"); + let value: Value = serde_json::from_slice(&encoded).expect("encoded config should be valid json"); + let webhook = value["notify"]["webhook"].as_object().expect("webhook shorthand object"); + + assert_eq!(webhook["endpoint"].as_str(), Some("https://notify.example/new")); + assert!(webhook.get("default").is_none()); + assert_eq!(webhook["future_control"]["endpoint"].as_str(), Some("leave-untouched")); + assert_eq!(webhook["future_control"]["mode"].as_str(), Some("keep")); + let decoded = decode_server_config_blob(&encoded).expect("updated shorthand target should remain decodable"); + assert_eq!( + decoded + .get_value(NOTIFY_WEBHOOK_SUB_SYS, DEFAULT_DELIMITER) + .expect("updated default webhook target") + .get(rustfs_config::WEBHOOK_ENDPOINT), + "https://notify.example/new" + ); + } + + #[test] + fn target_kvs_update_preserves_unknown_entries_and_attributes() { + let seed = br#"{ + "version":"33", + "storageclass":{"standard":"","rrs":""}, + "notify":{"webhook":{"primary":[ + {"key":"enable","value":"on","hidden_if_empty":false}, + {"key":"endpoint","value":"https://notify.example/old","future_attribute":"keep-endpoint"}, + {"key":"future_control","value":"keep","future_attribute":"keep-control"} + ]}} + }"#; + let mut cfg = decode_server_config_blob(seed).expect("target KVS seed should decode"); + cfg.0 + .get_mut(NOTIFY_WEBHOOK_SUB_SYS) + .and_then(|targets| targets.get_mut("primary")) + .expect("primary webhook target should exist") + .insert(rustfs_config::WEBHOOK_ENDPOINT.to_string(), "https://notify.example/new".to_string()); + + let encoded = encode_server_config_blob(&cfg, Some(seed)).expect("target KVS update should encode"); + let value: Value = serde_json::from_slice(&encoded).expect("encoded config should be valid json"); + let entries = value["notify"]["webhook"]["primary"] + .as_array() + .expect("target KVS shape should be preserved"); + let endpoint = entries + .iter() + .find(|entry| entry["key"].as_str() == Some(rustfs_config::WEBHOOK_ENDPOINT)) + .expect("endpoint entry should remain"); + let future = entries + .iter() + .find(|entry| entry["key"].as_str() == Some("future_control")) + .expect("unknown target entry should remain"); + + assert_eq!(endpoint["value"].as_str(), Some("https://notify.example/new")); + assert_eq!(endpoint["future_attribute"].as_str(), Some("keep-endpoint")); + assert_eq!(future["value"].as_str(), Some("keep")); + assert_eq!(future["future_attribute"].as_str(), Some("keep-control")); + } + + #[test] + fn target_default_alias_is_canonicalized_without_losing_unknown_fields() { + let seed = br#"{ + "version":"33", + "storageclass":{"standard":"","rrs":""}, + "notify":{"webhook":{ + "_":{"enable":false,"endpoint":"https://notify.example/alias","future_alias":"keep"}, + "default":{"enable":true,"endpoint":"https://notify.example/default","future_default":"keep"} + }} + }"#; + let cfg = decode_server_config_blob(seed).expect("dual default aliases should decode"); + let expected_endpoint = cfg + .get_value(NOTIFY_WEBHOOK_SUB_SYS, DEFAULT_DELIMITER) + .expect("default webhook target should exist") + .get(rustfs_config::WEBHOOK_ENDPOINT); + + let encoded = encode_server_config_blob(&cfg, Some(seed)).expect("default alias should canonicalize"); + let value: Value = serde_json::from_slice(&encoded).expect("encoded config should be valid json"); + let webhook = value["notify"]["webhook"].as_object().expect("webhook section"); + + assert!(webhook.get(DEFAULT_DELIMITER).is_none()); + assert_eq!(webhook["default"]["endpoint"].as_str(), Some(expected_endpoint.as_str())); + assert_eq!(webhook["default"]["future_alias"].as_str(), Some("keep")); + assert_eq!(webhook["default"]["future_default"].as_str(), Some("keep")); + } + #[test] fn test_scanner_config_changes_are_semantically_significant() { let baseline = Config::new(); @@ -4124,6 +4625,7 @@ mod tests { /// What reads of the config object currently return. enum RecoveryReadState { + Missing, Blob(Vec), QuorumError, } @@ -4136,6 +4638,7 @@ mod tests { heal_calls: AtomicUsize, write_calls: AtomicUsize, last_put_no_lock: AtomicBool, + last_put_preconditions: Mutex>, revision: AtomicUsize, drive_counts: Vec, lock_manager: Arc, @@ -4150,6 +4653,7 @@ mod tests { heal_calls: AtomicUsize::new(0), write_calls: AtomicUsize::new(0), last_put_no_lock: AtomicBool::new(false), + last_put_preconditions: Mutex::new(None), revision: AtomicUsize::new(1), drive_counts: vec![2], lock_manager: Arc::new(rustfs_lock::GlobalLockManager::new()), @@ -4206,6 +4710,7 @@ mod tests { _opts: &ObjectOptions, ) -> Result { let data = match &*self.state.lock().expect("state lock poisoned") { + RecoveryReadState::Missing => return Err(Error::ConfigNotFound), RecoveryReadState::Blob(data) => data.clone(), RecoveryReadState::QuorumError => return Err(Error::ErasureReadQuorum), }; @@ -4213,6 +4718,9 @@ mod tests { size: data.len() as i64, actual_size: data.len() as i64, etag: Some(format!("config-{}", self.revision.load(Ordering::SeqCst))), + data_dir: Some(uuid::Uuid::from_u128( + u128::try_from(self.revision.load(Ordering::SeqCst)).expect("test revision should fit in u128"), + )), ..Default::default() }; Ok(GetObjectReader { @@ -4231,15 +4739,19 @@ mod tests { opts: &ObjectOptions, ) -> Result { let current_etag = format!("config-{}", self.revision.load(Ordering::SeqCst)); + let object_exists = matches!(&*self.state.lock().expect("state lock poisoned"), RecoveryReadState::Blob(_)); if let Some(preconditions) = &opts.http_preconditions - && (preconditions.if_match_value().is_some_and(|etag| etag != current_etag) - || preconditions.if_none_match_value() == Some("*")) + && (preconditions + .if_match_value() + .is_some_and(|etag| !object_exists || etag != current_etag) + || (object_exists && preconditions.if_none_match_value() == Some("*"))) { return Err(Error::PreconditionFailed); } let mut body = Vec::new(); data.stream.read_to_end(&mut body).await?; self.last_put_no_lock.store(opts.no_lock, Ordering::SeqCst); + *self.last_put_preconditions.lock().expect("preconditions lock poisoned") = opts.http_preconditions.clone(); self.write_calls.fetch_add(1, Ordering::SeqCst); *self.state.lock().expect("state lock poisoned") = RecoveryReadState::Blob(body.clone()); let revision = self.revision.fetch_add(1, Ordering::SeqCst) + 1; @@ -4247,6 +4759,7 @@ mod tests { size: i64::try_from(body.len()).expect("test config should fit in i64"), actual_size: i64::try_from(body.len()).expect("test config should fit in i64"), etag: Some(format!("config-{revision}")), + data_dir: Some(uuid::Uuid::from_u128(u128::try_from(revision).expect("test revision should fit in u128"))), ..Default::default() }) } @@ -4284,10 +4797,18 @@ mod tests { .expect("scanner-only config change should be persisted"); assert_eq!(store.write_calls.load(Ordering::SeqCst), 1); - assert!(store.last_put_no_lock.load(Ordering::SeqCst)); + assert!(!store.last_put_no_lock.load(Ordering::SeqCst)); + let preconditions = store + .last_put_preconditions + .lock() + .expect("preconditions lock poisoned") + .clone() + .expect("existing config update must be conditional"); + assert_eq!(preconditions.if_match_value(), Some("config-1")); + assert_eq!(preconditions.if_none_match_value(), None); assert_eq!( store.lock_resources.lock().expect("lock resources mutex poisoned").as_slice(), - &[server_config_path()] + &[server_config_transaction_lock_path()] ); let decoded = read_config_without_migrate(store) .await @@ -4301,6 +4822,73 @@ mod tests { ); } + #[tokio::test] + async fn server_config_snapshot_save_returns_committed_generation() { + let baseline = encode_server_config_blob(&Config::new(), None).expect("baseline config should encode"); + let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(baseline), None)); + let snapshot = read_server_config_snapshot(store.clone()) + .await + .expect("server config snapshot"); + + let result = save_server_config_snapshot_with_generation(store, &config_with_scanner_cycle("61"), &snapshot) + .await + .expect("conditional config save"); + + assert!(result.persisted()); + assert_eq!(result.generation(), Some(uuid::Uuid::from_u128(2))); + } + + #[tokio::test] + async fn missing_server_config_is_created_with_if_none_match() { + let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Missing, None)); + let cfg = config_with_scanner_cycle("61"); + + save_server_config(store.clone(), &cfg) + .await + .expect("missing config should be created conditionally"); + + assert_eq!(store.write_calls.load(Ordering::SeqCst), 1); + assert!(!store.last_put_no_lock.load(Ordering::SeqCst)); + let preconditions = store + .last_put_preconditions + .lock() + .expect("preconditions lock poisoned") + .clone() + .expect("missing config create must be conditional"); + assert_eq!(preconditions.if_match_value(), None); + assert_eq!(preconditions.if_none_match_value(), Some("*")); + let persisted = read_config_without_migrate(store) + .await + .expect("created config should reload"); + assert_eq!( + persisted + .get_value(SCANNER_SUB_SYS, DEFAULT_DELIMITER) + .expect("persisted scanner config") + .get(SCANNER_CYCLE), + "61" + ); + } + + #[tokio::test] + async fn missing_config_initialization_recheck_preserves_concurrent_config() { + let existing = config_with_scanner_cycle("73"); + let baseline = encode_server_config_blob(&existing, None).expect("existing config should encode"); + let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(baseline), None)); + + let observed = new_and_save_server_config(store.clone()) + .await + .expect("initialization recheck should return the config created by another writer"); + + assert_eq!(store.write_calls.load(Ordering::SeqCst), 0); + assert_eq!( + observed + .get_value(SCANNER_SUB_SYS, DEFAULT_DELIMITER) + .expect("concurrent scanner config") + .get(SCANNER_CYCLE), + "73" + ); + } + #[tokio::test] async fn stale_server_config_snapshot_cannot_overwrite_newer_update() { let baseline = encode_server_config_blob(&Config::new(), None).expect("baseline config should encode"); @@ -4357,7 +4945,7 @@ mod tests { let lock = rustfs_lock::NamespaceLock::new("server-config-lease-loss".to_string(), client.clone()); let guard = lock .lock_guard( - rustfs_lock::ObjectKey::new(crate::disk::RUSTFS_META_BUCKET, server_config_path()), + rustfs_lock::ObjectKey::new(crate::disk::RUSTFS_META_BUCKET, server_config_transaction_lock_path()), "server-config-lease-loss", std::time::Duration::from_secs(1), std::time::Duration::from_millis(120), @@ -4372,6 +4960,7 @@ mod tests { raw: Some(baseline.clone()), seed: None, etag: Some("config-0".to_string()), + generation: Some(uuid::Uuid::from_u128(1)), _local_guard: local_guard, _guard: guard, }; diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 6ff7488a2..2a6e3895a 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -4714,6 +4714,7 @@ mod tests { use crate::layout::endpoints::SetupType; use crate::object_api::BLOCK_SIZE_V2; use crate::object_api::ObjectInfo; + use crate::set_disk::core::io_primitives::rename_fanout_barrier; use crate::storage_api_contracts::{ heal::HealOperations as _, lifecycle::TransitionedObject, list::ListOperations as _, multipart::CompletePart, namespace::NamespaceLocking as _, object::ObjectIO as _, object::ObjectOperations as _, @@ -9565,6 +9566,15 @@ mod tests { make_local_bucket_test_set_disks_with_drive_count(2).await } + fn assert_exclusive_object_lock_held(set_disks: &SetDisks, bucket: &str, object: &str) { + let lock = set_disks + .local_lock_manager_for_test() + .get_lock_info(&ObjectKey::new(bucket, object)) + .expect("object lock should be visible while rename is paused"); + assert!(matches!(lock.mode, rustfs_lock::LockMode::Exclusive)); + assert_eq!(lock.owner.as_ref(), set_disks.locker_owner.as_str()); + } + async fn make_local_bucket_test_set_disks_with_drive_count(drive_count: usize) -> Arc { let format = FormatV3::new(1, drive_count); let mut endpoints = Vec::new(); @@ -9599,7 +9609,9 @@ mod tests { disks.push(Some(disk)); } - let set_disks = SetDisks::new( + let instance_ctx = Arc::new(InstanceContext::new()); + instance_ctx.update_erasure_type(SetupType::Erasure).await; + let set_disks = SetDisks::new_with_instance_ctx( "test-owner".to_string(), Arc::new(RwLock::new(disks)), drive_count, @@ -9609,6 +9621,7 @@ mod tests { endpoints, format, Vec::new(), + instance_ctx, ) .await; set_disks.set_test_storage_class_config( @@ -10262,6 +10275,216 @@ mod tests { )); } + #[tokio::test] + async fn conditional_replace_holds_object_lock_through_rename() { + let set_disks = make_local_bucket_test_set_disks().await; + let bucket = "bucket-conditional-replace-fence"; + let object = "config/conditional-replace.json"; + set_disks + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + + let mut initial_reader = PutObjReader::from_vec(b"initial config".to_vec()); + let initial = set_disks + .put_object( + bucket, + object, + &mut initial_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("initial config should be written"); + let initial_etag = initial.etag.expect("initial config should have an ETag"); + + let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + let writer_store = set_disks.clone(); + let expected_etag = initial_etag.clone(); + let writer = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(b"replacement config".to_vec()); + writer_store + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + preserve_etag: Some("replacement-etag".to_string()), + http_preconditions: Some(HTTPPreconditions { + if_match: Some(expected_etag), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + }); + + tokio::time::timeout(std::time::Duration::from_secs(30), barrier.wait_until_paused()) + .await + .expect("conditional replace should reach the rename barrier"); + assert_exclusive_object_lock_held(&set_disks, bucket, object); + barrier.release(); + writer + .await + .expect("conditional writer task should finish") + .expect("matching conditional replace should commit"); + assert!( + set_disks + .local_lock_manager_for_test() + .get_lock_info(&ObjectKey::new(bucket, object)) + .is_none(), + "conditional replace should release the object lock after commit" + ); + let contender = set_disks + .new_ns_lock(bucket, object) + .await + .expect("contender namespace lock should be created"); + let contender_guard = contender + .get_write_lock(std::time::Duration::from_secs(30)) + .await + .expect("contender should acquire after conditional replace commits"); + drop(contender_guard); + + let mut stale_reader = PutObjReader::from_vec(b"stale config".to_vec()); + let err = set_disks + .put_object( + bucket, + object, + &mut stale_reader, + &ObjectOptions { + http_preconditions: Some(HTTPPreconditions { + if_match: Some(initial_etag), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + .expect_err("the old ETag must fail after the fenced replacement commits"); + assert_eq!(err, StorageError::PreconditionFailed); + } + + #[tokio::test] + async fn repeated_body_write_keeps_etag_but_changes_data_dir_generation() { + let set_disks = make_local_bucket_test_set_disks().await; + let bucket = "bucket-write-generation"; + let object = "config/write-generation.json"; + let body = b"identical config body".to_vec(); + set_disks + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + + let mut first_reader = PutObjReader::from_vec(body.clone()); + let first = set_disks + .put_object( + bucket, + object, + &mut first_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("first config body should be written"); + let mut second_reader = PutObjReader::from_vec(body); + let second = set_disks + .put_object( + bucket, + object, + &mut second_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("identical config body should be rewritten"); + + assert_eq!(first.etag, second.etag, "content ETag should expose the ABA collision"); + assert_ne!(first.data_dir, second.data_dir, "each committed body write needs a unique generation"); + assert!(first.data_dir.is_some() && second.data_dir.is_some()); + } + + #[tokio::test] + async fn conditional_create_holds_object_lock_through_rename() { + let set_disks = make_local_bucket_test_set_disks().await; + let bucket = "bucket-conditional-create-fence"; + let object = "config/conditional-create.json"; + set_disks + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + + let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + let writer_store = set_disks.clone(); + let writer = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(b"created config".to_vec()); + writer_store + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + http_preconditions: Some(HTTPPreconditions { + if_none_match: Some("*".to_string()), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + }); + + tokio::time::timeout(std::time::Duration::from_secs(30), barrier.wait_until_paused()) + .await + .expect("conditional create should reach the rename barrier"); + assert_exclusive_object_lock_held(&set_disks, bucket, object); + barrier.release(); + writer + .await + .expect("conditional writer task should finish") + .expect("first conditional create should commit"); + assert!( + set_disks + .local_lock_manager_for_test() + .get_lock_info(&ObjectKey::new(bucket, object)) + .is_none(), + "conditional create should release the object lock after commit" + ); + let contender = set_disks + .new_ns_lock(bucket, object) + .await + .expect("contender namespace lock should be created"); + let contender_guard = contender + .get_write_lock(std::time::Duration::from_secs(30)) + .await + .expect("contender should acquire after conditional create commits"); + drop(contender_guard); + + let mut duplicate_reader = PutObjReader::from_vec(b"duplicate config".to_vec()); + let err = set_disks + .put_object( + bucket, + object, + &mut duplicate_reader, + &ObjectOptions { + http_preconditions: Some(HTTPPreconditions { + if_none_match: Some("*".to_string()), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + .expect_err("a second create-only write must not replace the committed config"); + assert_eq!(err, StorageError::PreconditionFailed); + } + #[tokio::test] async fn set_level_if_none_match_fails_closed_without_read_quorum() { let set_disks = make_local_bucket_test_set_disks_with_drive_count(4).await; diff --git a/crates/notify/src/config_manager.rs b/crates/notify/src/config_manager.rs index 91df663d0..617a7b186 100644 --- a/crates/notify/src/config_manager.rs +++ b/crates/notify/src/config_manager.rs @@ -19,7 +19,7 @@ use crate::{ resolve_notify_object_store_handle, rule_engine::NotifyRuleEngine, runtime_facade::NotifyRuntimeFacade, - with_notify_server_config_read_lock, with_notify_server_config_write_lock, + with_notify_server_config_read_lock, }; use rustfs_config::notify::{ NOTIFY_AMQP_SUB_SYS, NOTIFY_KAFKA_SUB_SYS, NOTIFY_MQTT_SUB_SYS, NOTIFY_MYSQL_SUB_SYS, NOTIFY_NATS_SUB_SYS, @@ -41,12 +41,32 @@ enum NotifyConfigStoreError { StorageNotAvailable, Read(String), Save(String), + Converge { persisted: bool, error: String }, +} + +fn notification_convergence_error(persisted: bool, error: impl std::fmt::Display) -> NotificationError { + let durable_state = if persisted { "persisted" } else { "unchanged" }; + NotificationError::Configuration(format!( + "configuration was {durable_state} but notification runtime convergence failed: {error}" + )) +} + +async fn supervise_config_update( + mutation: impl std::future::Future> + Send + 'static, +) -> Result +where + T: Send + 'static, +{ + tokio::spawn(mutation).await.map_err(|error| { + let outcome = if error.is_cancelled() { "cancelled" } else { "panicked" }; + NotifyConfigStoreError::Lock(format!("notify config update task {outcome}")) + })? } async fn update_server_config( - modifier: F, + mut modifier: F, lifecycle: NotifyLifecycleCoordinator, -) -> Result, NotifyConfigStoreError> +) -> Result, NotifyConfigStoreError> where F: FnMut(&mut Config) -> bool + Send + 'static, { @@ -54,51 +74,31 @@ where return Err(NotifyConfigStoreError::StorageNotAvailable); }; - let store_for_read = store.clone(); - let store_for_save = store.clone(); - with_notify_server_config_write_lock(store, move || { - read_modify_write( - modifier, - move || async move { - crate::read_notify_server_config_without_migrate_no_lock(store_for_read) - .await - .map_err(NotifyConfigStoreError::Read) - }, - move |config| async move { - crate::save_notify_server_config_no_lock(store_for_save, &config) - .await - .map_err(NotifyConfigStoreError::Save) - }, - move |config| lifecycle.update_config(config), - ) + supervise_config_update(async move { + let snapshot = crate::read_notify_server_config_snapshot(store.clone()) + .await + .map_err(NotifyConfigStoreError::Read)?; + let mut config = snapshot.config.clone(); + if !modifier(&mut config) { + return Ok(None); + } + + let persisted = crate::save_notify_server_config_snapshot(store.clone(), &config, &snapshot) + .await + .map_err(NotifyConfigStoreError::Save)?; + drop(snapshot); + + let read_store = store.clone(); + with_notify_server_config_read_lock(store, move || async move { + let latest = crate::read_existing_notify_server_config_no_lock(read_store) + .await + .map_err(|error| NotifyConfigStoreError::Converge { persisted, error })?; + Ok::<_, NotifyConfigStoreError>(Some((lifecycle.update_config(latest), persisted))) + }) + .await + .map_err(|error| NotifyConfigStoreError::Converge { persisted, error })? }) .await - .map_err(NotifyConfigStoreError::Lock)? -} - -async fn read_modify_write( - mut modifier: F, - read: R, - save: S, - publish: P, -) -> Result, NotifyConfigStoreError> -where - F: FnMut(&mut Config) -> bool, - R: FnOnce() -> RFut, - RFut: std::future::Future>, - S: FnOnce(Config) -> SFut, - SFut: std::future::Future>, - P: FnOnce(Config) -> T, -{ - let mut new_config = read().await?; - - if !modifier(&mut new_config) { - return Ok(None); - } - - save(new_config.clone()).await?; - - Ok(Some(publish(new_config))) } pub(crate) fn notify_configuration_hint() -> String { @@ -345,16 +345,18 @@ impl NotifyConfigManager { if self.lifecycle.state() == NotificationRuntimeState::Terminated { return Err(NotificationError::Initialization("Notification runtime has terminated".to_string())); } - let Some(transition) = update_server_config(modifier, self.lifecycle.clone()) - .await - .map_err(|err| match err { - NotifyConfigStoreError::Lock(err) => NotificationError::StorageNotAvailable(err), - NotifyConfigStoreError::StorageNotAvailable => NotificationError::StorageNotAvailable( - "Failed to save target configuration: server storage not initialized".to_string(), - ), - NotifyConfigStoreError::Read(err) => NotificationError::ReadConfig(err), - NotifyConfigStoreError::Save(err) => NotificationError::SaveConfig(err), - })? + let Some((transition, persisted)) = + update_server_config(modifier, self.lifecycle.clone()) + .await + .map_err(|err| match err { + NotifyConfigStoreError::Lock(err) => NotificationError::StorageNotAvailable(err), + NotifyConfigStoreError::StorageNotAvailable => NotificationError::StorageNotAvailable( + "Failed to save target configuration: server storage not initialized".to_string(), + ), + NotifyConfigStoreError::Read(err) => NotificationError::ReadConfig(err), + NotifyConfigStoreError::Save(err) => NotificationError::SaveConfig(err), + NotifyConfigStoreError::Converge { persisted, error } => notification_convergence_error(persisted, error), + })? else { debug!( event = EVENT_NOTIFY_CONFIG_UPDATE, @@ -367,23 +369,53 @@ impl NotifyConfigManager { return Ok(()); }; + let result = if persisted { "updated" } else { "unchanged" }; info!( event = EVENT_NOTIFY_CONFIG_UPDATE, component = LOG_COMPONENT_NOTIFY, subsystem = LOG_SUBSYSTEM_CONFIG, action = "reload_if_changed", - result = "updated", + result, "notify config update" ); - transition.wait().await + transition + .wait() + .await + .map_err(|err| notification_convergence_error(persisted, err)) } } #[cfg(test)] mod tests { - use super::{NotifyConfigManager, NotifyConfigStoreError, read_modify_write, runtime_target_id_for_subsystem}; + use super::{ + NotifyConfigManager, NotifyConfigStoreError, notification_convergence_error, runtime_target_id_for_subsystem, + supervise_config_update, + }; use crate::rules::RulesMap; use crate::{NotificationError, NotificationRuntimeState}; + + #[tokio::test] + async fn supervised_config_update_survives_waiter_cancellation() { + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let (completed_tx, completed_rx) = tokio::sync::oneshot::channel(); + + let waiter = tokio::spawn(async move { + supervise_config_update(async move { + let _ = started_tx.send(()); + let _ = release_rx.await; + let _ = completed_tx.send(()); + Ok::<_, NotifyConfigStoreError>(()) + }) + .await + }); + + started_rx.await.expect("config update should start"); + waiter.abort(); + release_tx.send(()).expect("config update should be released"); + let completed = tokio::time::timeout(std::time::Duration::from_secs(30), completed_rx).await; + assert!(matches!(completed, Ok(Ok(()))), "detached config update should complete"); + } use crate::{ integration::NotificationMetrics, notifier::EventNotifier, registry::TargetRegistry, rule_engine::NotifyRuleEngine, runtime_facade::NotifyRuntimeFacade, @@ -392,14 +424,11 @@ mod tests { NOTIFY_AMQP_SUB_SYS, NOTIFY_KAFKA_SUB_SYS, NOTIFY_MQTT_SUB_SYS, NOTIFY_NATS_SUB_SYS, NOTIFY_POSTGRES_SUB_SYS, NOTIFY_PULSAR_SUB_SYS, NOTIFY_REDIS_SUB_SYS, NOTIFY_WEBHOOK_SUB_SYS, }; - use rustfs_config::server_config::{Config, KVS}; + use rustfs_config::server_config::Config; use rustfs_s3_types::EventName; use rustfs_targets::ReplayWorkerManager; use rustfs_targets::arn::TargetID; - use std::sync::{ - Arc, - atomic::{AtomicBool, Ordering}, - }; + use std::sync::Arc; use tokio::sync::{RwLock, Semaphore}; fn build_manager() -> NotifyConfigManager { @@ -463,38 +492,6 @@ mod tests { assert!(matches!(manager.lifecycle().state(), NotificationRuntimeState::TargetsEnabled { .. })); } - #[tokio::test] - async fn read_modify_write_publishes_only_after_save() { - let saved = Arc::new(AtomicBool::new(false)); - let saved_by_writer = saved.clone(); - let observed_by_publisher = saved.clone(); - - read_modify_write( - |config| { - config - .0 - .entry(NOTIFY_WEBHOOK_SUB_SYS.to_string()) - .or_default() - .insert("primary".to_string(), KVS::default()); - true - }, - || async { Ok::<_, NotifyConfigStoreError>(Config::default()) }, - move |_config| async move { - saved_by_writer.store(true, Ordering::Release); - Ok::<_, NotifyConfigStoreError>(()) - }, - move |_config| { - assert!( - observed_by_publisher.load(Ordering::Acquire), - "publication must observe the completed save" - ); - }, - ) - .await - .expect("read-modify-write should succeed") - .expect("changed config should publish"); - } - #[test] fn runtime_target_id_for_subsystem_maps_notify_webhook_to_runtime_type() { let target_id = runtime_target_id_for_subsystem(NOTIFY_WEBHOOK_SUB_SYS, "Primary"); @@ -550,4 +547,16 @@ mod tests { assert_eq!(target_id.id, "audittrail"); assert_eq!(target_id.name, "postgres"); } + + #[test] + fn convergence_error_reports_durable_write_state() { + for (persisted, expected) in [(true, "configuration was persisted"), (false, "configuration was unchanged")] { + let error = notification_convergence_error(persisted, "injected failure"); + let NotificationError::Configuration(message) = error else { + panic!("convergence error should be a configuration error"); + }; + assert!(message.contains(expected), "unexpected convergence error: {message}"); + assert!(message.contains("injected failure")); + } + } } diff --git a/crates/notify/src/lib.rs b/crates/notify/src/lib.rs index df93c5a48..dcc421d88 100644 --- a/crates/notify/src/lib.rs +++ b/crates/notify/src/lib.rs @@ -57,7 +57,6 @@ pub use services::NotifyServices; pub use status_view::NotifyStatusView; pub use storage_api::NotifyStore; pub(crate) use storage_api::crate_boundary::{ - read_existing_notify_server_config_no_lock, read_notify_server_config_without_migrate_no_lock, - resolve_notify_object_store_handle, save_notify_server_config_no_lock, with_notify_server_config_read_lock, - with_notify_server_config_write_lock, + read_existing_notify_server_config_no_lock, read_notify_server_config_snapshot, resolve_notify_object_store_handle, + save_notify_server_config_snapshot, with_notify_server_config_read_lock, }; diff --git a/crates/notify/src/storage_api.rs b/crates/notify/src/storage_api.rs index 9f4a44368..ba1ac98ea 100644 --- a/crates/notify/src/storage_api.rs +++ b/crates/notify/src/storage_api.rs @@ -15,11 +15,10 @@ use std::sync::Arc; use rustfs_ecstore::api::config::com::{ - read_config_without_migrate_no_lock as read_notify_config_without_migrate_from_backend_no_lock, read_existing_server_config_no_lock as read_existing_notify_config_from_backend_no_lock, - save_server_config_no_lock as save_notify_server_config_to_backend_no_lock, + read_server_config_snapshot as read_notify_server_config_snapshot_from_backend, + save_server_config_snapshot as save_notify_server_config_snapshot_to_backend, with_server_config_read_lock as with_notify_server_config_read_lock_from_backend, - with_server_config_write_lock as with_notify_server_config_write_lock_from_backend, }; use rustfs_ecstore::api::runtime::object_store_handle as resolve_notify_object_store_handle_from_backend; pub use rustfs_ecstore::api::storage::ECStore as NotifyStore; @@ -28,10 +27,10 @@ pub(crate) fn resolve_notify_object_store_handle() -> Option> { resolve_notify_object_store_handle_from_backend() } -pub(crate) async fn read_notify_server_config_without_migrate_no_lock( - store: Arc, -) -> Result { - read_notify_config_without_migrate_from_backend_no_lock(store) +pub(crate) type NotifyServerConfigSnapshot = rustfs_ecstore::api::config::com::ServerConfigSnapshot; + +pub(crate) async fn read_notify_server_config_snapshot(store: Arc) -> Result { + read_notify_server_config_snapshot_from_backend(store) .await .map_err(|err| err.to_string()) } @@ -44,22 +43,12 @@ pub(crate) async fn read_existing_notify_server_config_no_lock( .map_err(|err| err.to_string()) } -pub(crate) async fn save_notify_server_config_no_lock( +pub(crate) async fn save_notify_server_config_snapshot( store: Arc, config: &rustfs_config::server_config::Config, -) -> Result<(), String> { - save_notify_server_config_to_backend_no_lock(store, config) - .await - .map_err(|err| err.to_string()) -} - -pub(crate) async fn with_notify_server_config_write_lock(store: Arc, operation: F) -> Result -where - F: FnOnce() -> Fut + Send + 'static, - Fut: std::future::Future + Send + 'static, - T: Send + 'static, -{ - with_notify_server_config_write_lock_from_backend(store, operation) + snapshot: &NotifyServerConfigSnapshot, +) -> Result { + save_notify_server_config_snapshot_to_backend(store, config, snapshot) .await .map_err(|err| err.to_string()) } @@ -77,8 +66,7 @@ where pub(crate) mod crate_boundary { pub(crate) use super::{ - read_existing_notify_server_config_no_lock, read_notify_server_config_without_migrate_no_lock, - resolve_notify_object_store_handle, save_notify_server_config_no_lock, with_notify_server_config_read_lock, - with_notify_server_config_write_lock, + read_existing_notify_server_config_no_lock, read_notify_server_config_snapshot, resolve_notify_object_store_handle, + save_notify_server_config_snapshot, with_notify_server_config_read_lock, }; } diff --git a/rustfs/src/admin/handlers/audit_runtime_config.rs b/rustfs/src/admin/handlers/audit_runtime_config.rs index b4d7e5f9f..1055d3bf3 100644 --- a/rustfs/src/admin/handlers/audit_runtime_config.rs +++ b/rustfs/src/admin/handlers/audit_runtime_config.rs @@ -15,13 +15,15 @@ use crate::admin::handlers::supervise_admin_mutation; use crate::admin::handlers::target_descriptor::AdminTargetSpec; use crate::admin::runtime_sources::{AppContext, current_app_context, current_object_store_handle_for_context}; +use crate::admin::service::config::with_runtime_config_reload_lock; use crate::admin::storage_api::config::{ - read_admin_config_without_migrate, read_admin_server_config_snapshot, save_admin_server_config_snapshot, + read_admin_config_without_migrate, read_admin_server_config_snapshot, read_existing_admin_server_config_no_lock, + save_admin_server_config_snapshot, with_admin_server_config_read_lock, }; use rustfs_audit::{audit_system, start_audit_system as start_global_audit_system, system::AuditSystemState}; use rustfs_config::DEFAULT_DELIMITER; use rustfs_config::server_config::Config; -use s3s::{S3Result, s3_error}; +use s3s::{S3Error, S3Result, s3_error}; use tracing::warn; pub(crate) async fn load_server_config_from_store_for_context(context: Option<&AppContext>) -> S3Result { @@ -48,6 +50,14 @@ fn has_any_audit_targets(specs: &[AdminTargetSpec], config: &Config) -> bool { }) } +fn audit_config_convergence_error(persisted: bool, error: impl std::fmt::Display) -> S3Error { + if persisted { + s3_error!(InternalError, "audit config persisted but runtime convergence failed: {}", error) + } else { + s3_error!(InternalError, "audit config unchanged but runtime convergence failed: {}", error) + } +} + pub(crate) async fn apply_audit_runtime_config(specs: &[AdminTargetSpec], config: Config) -> S3Result<()> { let has_targets = has_any_audit_targets(specs, &config); @@ -107,15 +117,23 @@ where return Ok(()); } - save_admin_server_config_snapshot(store, &config, &snapshot) + let persisted = save_admin_server_config_snapshot(store.clone(), &config, &snapshot) .await - .map(|_| ()) - .map_err(|e| s3_error!(InternalError, "failed to save audit config: {}", e))?; + .map_err(|e| s3_error!(InternalError, "failed to save audit config: {}", e))? + .persisted(); + drop(snapshot); - // Keep persistence and runtime publication in one detached, serialized - // mutation. Otherwise a cancelled caller or two concurrent updates can - // leave the persisted config and active audit generation disagreeing. - apply_audit_runtime_config(&specs, config).await + let read_store = store.clone(); + with_runtime_config_reload_lock(async move { + let latest = with_admin_server_config_read_lock(store, move || read_existing_admin_server_config_no_lock(read_store)) + .await + .map_err(|e| s3_error!(InternalError, "failed to lock server config for audit reload: {}", e))? + .map_err(|e| s3_error!(InternalError, "failed to read latest server config for audit reload: {}", e))?; + + apply_audit_runtime_config(&specs, latest).await + }) + .await + .map_err(|e| audit_config_convergence_error(persisted, e)) }) .await } @@ -164,3 +182,154 @@ pub(crate) async fn remove_audit_target_config(specs: &[AdminTargetSpec], subsys }) .await } + +#[cfg(test)] +mod tests { + use super::*; + use crate::admin::handlers::target_descriptor::admin_target_spec_from_builtin; + use crate::admin::runtime_sources::{IamInterface, KmsInterface}; + use crate::admin::storage_api::config::save_admin_server_config; + use rustfs_config::audit::AUDIT_WEBHOOK_SUB_SYS; + use rustfs_config::server_config::KVS; + use rustfs_config::{ENABLE_KEY, EnableState, SCANNER_CYCLE, SCANNER_SUB_SYS, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR}; + use rustfs_iam::{store::object::ObjectStore, sys::IamSys}; + use rustfs_kms::KmsServiceManager; + use rustfs_targets::catalog::builtin::builtin_audit_target_admin_descriptors; + use std::sync::Arc; + use std::time::Duration; + use tempfile::TempDir; + + struct TestIam; + + impl IamInterface for TestIam { + fn handle(&self) -> Arc> { + unreachable!("audit config tests do not use IAM") + } + + fn is_ready(&self) -> bool { + false + } + } + + struct TestKms; + + impl KmsInterface for TestKms { + fn handle(&self) -> Arc { + Arc::new(KmsServiceManager::new()) + } + } + + fn audit_specs() -> Vec { + builtin_audit_target_admin_descriptors() + .into_iter() + .map(|descriptor| admin_target_spec_from_builtin(&descriptor)) + .collect() + } + + async fn wait_for_persisted_target(store: Arc, subsystem: &str, target: &str) { + tokio::time::timeout(Duration::from_secs(30), async { + let mut poll = tokio::time::interval(Duration::from_millis(10)); + loop { + poll.tick().await; + let config = read_admin_config_without_migrate(store.clone()) + .await + .expect("read persisted server config"); + if config.0.get(subsystem).is_some_and(|targets| targets.contains_key(target)) { + return; + } + } + }) + .await + .expect("config mutation should become durable"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial_test::serial] + async fn audit_reload_reads_latest_durable_config_after_releasing_write_snapshot() { + let temp_dir = TempDir::new().expect("audit config temp dir"); + let env = rustfs_test_utils::TestECStoreEnv::builder() + .base_dir(temp_dir.path()) + .disk_count(1) + .init_bucket_metadata(false) + .build() + .await; + save_admin_server_config(env.ecstore.clone(), &Config::new()) + .await + .expect("persist baseline server config"); + let context = Arc::new(AppContext::new(env.ecstore.clone(), Arc::new(TestIam), Arc::new(TestKms))); + + let (locked_tx, locked_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let blocker = tokio::spawn(async move { + with_runtime_config_reload_lock(async move { + locked_tx.send(()).expect("signal runtime reload lock acquisition"); + release_rx.await.expect("release runtime reload lock"); + Ok(()) + }) + .await + .expect("runtime reload lock blocker"); + }); + locked_rx.await.expect("runtime reload lock should be held"); + + let older_context = context.clone(); + let older = tokio::spawn(async move { + let specs = audit_specs(); + update_audit_config_and_reload_for_context(Some(older_context.as_ref()), &specs, |config| { + config + .0 + .entry(SCANNER_SUB_SYS.to_string()) + .or_default() + .entry(DEFAULT_DELIMITER.to_string()) + .or_insert_with(KVS::new) + .insert(SCANNER_CYCLE.to_string(), "15s".to_string()); + true + }) + .await + }); + wait_for_persisted_target(env.ecstore.clone(), SCANNER_SUB_SYS, DEFAULT_DELIMITER).await; + assert!(!older.is_finished(), "audit runtime publication must wait for the shared reload lock"); + + let snapshot = read_admin_server_config_snapshot(env.ecstore.clone()) + .await + .expect("read newer server config snapshot"); + let mut latest = snapshot.config.clone(); + let mut latest_target = KVS::new(); + latest_target.insert(ENABLE_KEY.to_string(), EnableState::On.to_string()); + latest_target.insert(WEBHOOK_ENDPOINT.to_string(), "https://audit.invalid/hook".to_string()); + latest_target.insert( + WEBHOOK_QUEUE_DIR.to_string(), + temp_dir.path().join("audit-queue").to_string_lossy().into_owned(), + ); + latest + .0 + .entry(AUDIT_WEBHOOK_SUB_SYS.to_string()) + .or_default() + .insert("latest".to_string(), latest_target); + save_admin_server_config_snapshot(env.ecstore.clone(), &latest, &snapshot) + .await + .expect("persist newer audit config"); + drop(snapshot); + + release_tx.send(()).expect("release runtime reload blocker"); + blocker.await.expect("runtime reload blocker task"); + tokio::time::timeout(Duration::from_secs(30), older) + .await + .expect("older audit update should complete") + .expect("older audit update task should not panic") + .expect("older audit update should converge from the latest durable config"); + + let system = audit_system().expect("latest durable audit target should start the audit system"); + let targets = system.list_targets().await; + assert!(targets.iter().any(|target| target.contains("latest")), "active targets: {targets:?}"); + system.close().await.expect("audit system should stop after the test"); + } + + #[test] + fn audit_convergence_error_reports_durable_write_state() { + for (persisted, expected) in [(true, "audit config persisted"), (false, "audit config unchanged")] { + let error = audit_config_convergence_error(persisted, "injected failure"); + assert!(error.to_string().contains(expected), "unexpected convergence error: {error}"); + assert!(error.to_string().contains("injected failure")); + } + } +} diff --git a/rustfs/src/admin/handlers/config_admin.rs b/rustfs/src/admin/handlers/config_admin.rs index b9d409090..83641b017 100644 --- a/rustfs/src/admin/handlers/config_admin.rs +++ b/rustfs/src/admin/handlers/config_admin.rs @@ -17,18 +17,17 @@ use crate::admin::handlers::supervise_admin_mutation; use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::admin::runtime_sources::{ current_action_credentials, current_app_context, current_object_store_handle_for_context, current_server_config_for_context, - publish_server_config, }; use crate::admin::service::config::{ - CONFIG_WORKER_RELOAD_FAILURE_STATE, EVENT_CONFIG_WORKER_RELOAD_FAILED, FULL_CONFIG_WORKER_SUBSYSTEMS, LOG_COMPONENT_ADMIN, - LOG_SUBSYSTEM_CONFIG, PreparedRuntimeConfig, is_dynamic_config_subsystem, preflight_dynamic_config_reload, - prepare_server_config, reload_dynamic_config_runtime_state, reload_runtime_config_snapshot, signal_config_snapshot_reload, - signal_config_snapshot_reload_checked, signal_dynamic_config_reload_checked, + FULL_CONFIG_WORKER_SUBSYSTEMS, is_dynamic_config_subsystem, preflight_dynamic_config_reload, prepare_server_config, + publish_latest_runtime_config_snapshot, reload_dynamic_config_runtime_state, reload_runtime_config_snapshot, + signal_config_snapshot_reload, signal_config_snapshot_reload_checked, signal_dynamic_config_reload_checked, }; use crate::admin::storage_api::config::storageclass::{INLINE_BLOCK_ENV, OPTIMIZE_ENV, RRS_ENV, STANDARD_ENV}; use crate::admin::storage_api::config::{ - AdminServerConfigSnapshot, RUSTFS_META_BUCKET, STORAGE_CLASS_SUB_SYS, delete_admin_config, read_admin_config, - read_admin_config_without_migrate, read_admin_server_config_snapshot, save_admin_config, save_admin_server_config_snapshot, + AdminServerConfigSaveResult, AdminServerConfigSnapshot, RUSTFS_META_BUCKET, STORAGE_CLASS_SUB_SYS, delete_admin_config, + read_admin_config, read_admin_config_without_migrate, read_admin_server_config_snapshot, save_admin_config, + save_admin_server_config_snapshot, }; use crate::admin::storage_api::contract::list::ListOperations as _; use crate::admin::utils::{encode_compatible_admin_payload, is_compat_admin_request, read_compatible_admin_body}; @@ -766,7 +765,10 @@ async fn load_server_config_snapshot_from_store() -> S3Result S3Result { +async fn save_server_config_to_store( + config: &ServerConfig, + snapshot: &AdminServerConfigSnapshot, +) -> S3Result { let store = object_store()?; save_admin_server_config_snapshot(store, config, snapshot) .await @@ -987,8 +989,8 @@ fn decode_config_history_snapshot(data: &[u8]) -> S3Result { Ok(config) } -fn validate_restore_rollback_generation(current: &ServerConfig, restored: &ServerConfig) -> S3Result<()> { - if current == restored { +fn validate_restore_rollback_generation(current: Option, committed: Option) -> S3Result<()> { + if current.is_some() && current == committed { Ok(()) } else { Err(s3_error!( @@ -1004,12 +1006,12 @@ async fn save_server_config_history_snapshot(config: &ServerConfig) -> S3Result< save_server_config_history(&sealed).await } -async fn cleanup_failed_config_history_snapshot(restore_id: &str) { +async fn cleanup_config_history_snapshot(restore_id: &str) { if let Err(err) = delete_server_config_history(restore_id).await { warn!( restore_id, error = %err, - "Failed to remove config history snapshot for an uncommitted mutation" + "Failed to remove config history snapshot" ); } } @@ -1843,23 +1845,6 @@ async fn reject_lost_config_transaction(snapshot: AdminServerConfigSnapshot, )) } -fn publish_prepared_config_snapshots(config: ServerConfig, prepared: PreparedRuntimeConfig) -> S3Result<()> { - prepared.publish_storage_class()?; - publish_server_config(config); - Ok(()) -} - -/// Re-apply local mutable worker families after a full-config replacement. -/// Peers receive one full-snapshot signal after this returns; signaling each -/// family here as well would recreate audit/scanner targets twice per peer. -fn publish_notify_config_intent( - config: &ServerConfig, - sub_system: Option<&str>, -) -> Option { - (sub_system.is_none() || sub_system.is_some_and(|sub_system| NOTIFY_SUB_SYSTEMS.contains(&sub_system))) - .then(|| rustfs_notify::ensure_live_events().publish_config(config.clone())) -} - fn config_preflight_subsystems(sub_system: Option<&str>) -> Vec<&str> { if let Some(sub_system) = sub_system { return is_dynamic_config_subsystem(sub_system) @@ -1884,78 +1869,26 @@ async fn preflight_config_intent(sub_system: Option<&str>) -> S3Result<()> { Ok(()) } -async fn wait_notify_config_intent(transition: Option) -> S3Result { - let Some(transition) = transition else { - return Ok(false); - }; - transition.wait().await.map_err(|err| { - warn!(error = %err, "Failed to apply local notification config"); - s3_error!(InternalError, "failed to apply notification config") - })?; - Ok(true) -} - -async fn reload_non_notify_dynamic_subsystems() -> Vec { - let mut failures = Vec::new(); - for sub_system in FULL_CONFIG_WORKER_SUBSYSTEMS { - if NOTIFY_SUB_SYSTEMS.contains(&sub_system) { - continue; - } - if reload_dynamic_config_runtime_state(sub_system).await.is_err() { - failures.push(format!("local {sub_system}")); - warn!( - event = EVENT_CONFIG_WORKER_RELOAD_FAILED, - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_CONFIG, - config_subsystem = sub_system, - state = CONFIG_WORKER_RELOAD_FAILURE_STATE, - reason = "apply_failed", - "Published server config but failed to reload a local worker subsystem" - ); - } - } - failures -} - -fn finish_config_reconciliation(errors: Vec) -> S3Result<()> { +fn finish_config_reconciliation(errors: Vec, persisted: bool) -> S3Result<()> { if errors.is_empty() { Ok(()) - } else { + } else if persisted { Err(s3_error!( InternalError, "server config persisted but runtime convergence failed: {}", errors.join("; ") )) + } else { + Err(s3_error!( + InternalError, + "server config was unchanged but runtime convergence failed: {}", + errors.join("; ") + )) } } -async fn reconcile_targeted_config( - sub_system: Option, - storage_class_applied: bool, - notify_transition: Option, -) -> S3Result { - let mut errors = Vec::new(); - let notify_applied = notify_transition.is_some(); - if let Err(err) = wait_notify_config_intent(notify_transition).await { - warn!(error = %err, "Local notification config failed to converge"); - errors.push("local notify".to_string()); - } - - let config_applied = if notify_applied { - if let Some(sub_system) = sub_system.as_deref() - && let Err(err) = signal_dynamic_config_reload_checked(sub_system).await - { - warn!(config_subsystem = sub_system, error = %err, "Peer config reload failed"); - errors.push(format!("peer {sub_system}")); - } - true - } else if storage_class_applied { - if let Err(err) = signal_dynamic_config_reload_checked(STORAGE_CLASS_SUB_SYS).await { - warn!(error = %err, "Peer storage-class reload failed"); - errors.push(format!("peer {STORAGE_CLASS_SUB_SYS}")); - } - true - } else if let Some(sub_system) = sub_system.as_deref() +async fn reconcile_targeted_config(sub_system: Option, persisted: bool, mut errors: Vec) -> S3Result { + let config_applied = if let Some(sub_system) = sub_system.as_deref() && is_dynamic_config_subsystem(sub_system) { let config_applied = match reload_dynamic_config_runtime_state(sub_system).await { @@ -1979,21 +1912,19 @@ async fn reconcile_targeted_config( false }; - finish_config_reconciliation(errors)?; + finish_config_reconciliation(errors, persisted)?; Ok(config_applied) } -async fn reconcile_full_config(notify_transition: Option) -> S3Result<()> { - let mut errors = Vec::new(); - if let Err(err) = wait_notify_config_intent(notify_transition).await { - warn!(error = %err, "Local notification config failed to converge"); +async fn reconcile_full_config(persisted: bool, mut errors: Vec) -> S3Result<()> { + if let Err(err) = reload_dynamic_config_runtime_state(NOTIFY_WEBHOOK_SUB_SYS).await { + warn!(error = %err, "Local notification config failed to converge from durable config"); errors.push("local notify".to_string()); } if let Err(err) = signal_dynamic_config_reload_checked(STORAGE_CLASS_SUB_SYS).await { warn!(error = %err, "Peer storage-class reload failed"); errors.push(format!("peer {STORAGE_CLASS_SUB_SYS}")); } - errors.extend(reload_non_notify_dynamic_subsystems().await); if let Err(err) = signal_dynamic_config_reload_checked(NOTIFY_WEBHOOK_SUB_SYS).await { warn!(error = %err, "Peer notification config reload failed"); errors.push("peer notify".to_string()); @@ -2002,91 +1933,82 @@ async fn reconcile_full_config(notify_transition: Option, - storage_class_applied: bool, - notify_transition: Option, + persisted: bool, + committed_generation: Option, } async fn persist_server_config_transaction( config: ServerConfig, - prepared: PreparedRuntimeConfig, snapshot: AdminServerConfigSnapshot, - sub_system: Option<&str>, ) -> S3Result { snapshot.ensure_lock_held().map_err(ApiError::from).map_err(S3Error::from)?; let previous_config = snapshot.config.clone(); let history_restore_id = save_server_config_history_snapshot(&previous_config).await?; if snapshot.is_lock_lost() { - cleanup_failed_config_history_snapshot(&history_restore_id).await; + cleanup_config_history_snapshot(&history_restore_id).await; return reject_lost_config_transaction(snapshot, "before persistence").await; } - let persisted = match save_server_config_to_store(&config, &snapshot).await { - Ok(persisted) => persisted, + let save_result = match save_server_config_to_store(&config, &snapshot).await { + Ok(result) => result, Err(err) => { - cleanup_failed_config_history_snapshot(&history_restore_id).await; + cleanup_config_history_snapshot(&history_restore_id).await; return Err(err); } }; + let persisted = save_result.persisted(); + let committed_generation = save_result.generation(); let history_restore_id = if persisted { Some(history_restore_id) } else { - cleanup_failed_config_history_snapshot(&history_restore_id).await; + cleanup_config_history_snapshot(&history_restore_id).await; None }; - if snapshot.is_lock_lost() { - return reject_lost_config_transaction(snapshot, "after persistence").await; - } - - let storage_class_applied = sub_system.is_none_or(|value| value == STORAGE_CLASS_SUB_SYS); - if storage_class_applied { - if let Err(err) = publish_prepared_config_snapshots(config.clone(), prepared) { - return Err(match history_restore_id.as_deref() { - Some(restore_id) => s3_error!( - InternalError, - "config persisted but runtime publish failed; recovery snapshot restoreId={}: {}", - restore_id, - err - ), - None => err, - }); - } - } else { - publish_server_config(config.clone()); - } - let notify_transition = publish_notify_config_intent(&config, sub_system); - if snapshot.is_lock_lost() { - return reject_lost_config_transaction(snapshot, "while publishing runtime snapshots").await; - } - drop(snapshot); Ok(PersistedConfigTransaction { previous_config, history_restore_id, - storage_class_applied, - notify_transition, + persisted, + committed_generation, }) } +async fn reconcile_committed_config(sub_system: Option, persisted: bool) -> S3Result { + let mut errors = Vec::new(); + let publication_result = match sub_system.as_deref() { + Some(sub_system) => publish_latest_runtime_config_snapshot(sub_system).await, + None => reload_runtime_config_snapshot().await, + }; + if let Err(err) = publication_result { + warn!(error = %err, "Failed to publish the latest durable server config"); + errors.push("local config snapshot".to_string()); + } + + if sub_system.is_none() { + reconcile_full_config(persisted, errors).await?; + return Ok(false); + } + + reconcile_targeted_config(sub_system, persisted, errors).await +} + async fn commit_server_config_transaction( config: ServerConfig, - prepared: PreparedRuntimeConfig, snapshot: AdminServerConfigSnapshot, sub_system: Option, ) -> S3Result { - let transaction = persist_server_config_transaction(config, prepared, snapshot, sub_system.as_deref()).await?; - - if sub_system.is_none() { - reconcile_full_config(transaction.notify_transition).await?; - return Ok(false); + let transaction = persist_server_config_transaction(config, snapshot).await?; + let result = reconcile_committed_config(sub_system, transaction.persisted).await; + match (result, transaction.history_restore_id.as_deref()) { + (Err(err), Some(restore_id)) => Err(s3_error!(InternalError, "{}; recovery snapshot restoreId={}", err, restore_id)), + (result, _) => result, } - - reconcile_targeted_config(sub_system, transaction.storage_class_applied, transaction.notify_transition).await } pub struct GetConfigKVHandler {} @@ -2127,8 +2049,8 @@ impl Operation for SetConfigKVHandler { let snapshot = load_server_config_snapshot_from_store().await?; let mut config = snapshot.config.clone(); apply_set_directives(&mut config, &directives)?; - let prepared = prepare_server_config(&config, sub_system.as_deref()).await?; - commit_server_config_transaction(config, prepared, snapshot, sub_system).await + prepare_server_config(&config, sub_system.as_deref()).await?; + commit_server_config_transaction(config, snapshot, sub_system).await }) .await?; @@ -2155,8 +2077,8 @@ impl Operation for DelConfigKVHandler { let snapshot = load_server_config_snapshot_from_store().await?; let mut config = snapshot.config.clone(); apply_delete_directives(&mut config, &directives); - let prepared = prepare_server_config(&config, sub_system.as_deref()).await?; - commit_server_config_transaction(config, prepared, snapshot, sub_system).await + prepare_server_config(&config, sub_system.as_deref()).await?; + commit_server_config_transaction(config, snapshot, sub_system).await }) .await?; @@ -2239,15 +2161,19 @@ impl Operation for RestoreConfigHistoryKVHandler { supervise_admin_mutation("config mutation", async move { preflight_config_intent(None).await?; let snapshot = load_server_config_snapshot_from_store().await?; - let prepared = prepare_server_config(&config, None).await?; - let restored_config = config.clone(); - let transaction = persist_server_config_transaction(config, prepared, snapshot, None).await?; - let Err(restore_error) = reconcile_full_config(transaction.notify_transition).await else { + prepare_server_config(&config, None).await?; + let transaction = persist_server_config_transaction(config, snapshot).await?; + let Err(restore_error) = reconcile_committed_config(None, transaction.persisted).await else { return Ok(()); }; + if !transaction.persisted { + return Err(restore_error); + } + let previous_config = transaction.previous_config; let recovery_restore_id = transaction.history_restore_id; + let committed_generation = transaction.committed_generation; let recovery_reference = recovery_restore_id.as_deref().unwrap_or("not-created"); let rollback_snapshot = match load_server_config_snapshot_from_store().await { Ok(snapshot) => snapshot, @@ -2261,7 +2187,9 @@ impl Operation for RestoreConfigHistoryKVHandler { )); } }; - if let Err(rollback_error) = validate_restore_rollback_generation(&rollback_snapshot.config, &restored_config) { + if let Err(rollback_error) = + validate_restore_rollback_generation(rollback_snapshot.generation(), committed_generation) + { return Err(s3_error!( InternalError, "config restore failed: {}; automatic rollback failed: {}; recovery snapshot restoreId={}", @@ -2271,12 +2199,20 @@ impl Operation for RestoreConfigHistoryKVHandler { )); } - let rollback_prepared = prepare_server_config(&previous_config, None).await?; - rollback_snapshot + if let Err(rollback_error) = prepare_server_config(&previous_config, None).await { + return Err(s3_error!( + InternalError, + "config restore failed: {}; automatic rollback failed: {}; recovery snapshot restoreId={}", + restore_error, + rollback_error, + recovery_reference + )); + } + if let Err(rollback_error) = rollback_snapshot .ensure_lock_held() .map_err(ApiError::from) - .map_err(S3Error::from)?; - if let Err(rollback_error) = save_server_config_to_store(&previous_config, &rollback_snapshot).await { + .map_err(S3Error::from) + { return Err(s3_error!( InternalError, "config restore failed: {}; automatic rollback failed: {}; recovery snapshot restoreId={}", @@ -2285,10 +2221,9 @@ impl Operation for RestoreConfigHistoryKVHandler { recovery_reference )); } - if rollback_snapshot.is_lock_lost() { - if let Err(rollback_error) = - reject_lost_config_transaction::<()>(rollback_snapshot, "after restore rollback persistence").await - { + let rollback_persisted = match save_server_config_to_store(&previous_config, &rollback_snapshot).await { + Ok(result) => result.persisted(), + Err(rollback_error) => { return Err(s3_error!( InternalError, "config restore failed: {}; automatic rollback failed: {}; recovery snapshot restoreId={}", @@ -2297,46 +2232,20 @@ impl Operation for RestoreConfigHistoryKVHandler { recovery_reference )); } - return Err(s3_error!(InternalError, "restore rollback lock-loss handling returned unexpectedly")); - } - - if let Err(rollback_error) = publish_prepared_config_snapshots(previous_config.clone(), rollback_prepared) { - return Err(s3_error!( - InternalError, - "config restore failed: {}; automatic rollback publish failed: {}; recovery snapshot restoreId={}", - restore_error, - rollback_error, - recovery_reference - )); - } - let rollback_transition = publish_notify_config_intent(&previous_config, None); - if rollback_snapshot.is_lock_lost() { - if let Err(rollback_error) = - reject_lost_config_transaction::<()>(rollback_snapshot, "while publishing restore rollback").await - { - return Err(s3_error!( - InternalError, - "config restore failed: {}; automatic rollback failed: {}; recovery snapshot restoreId={}", - restore_error, - rollback_error, - recovery_reference - )); - } - return Err(s3_error!(InternalError, "restore rollback lock-loss handling returned unexpectedly")); - } + }; drop(rollback_snapshot); - if let Err(rollback_error) = reconcile_full_config(rollback_transition).await { + if let Err(rollback_error) = reconcile_committed_config(None, rollback_persisted).await { return Err(s3_error!( InternalError, - "config restore failed: {}; persisted rollback did not converge: {}; recovery snapshot restoreId={}", + "config restore failed: {}; automatic rollback convergence failed: {}; recovery snapshot restoreId={}", restore_error, rollback_error, recovery_reference )); } - if let Some(recovery_restore_id) = recovery_restore_id.as_deref() { - cleanup_failed_config_history_snapshot(recovery_restore_id).await; + if rollback_persisted && let Some(recovery_restore_id) = recovery_restore_id.as_deref() { + cleanup_config_history_snapshot(recovery_restore_id).await; } Err(s3_error!(InternalError, "config restore failed and was rolled back: {}", restore_error)) }) @@ -2377,8 +2286,8 @@ impl Operation for SetConfigHandler { let snapshot = load_server_config_snapshot_from_store().await?; let mut config = ServerConfig::new(); apply_set_directives(&mut config, &directives)?; - let prepared = prepare_server_config(&config, None).await?; - commit_server_config_transaction(config, prepared, snapshot, None).await?; + prepare_server_config(&config, None).await?; + commit_server_config_transaction(config, snapshot, None).await?; Ok(()) }) .await?; @@ -2408,6 +2317,23 @@ mod tests { ); } + #[test] + fn reconciliation_error_reports_whether_config_was_persisted() { + let persisted = finish_config_reconciliation(vec!["local config snapshot".to_string()], true) + .expect_err("committed config convergence failure must be reported"); + let unchanged = finish_config_reconciliation(vec!["local config snapshot".to_string()], false) + .expect_err("unchanged config convergence failure must be reported"); + + assert_eq!( + persisted.message(), + Some("server config persisted but runtime convergence failed: local config snapshot") + ); + assert_eq!( + unchanged.message(), + Some("server config was unchanged but runtime convergence failed: local config snapshot") + ); + } + #[test] fn tokenize_config_line_handles_quotes_and_escapes() { let tokens = tokenize_config_line(r#"identity_openid client_id="console app" client_secret="s3cr\"et" enable=on"#) @@ -3191,23 +3117,32 @@ notify_webhook:secondary endpoint="https://secondary.example" auth_token="second } #[test] - fn restore_rollback_generation_rejects_concurrent_config_change() { - crate::admin::storage_api::config::init_admin_config_defaults(); - let restored = ServerConfig::new(); - let mut concurrent = restored.clone(); - apply_set_directives( - &mut concurrent, - &parse_config_directives(r#"identity_openid client_id="concurrent-client""#, false).expect("parse concurrent"), - ) - .expect("apply concurrent"); + fn restore_rollback_generation_accepts_committed_write_identity() { + let committed = Uuid::from_u128(1); + validate_restore_rollback_generation(Some(committed), Some(committed)).expect("matching committed generation"); + } - validate_restore_rollback_generation(&restored, &restored).expect("unchanged restore generation"); - let error = validate_restore_rollback_generation(&concurrent, &restored) - .expect_err("concurrent change must fence automatic rollback"); + #[test] + fn restore_rollback_generation_rejects_aba_with_matching_config_content() { + let error = validate_restore_rollback_generation(Some(Uuid::from_u128(2)), Some(Uuid::from_u128(1))) + .expect_err("a later generation must fence automatic rollback even when config content matches"); assert_eq!(error.code(), &S3ErrorCode::InvalidRequest); assert!(error.to_string().contains("concurrent configuration change")); } + #[test] + fn restore_rollback_generation_rejects_missing_write_identity() { + for (current, committed) in [ + (None, Some(Uuid::from_u128(1))), + (Some(Uuid::from_u128(1)), None), + (None, None), + ] { + let error = validate_restore_rollback_generation(current, committed) + .expect_err("missing generation metadata must fence automatic rollback"); + assert_eq!(error.code(), &S3ErrorCode::InvalidRequest); + } + } + #[test] fn legacy_directive_history_is_rejected_without_being_applied_as_a_snapshot() { let error = decode_config_history_snapshot(b"identity_openid client_id=\"legacy\"") diff --git a/rustfs/src/admin/service/config.rs b/rustfs/src/admin/service/config.rs index 43e500d7c..c406dfafa 100644 --- a/rustfs/src/admin/service/config.rs +++ b/rustfs/src/admin/service/config.rs @@ -16,7 +16,9 @@ use crate::admin::runtime_sources::{ AppContext, current_app_context, current_notification_system_for_context, current_object_store_handle_for_context, publish_server_config, publish_storage_class_config, }; -use crate::admin::storage_api::config::{STORAGE_CLASS_SUB_SYS, read_admin_config_without_migrate, storageclass}; +use crate::admin::storage_api::config::{ + STORAGE_CLASS_SUB_SYS, read_existing_admin_server_config_no_lock, storageclass, with_admin_server_config_read_lock, +}; use crate::admin::storage_api::contract::admin::StorageAdminApi; use crate::admin::storage_api::runtime::ECStore; use crate::server::{ @@ -43,6 +45,25 @@ use url::Url; static RUNTIME_CONFIG_RELOAD_MUTEX: AsyncMutex<()> = AsyncMutex::const_new(()); +// Runtime publication lock order: reload mutex -> server-config local lock -> +// transaction lock -> config object lock. +pub(crate) async fn with_runtime_config_reload_lock(operation: Fut) -> S3Result +where + Fut: Future> + Send + 'static, + T: Send + 'static, +{ + let reload_guard = RUNTIME_CONFIG_RELOAD_MUTEX.lock().await; + tokio::spawn(async move { + let _reload_guard = reload_guard; + operation.await + }) + .await + .map_err(|err| { + let outcome = if err.is_cancelled() { "cancelled" } else { "panicked" }; + internal_error(format!("runtime config reload task {outcome}")) + })? +} + pub fn is_dynamic_config_subsystem(sub_system: &str) -> bool { NOTIFY_SUB_SYSTEMS.contains(&sub_system) || matches!( @@ -121,11 +142,6 @@ impl PreparedRuntimeConfig { fn publish_storage_class_for_context(self, context: Option<&AppContext>) -> S3Result<()> { self.publish_storage_class_for_context_with(context, publish_storage_class_config) } - - pub(crate) fn publish_storage_class(self) -> S3Result<()> { - let context = current_app_context(); - self.publish_storage_class_for_context(context.as_deref()) - } } fn publish_server_config_for_context(context: Option<&AppContext>, config: ServerConfig) { @@ -396,7 +412,18 @@ pub async fn apply_dynamic_config_for_subsystem(config: &ServerConfig, sub_syste } pub async fn reload_dynamic_config_runtime_state_for_context(context: Option<&AppContext>, sub_system: &str) -> S3Result<()> { - let _reload_guard = RUNTIME_CONFIG_RELOAD_MUTEX.lock().await; + let context = context.cloned(); + let sub_system = sub_system.to_owned(); + with_runtime_config_reload_lock(async move { + reload_dynamic_config_runtime_state_under_reload_lock_for_context(context.as_ref(), &sub_system).await + }) + .await +} + +async fn reload_dynamic_config_runtime_state_under_reload_lock_for_context( + context: Option<&AppContext>, + sub_system: &str, +) -> S3Result<()> { if sub_system == MODULE_SWITCHES_SIGNAL_SUBSYSTEM { let store = resolve_runtime_config_store_for_context(context)?; let notify_result = reconcile_event_notifier_from_store(store).await; @@ -414,18 +441,54 @@ pub async fn reload_dynamic_config_runtime_state_for_context(context: Option<&Ap } let store = resolve_runtime_config_store_for_context(context)?; - let config = read_admin_config_without_migrate(store).await.map_err(|err| { - warn!("peer reload_dynamic_config: failed to load server config for {sub_system}: {err}"); - internal_error(format!("failed to load server config: {err}")) - })?; + let read_store = store.clone(); + let publication_context = context.cloned(); + let publication_sub_system = sub_system.to_owned(); + let notify_config = with_admin_server_config_read_lock(store, move || async move { + let config = read_existing_admin_server_config_no_lock(read_store).await.map_err(|err| { + warn!("peer reload_dynamic_config: failed to load server config for {publication_sub_system}: {err}"); + internal_error(format!("failed to load server config: {err}")) + })?; + let prepared = prepare_server_config_for_context(publication_context.as_ref(), &config, Some(&publication_sub_system)) + .await + .map_err(|err| { + if publication_sub_system == STORAGE_CLASS_SUB_SYS { + internal_error(format!("failed to apply storage class config: {err}")) + } else { + err + } + })?; - if matches!(sub_system, SCANNER_SUB_SYS | HEAL_SUB_SYS) { - validate_server_config_for_context(context, &config, Some(sub_system)).await?; - // Scanner cycles refresh from the process-wide server config before - // each pass. Publish the same validated snapshot first so that refresh - // cannot overwrite this peer's dynamic scanner update with stale data. - publish_server_config_for_context(context, config.clone()); - } + if publication_sub_system == STORAGE_CLASS_SUB_SYS { + prepared.publish_storage_class_for_context(publication_context.as_ref())?; + return Ok::, S3Error>(None); + } + if NOTIFY_SUB_SYSTEMS.contains(&publication_sub_system.as_str()) { + return Ok(Some(config)); + } + if matches!(publication_sub_system.as_str(), SCANNER_SUB_SYS | HEAL_SUB_SYS) { + publish_server_config_for_context(publication_context.as_ref(), config.clone()); + } + apply_dynamic_config_for_subsystem_for_context(publication_context.as_ref(), &config, &publication_sub_system) + .await + .inspect_err(|_| { + warn!( + config_subsystem = publication_sub_system, + reason = "apply_failed", + "Peer dynamic config apply failed" + ); + })?; + Ok(None) + }) + .await + .map_err(|err| { + warn!("peer reload_dynamic_config: failed to acquire server config publication fence for {sub_system}: {err}"); + internal_error(format!("failed to lock server config: {err}")) + })??; + + let Some(config) = notify_config else { + return Ok(()); + }; apply_dynamic_config_for_subsystem_for_context(context, &config, sub_system) .await .inspect_err(|_| { @@ -439,23 +502,16 @@ pub async fn reload_dynamic_config_runtime_state(sub_system: &str) -> S3Result<( reload_dynamic_config_runtime_state_for_context(context.as_deref(), sub_system).await } -async fn reload_runtime_config_snapshot_with( - read: ReadFuture, - prepare: Prepare, - publish: Publish, +async fn reload_runtime_config_snapshot_with( + publish_snapshot: PublishFuture, apply_workers: ApplyWorkers, ) -> S3Result<()> where - ReadFuture: Future>, - Prepare: FnOnce(ServerConfig) -> PrepareFuture, - PrepareFuture: Future>, - Publish: FnOnce(&ServerConfig, PreparedRuntimeConfig) -> S3Result<()>, + PublishFuture: Future>, ApplyWorkers: FnOnce(ServerConfig) -> ApplyWorkersFuture, ApplyWorkersFuture: Future>, { - let config = read.await?; - let (config, prepared) = prepare(config).await?; - publish(&config, prepared)?; + let config = publish_snapshot.await?; // Worker reloads mutate live state and have no rollback contract, so the // validated snapshots stay published. The RPC still reports convergence @@ -474,55 +530,105 @@ where Ok(()) } -pub async fn reload_runtime_config_snapshot_for_context(context: Option<&AppContext>) -> S3Result<()> { - let _reload_guard = RUNTIME_CONFIG_RELOAD_MUTEX.lock().await; +async fn publish_latest_runtime_config_snapshot_under_reload_lock_for_context( + context: Option<&AppContext>, + sub_system: Option<&str>, + apply_workers: ApplyWorkers, +) -> S3Result<()> +where + ApplyWorkers: FnOnce(ServerConfig) -> ApplyWorkersFuture + Send + 'static, + ApplyWorkersFuture: Future> + Send + 'static, +{ let store = resolve_runtime_config_store_for_context(context)?; + let read_store = store.clone(); + let publication_context = context.cloned(); + let publication_sub_system = sub_system.map(str::to_owned); - reload_runtime_config_snapshot_with( - async move { - read_admin_config_without_migrate(store).await.map_err(|err| { - warn!("peer reload_runtime_config_snapshot: failed to load server config: {err}"); - internal_error(format!("failed to load server config: {err}")) - }) - }, - |config| async move { - let prepared = prepare_server_config_for_context(context, &config, None).await.map_err(|_| { - warn!("peer reload_runtime_config_snapshot: failed to prepare server config"); - internal_error("failed to prepare server config") - })?; - Ok((config, prepared)) - }, - |config, prepared| { - prepared.publish_storage_class_for_context(context)?; - publish_server_config_for_context(context, config.clone()); - Ok(()) - }, - |config| async move { - let mut failures = Vec::new(); - for sub_system in FULL_CONFIG_WORKER_SUBSYSTEMS { - if apply_dynamic_config_for_subsystem_for_context(context, &config, sub_system) - .await - .is_err() + with_admin_server_config_read_lock(store, move || async move { + let config = read_existing_admin_server_config_no_lock(read_store).await.map_err(|err| { + warn!("runtime config publication failed to load server config: {err}"); + internal_error(format!("failed to load server config: {err}")) + })?; + let prepared = + prepare_server_config_for_context(publication_context.as_ref(), &config, publication_sub_system.as_deref()) + .await + .map_err(|err| match publication_sub_system.as_deref() { + None => { + warn!("peer reload_runtime_config_snapshot: failed to prepare server config"); + internal_error("failed to prepare server config") + } + Some(STORAGE_CLASS_SUB_SYS) => internal_error(format!("failed to apply storage class config: {err}")), + Some(_) => err, + })?; + reload_runtime_config_snapshot_with( + async move { + if publication_sub_system + .as_deref() + .is_none_or(|sub_system| sub_system == STORAGE_CLASS_SUB_SYS) { - failures.push(sub_system); - warn!( - event = EVENT_CONFIG_WORKER_RELOAD_FAILED, - component = LOG_COMPONENT_ADMIN, - subsystem = LOG_SUBSYSTEM_CONFIG, - config_subsystem = sub_system, - state = CONFIG_WORKER_RELOAD_FAILURE_STATE, - reason = "apply_failed", - "Peer runtime config snapshot was published but a subsystem worker reload failed" - ); + prepared.publish_storage_class_for_context(publication_context.as_ref())?; } + publish_server_config_for_context(publication_context.as_ref(), config.clone()); + Ok(config) + }, + apply_workers, + ) + .await + }) + .await + .map_err(|err| { + warn!("runtime config publication failed to acquire server config publication fence: {err}"); + internal_error(format!("failed to lock server config: {err}")) + })? +} + +pub(crate) async fn publish_latest_runtime_config_snapshot(sub_system: &str) -> S3Result<()> { + let context = current_app_context(); + let sub_system = sub_system.to_owned(); + with_runtime_config_reload_lock(async move { + publish_latest_runtime_config_snapshot_under_reload_lock_for_context(context.as_deref(), Some(&sub_system), |_| async { + Ok(()) + }) + .await + }) + .await +} + +pub async fn reload_runtime_config_snapshot_for_context(context: Option<&AppContext>) -> S3Result<()> { + let context = context.cloned(); + with_runtime_config_reload_lock(async move { + reload_runtime_config_snapshot_under_reload_lock_for_context(context.as_ref()).await + }) + .await +} + +async fn reload_runtime_config_snapshot_under_reload_lock_for_context(context: Option<&AppContext>) -> S3Result<()> { + let worker_context = context.cloned(); + publish_latest_runtime_config_snapshot_under_reload_lock_for_context(context, None, move |config| async move { + let mut failures = Vec::new(); + for sub_system in FULL_CONFIG_WORKER_SUBSYSTEMS { + if apply_dynamic_config_for_subsystem_for_context(worker_context.as_ref(), &config, sub_system) + .await + .is_err() + { + failures.push(sub_system); + warn!( + event = EVENT_CONFIG_WORKER_RELOAD_FAILED, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_CONFIG, + config_subsystem = sub_system, + state = CONFIG_WORKER_RELOAD_FAILURE_STATE, + reason = "apply_failed", + "Peer runtime config snapshot was published but a subsystem worker reload failed" + ); } - if failures.is_empty() { - Ok(()) - } else { - Err(internal_error(format!("runtime worker reload failed: {}", failures.join("; ")))) - } - }, - ) + } + if failures.is_empty() { + Ok(()) + } else { + Err(internal_error(format!("runtime worker reload failed: {}", failures.join("; ")))) + } + }) .await } @@ -699,6 +805,77 @@ mod tests { } } + #[tokio::test] + async fn runtime_reload_waiter_cancellation_does_not_leave_queued_work() { + let _blocker = RUNTIME_CONFIG_RELOAD_MUTEX.lock().await; + let (operation_started_tx, mut operation_started_rx) = tokio::sync::oneshot::channel(); + let mut reload = Box::pin(with_runtime_config_reload_lock(async move { + let _ = operation_started_tx.send(()); + Ok(()) + })); + + poll_fn(|cx| match reload.as_mut().poll(cx) { + Poll::Pending => Poll::Ready(()), + Poll::Ready(_) => panic!("reload must wait while the mutex is held"), + }) + .await; + drop(reload); + + assert!( + matches!(operation_started_rx.try_recv(), Err(tokio::sync::oneshot::error::TryRecvError::Closed)), + "cancelling a queued reload must drop its work instead of detaching it" + ); + } + + #[tokio::test] + async fn runtime_reload_lock_survives_waiter_cancellation() { + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let (completed_tx, completed_rx) = tokio::sync::oneshot::channel(); + + let waiter = tokio::spawn(async move { + with_runtime_config_reload_lock(async move { + started_tx.send(()).expect("signal runtime reload start"); + release_rx.await.expect("release supervised runtime reload"); + completed_tx.send(()).expect("signal runtime reload completion"); + Ok(()) + }) + .await + }); + + started_rx.await.expect("supervised runtime reload started"); + waiter.abort(); + assert!(waiter.await.expect_err("reload waiter should be cancelled").is_cancelled()); + + let (contender_polled_tx, contender_polled_rx) = tokio::sync::oneshot::channel(); + let (contender_entered_tx, mut contender_entered_rx) = tokio::sync::oneshot::channel(); + let contender = tokio::spawn(async move { + let mut lock = Box::pin(RUNTIME_CONFIG_RELOAD_MUTEX.lock()); + let mut contender_polled_tx = Some(contender_polled_tx); + let _guard = poll_fn(|cx| { + if let Some(tx) = contender_polled_tx.take() { + let _ = tx.send(()); + } + lock.as_mut().poll(cx) + }) + .await; + let _ = contender_entered_tx.send(()); + }); + + contender_polled_rx.await.expect("contending reload should be polled"); + assert!( + matches!(contender_entered_rx.try_recv(), Err(tokio::sync::oneshot::error::TryRecvError::Empty)), + "a cancelled waiter must not release the reload mutex while its detached reload is still running" + ); + + release_tx.send(()).expect("release detached runtime reload"); + completed_rx.await.expect("detached runtime reload should complete"); + contender_entered_rx + .await + .expect("contending reload should enter after completion"); + contender.await.expect("contending reload task should not panic"); + } + #[tokio::test] async fn checked_scanner_reload_reports_unreachable_peer() { let temp_dir = TempDir::new().expect("scanner reload temp dir"); @@ -1221,6 +1398,123 @@ mod tests { .await; } + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + #[serial_test::serial(storage_class_env)] + async fn full_reload_keeps_worker_apply_inside_durable_read_fence() { + temp_env::async_with_vars( + [ + (storageclass::STANDARD_ENV, None::<&str>), + (storageclass::RRS_ENV, None::<&str>), + (storageclass::OPTIMIZE_ENV, None::<&str>), + (storageclass::INLINE_BLOCK_ENV, None::<&str>), + ], + async { + let fixture = runtime_config_reload_fixture().await; + let older = scanner_server_config("41"); + save_admin_server_config(fixture.context.object_store(), &older) + .await + .expect("persist older scanner config"); + + let (worker_entered_tx, worker_entered_rx) = tokio::sync::oneshot::channel(); + let (release_worker_tx, release_worker_rx) = tokio::sync::oneshot::channel(); + let reload_context = fixture.context.clone(); + let reload = tokio::spawn(async move { + publish_latest_runtime_config_snapshot_under_reload_lock_for_context( + Some(&reload_context), + None, + move |config| async move { + worker_entered_tx.send(config).expect("signal runtime config worker entry"); + release_worker_rx.await.expect("release runtime config worker"); + Ok(()) + }, + ) + .await + }); + + let worker_config = tokio::time::timeout(REAL_STORE_TEST_TIMEOUT, worker_entered_rx) + .await + .expect("worker apply should enter") + .expect("worker apply entry signal should be delivered"); + assert_eq!( + worker_config + .get_value(SCANNER_SUB_SYS, DEFAULT_DELIMITER) + .expect("older scanner config should be loaded") + .get(SCANNER_CYCLE), + "41" + ); + + let latest = scanner_server_config("71"); + let writer_store = fixture.context.object_store(); + let (writer_polled_tx, writer_polled_rx) = tokio::sync::oneshot::channel(); + let (writer_entered_tx, mut writer_entered_rx) = tokio::sync::oneshot::channel(); + let writer = tokio::spawn(async move { + let transaction_store = writer_store.clone(); + let transaction = with_admin_server_config_write_lock(writer_store, move || async move { + writer_entered_tx.send(()).expect("signal writer entry"); + save_admin_server_config_no_lock(transaction_store, &latest).await + }); + tokio::pin!(transaction); + let mut writer_polled_tx = Some(writer_polled_tx); + poll_fn(|cx| match transaction.as_mut().poll(cx) { + Poll::Pending => { + if let Some(tx) = writer_polled_tx.take() { + let _ = tx.send(()); + } + Poll::Ready(()) + } + Poll::Ready(_) => panic!("writer entered before the worker apply released its read fence"), + }) + .await; + transaction + .await + .expect("writer should acquire the server-config locks") + .expect("writer should persist the latest config"); + }); + + tokio::time::timeout(REAL_STORE_TEST_TIMEOUT, writer_polled_rx) + .await + .expect("writer should be polled") + .expect("writer poll signal should be delivered"); + assert!( + matches!(writer_entered_rx.try_recv(), Err(tokio::sync::oneshot::error::TryRecvError::Empty)), + "a server-config writer must wait until the old worker apply completes" + ); + + release_worker_tx.send(()).expect("release worker apply"); + tokio::time::timeout(REAL_STORE_TEST_TIMEOUT, async { + reload + .await + .expect("full reload task should not panic") + .expect("full reload should finish"); + writer.await.expect("writer task should not panic"); + }) + .await + .expect("reload and writer should finish"); + + reload_runtime_config_snapshot_for_context(Some(&fixture.context)) + .await + .expect("a later full reload should publish the latest durable config"); + let snapshot = fixture + .server_snapshot + .lock() + .expect("server config result lock") + .clone() + .expect("latest server config should be published"); + assert_eq!( + snapshot + .get_value(SCANNER_SUB_SYS, DEFAULT_DELIMITER) + .expect("latest scanner config should be present") + .get(SCANNER_CYCLE), + "71" + ); + assert_eq!(rustfs_scanner::scanner_runtime_config_status().cycle_interval_seconds.value, 71); + + rustfs_scanner::apply_scanner_runtime_config(&ServerConfig::new()).expect("restore scanner runtime defaults"); + }, + ) + .await; + } + #[tokio::test] #[serial_test::serial(storage_class_env)] async fn peer_full_reload_rejects_later_pool_without_publishing() { @@ -1251,11 +1545,9 @@ mod tests { let worker_events = events.clone(); let err = reload_runtime_config_snapshot_with( - async { Ok(ServerConfig::new()) }, - |config| async { Ok((config, PreparedRuntimeConfig::default())) }, - move |_config, _prepared| { + async move { publish_events.lock().expect("reload event lock").push("publish"); - Ok(()) + Ok(ServerConfig::new()) }, move |_config| async move { let mut events = worker_events.lock().expect("reload event lock"); diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index c678b3e88..cd437819e 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -644,12 +644,17 @@ pub(crate) async fn read_admin_config_without_migrate(api: Arc) -> Resu ecstore_config::com::read_config_without_migrate(api).await } +pub(crate) async fn read_existing_admin_server_config_no_lock(api: Arc) -> Result { + ecstore_config::com::read_existing_server_config_no_lock(api).await +} + #[cfg(test)] pub(crate) async fn read_admin_config_without_migrate_no_lock(api: Arc) -> Result { ecstore_config::com::read_config_without_migrate_no_lock(api).await } pub(crate) type AdminServerConfigSnapshot = ecstore_config::com::ServerConfigSnapshot; +pub(crate) type AdminServerConfigSaveResult = ecstore_config::com::ServerConfigSaveResult; pub(crate) async fn save_admin_config(api: Arc, file: &str, data: Vec) -> Result<()> { ecstore_config::com::save_config(api, file, data).await @@ -690,8 +695,17 @@ pub(crate) async fn save_admin_server_config_snapshot( api: Arc, cfg: &rustfs_config::server_config::Config, snapshot: &AdminServerConfigSnapshot, -) -> Result { - ecstore_config::com::save_server_config_snapshot(api, cfg, snapshot).await +) -> Result { + ecstore_config::com::save_server_config_snapshot_with_generation(api, cfg, snapshot).await +} + +pub(crate) async fn with_admin_server_config_read_lock(api: Arc, operation: F) -> Result +where + F: FnOnce() -> Fut + Send + 'static, + Fut: std::future::Future + Send + 'static, + T: Send + 'static, +{ + ecstore_config::com::with_server_config_read_lock(api, operation).await } pub(crate) fn init_admin_config_defaults() { @@ -793,9 +807,10 @@ pub(crate) mod cluster { pub(crate) mod config { pub(crate) use super::storageclass; pub(crate) use super::{ - AdminServerConfigSnapshot, RUSTFS_META_BUCKET, STORAGE_CLASS_SUB_SYS, delete_admin_config, init_admin_config_defaults, - read_admin_config, read_admin_config_without_migrate, read_admin_server_config_snapshot, save_admin_config, - save_admin_server_config_snapshot, + AdminServerConfigSaveResult, AdminServerConfigSnapshot, RUSTFS_META_BUCKET, STORAGE_CLASS_SUB_SYS, delete_admin_config, + init_admin_config_defaults, read_admin_config, read_admin_config_without_migrate, read_admin_server_config_snapshot, + read_existing_admin_server_config_no_lock, save_admin_config, save_admin_server_config_snapshot, + with_admin_server_config_read_lock, }; #[cfg(test)] pub(crate) use super::{ diff --git a/scripts/e2e-run.sh b/scripts/e2e-run.sh index ed66c1aed..645184736 100755 --- a/scripts/e2e-run.sh +++ b/scripts/e2e-run.sh @@ -40,7 +40,19 @@ export RUST_BACKTRACE=full "$BIN" --address "127.0.0.1:${RUSTFS_TEST_PORT}" "$VOLUME" > "${RUSTFS_TEST_LOG}" 2>&1 & RUSTFS_PID=$! -sleep 10 +for _ in {1..60}; do + if curl --noproxy '*' --silent --fail "http://127.0.0.1:${RUSTFS_TEST_PORT}/health/ready" >/dev/null; then + break + fi + + if ! kill -0 "${RUSTFS_PID}" 2>/dev/null; then + wait "${RUSTFS_PID}" + fi + + sleep 1 +done + +curl --noproxy '*' --silent --show-error --fail "http://127.0.0.1:${RUSTFS_TEST_PORT}/health/ready" >/dev/null export AWS_ACCESS_KEY_ID="${RUSTFS_ACCESS_KEY:-rustfsadmin}" export AWS_SECRET_ACCESS_KEY="${RUSTFS_SECRET_KEY:-rustfsadmin}"