From accc906b33336e9d61ef5de3cd157330ae059e04 Mon Sep 17 00:00:00 2001 From: cxymds Date: Mon, 3 Aug 2026 22:00:49 +0800 Subject: [PATCH] fix(replication): fence force-delete journal updates (#5661) * feat(replication): add conditional config store APIs * fix(replication): fence force-delete journal updates * fix(replication): retry durable force-delete commits * test(replication): fence lost force-delete journal leases * fix(replication): bound force-delete journal retries --- .../replication/replication_config_store.rs | 39 +- .../replication_metadata_boundary.rs | 5 + .../bucket/replication/replication_pool.rs | 435 +++++++++++++++--- .../replication_storage_boundary.rs | 2 +- crates/ecstore/src/config/com.rs | 43 +- 5 files changed, 462 insertions(+), 62 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_config_store.rs b/crates/ecstore/src/bucket/replication/replication_config_store.rs index 28a90a83b..f0db89f2e 100644 --- a/crates/ecstore/src/bucket/replication/replication_config_store.rs +++ b/crates/ecstore/src/bucket/replication/replication_config_store.rs @@ -13,7 +13,7 @@ // limitations under the License. use super::replication_error_boundary::Result; -use super::replication_storage_boundary::ReplicationObjectIO; +use super::replication_storage_boundary::{HTTPPreconditions, ObjectInfo, ObjectOptions, ReplicationObjectIO}; use crate::config::{com, storageclass}; use std::sync::Arc; @@ -37,6 +37,21 @@ impl ReplicationConfigStore { com::read_config_no_lock(api, file).await } + pub(crate) async fn read_no_lock_with_metadata(api: Arc, file: &str) -> Result<(Vec, ObjectInfo)> + where + S: ReplicationObjectIO, + { + com::read_config_with_metadata( + api, + file, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + } + pub(crate) async fn save(api: Arc, file: &str, data: Vec) -> Result<()> where S: ReplicationObjectIO, @@ -50,4 +65,26 @@ impl ReplicationConfigStore { { com::save_config_no_lock(api, file, data).await } + + pub(crate) async fn save_conditional( + api: Arc, + file: &str, + data: Vec, + http_preconditions: HTTPPreconditions, + ) -> Result<()> + where + S: ReplicationObjectIO, + { + com::save_config_with_opts_quiet( + api, + file, + data, + &ObjectOptions { + max_parity: true, + http_preconditions: Some(http_preconditions), + ..Default::default() + }, + ) + .await + } } diff --git a/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs b/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs index 64f6c362f..c7d7ba37c 100644 --- a/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs @@ -33,6 +33,7 @@ pub(crate) struct ReplicationMetadataStore; impl ReplicationMetadataStore { pub(crate) const MRF_REPLICATION_FILE: &'static str = "config/replication/mrf.bin"; pub(crate) const FORCE_DELETE_REPLICATION_FILE: &'static str = "config/replication/force-delete.bin"; + pub(crate) const FORCE_DELETE_REPLICATION_TRANSACTION_LOCK: &'static str = "config/replication/force-delete.bin.transaction"; pub(crate) async fn replication_config(bucket: &str) -> Result<(ReplicationConfiguration, OffsetDateTime)> { metadata_sys::get_replication_config(bucket).await @@ -114,5 +115,9 @@ mod tests { ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE, "config/replication/force-delete.bin" ); + assert_eq!( + ReplicationMetadataStore::FORCE_DELETE_REPLICATION_TRANSACTION_LOCK, + "config/replication/force-delete.bin.transaction" + ); } } diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index ee27c422b..af727334a 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -42,7 +42,8 @@ use super::replication_resyncer::{ }; use super::replication_state::ReplicationStats; use super::replication_storage_boundary::{ - ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationDeletedObject, ReplicationObjectIO, ReplicationStorage, + HTTPPreconditions, ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationDeletedObject, ReplicationObjectIO, + ReplicationStorage, }; use super::replication_target_boundary::{ReplicationTargetStore, replication_object_is_ssec_encrypted}; use super::replication_versioning_boundary::ReplicationVersioningStore; @@ -403,75 +404,143 @@ pub async fn persist_force_delete_intent( mut entry: MrfReplicateEntry, ) -> Result<(), EcstoreError> { entry.force_delete_local_commit = false; - let file = ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE; - let lock = storage - .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), file) - .await?; - let _guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?; - - let mut entries = match ReplicationConfigStore::read_no_lock(storage.clone(), file).await { - Ok(data) => decode_mrf_file(&data)?, - Err(EcstoreError::ConfigNotFound) => Vec::new(), - Err(err) => return Err(err), - }; - - if entries - .iter() - .any(|existing| existing.force_delete_id == entry.force_delete_id) - { - return Ok(()); - } - - entries.push(entry); - let data = encode_mrf_file(&entries)?; - ReplicationConfigStore::save_no_lock(storage, file, data).await + update_force_delete_intents(storage, move |entries, _exists| { + if entries + .iter() + .any(|existing| existing.force_delete_id == entry.force_delete_id) + { + return Ok(false); + } + entries.push(entry.clone()); + Ok(true) + }) + .await } pub async fn commit_force_delete_intent( storage: Arc, operation_id: uuid::Uuid, ) -> Result<(), EcstoreError> { - let file = ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE; - let lock = storage - .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), file) - .await?; - let _guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?; - - let data = ReplicationConfigStore::read_no_lock(storage.clone(), file).await?; - let mut entries = decode_mrf_file(&data)?; - let Some(entry) = entries.iter_mut().find(|entry| entry.force_delete_id == Some(operation_id)) else { - return Err(EcstoreError::ConfigNotFound); - }; - if entry.force_delete_local_commit { - return Ok(()); - } - entry.force_delete_local_commit = true; - ReplicationConfigStore::save_no_lock(storage, file, encode_mrf_file(&entries)?).await + update_force_delete_intents(storage, move |entries, exists| { + if !exists { + return Err(EcstoreError::ConfigNotFound); + } + let Some(entry) = entries.iter_mut().find(|entry| entry.force_delete_id == Some(operation_id)) else { + return Err(EcstoreError::ConfigNotFound); + }; + if entry.force_delete_local_commit { + return Ok(false); + } + entry.force_delete_local_commit = true; + Ok(true) + }) + .await } pub async fn complete_force_delete_intent( storage: Arc, operation_id: uuid::Uuid, ) -> Result<(), EcstoreError> { - let file = ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE; - let lock = storage - .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), file) - .await?; - let _guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?; + update_force_delete_intents(storage, move |entries, exists| { + if !exists { + return Ok(false); + } + let original_len = entries.len(); + entries.retain(|entry| entry.force_delete_id != Some(operation_id)); + Ok(entries.len() != original_len) + }) + .await +} - let data = match ReplicationConfigStore::read_no_lock(storage.clone(), file).await { - Ok(data) => data, - Err(EcstoreError::ConfigNotFound) => return Ok(()), - Err(err) => return Err(err), - }; - let mut entries = decode_mrf_file(&data)?; - let original_len = entries.len(); - entries.retain(|entry| entry.force_delete_id != Some(operation_id)); - if entries.len() == original_len { - return Ok(()); +const FORCE_DELETE_INTENT_CAS_RETRIES: usize = 3; + +fn is_retryable_force_delete_error(error: &EcstoreError) -> bool { + matches!(error, EcstoreError::PreconditionFailed) || error.to_string().contains("force-delete journal lock lost") +} + +async fn update_force_delete_intents(storage: Arc, mut update: F) -> Result<(), EcstoreError> +where + S: ReplicationStorage, + F: FnMut(&mut Vec, bool) -> Result, +{ + let file = ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE; + for attempt in 0..=FORCE_DELETE_INTENT_CAS_RETRIES { + let result = { + let lock = storage + .new_ns_lock( + ReplicationMetadataStore::rustfs_meta_bucket(), + ReplicationMetadataStore::FORCE_DELETE_REPLICATION_TRANSACTION_LOCK, + ) + .await?; + // Lock order is transaction namespace lock -> force-delete journal object lock. + // Keep the transaction guard alive through the conditional write so legacy + // writers cannot interleave a read-modify-write transition within this process. + let guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?; + let (mut entries, preconditions, exists) = read_force_delete_intents(storage.clone(), file).await?; + if !update(&mut entries, exists)? { + return Ok(()); + } + save_force_delete_intents(storage.clone(), file, &guard, entries, preconditions).await + }; + + match result { + Err(error) if is_retryable_force_delete_error(&error) && attempt < FORCE_DELETE_INTENT_CAS_RETRIES => { + tokio::time::sleep(Duration::from_millis(25)).await; + } + result => return result, + } } - ReplicationConfigStore::save_no_lock(storage, file, encode_mrf_file(&entries)?).await + Err(EcstoreError::other("force-delete journal update retries exhausted")) +} + +async fn read_force_delete_intents( + storage: Arc, + file: &str, +) -> Result<(Vec, HTTPPreconditions, bool), EcstoreError> { + match ReplicationConfigStore::read_no_lock_with_metadata(storage, file).await { + Ok((data, object_info)) => { + let etag = object_info + .etag + .filter(|etag| !etag.trim().is_empty()) + .ok_or_else(|| EcstoreError::other("force-delete journal has no ETag for conditional update"))?; + Ok(( + decode_mrf_file(&data)?, + HTTPPreconditions { + if_match: Some(etag), + ..Default::default() + }, + true, + )) + } + Err(EcstoreError::ConfigNotFound) => Ok(( + Vec::new(), + HTTPPreconditions { + if_none_match: Some("*".to_string()), + ..Default::default() + }, + false, + )), + Err(err) => Err(err), + } +} + +async fn save_force_delete_intents( + storage: Arc, + file: &str, + guard: &rustfs_lock::NamespaceLockGuard, + entries: Vec, + preconditions: HTTPPreconditions, +) -> Result<(), EcstoreError> { + ensure_force_delete_journal_lock_held(guard.is_lock_lost())?; + ReplicationConfigStore::save_conditional(storage, file, encode_mrf_file(&entries)?, preconditions).await +} + +fn ensure_force_delete_journal_lock_held(lock_lost: bool) -> Result<(), EcstoreError> { + if lock_lost { + return Err(EcstoreError::other("force-delete journal lock lost before conditional update")); + } + Ok(()) } #[derive(Debug, thiserror::Error)] @@ -2522,7 +2591,7 @@ mod tests { StorageListObjectVersionsInfo, StorageListObjectsV2Info, StorageNamespaceLocking, StorageObjectInfoOrErr, WalkOptions, }; use super::*; - use std::collections::HashMap; + use std::collections::{HashMap, VecDeque}; use std::fmt::{Debug, Formatter}; use std::io::Cursor; use std::sync::Mutex as StdMutex; @@ -2537,6 +2606,11 @@ mod tests { struct LoadResyncSharedState { data: StdMutex>, + etag_revision: AtomicUsize, + last_put_preconditions: StdMutex>, + last_put_no_lock: AtomicBool, + omit_etag: AtomicBool, + conditional_write_replacements: StdMutex>>, writes: StdMutex)>>, lock_manager: Arc, first_read_started: Notify, @@ -2615,6 +2689,8 @@ mod tests { object_info: ObjectInfo { size, actual_size: size, + etag: (!self.shared.omit_etag.load(Ordering::SeqCst)) + .then(|| format!("mrf-{}", self.shared.etag_revision.load(Ordering::SeqCst))), ..Default::default() }, buffered_body: None, @@ -2627,8 +2703,44 @@ mod tests { _bucket: &str, object: &str, data: &mut Self::PutObjectReader, - _opts: &Self::ObjectOptions, + opts: &Self::ObjectOptions, ) -> Result { + if opts.http_preconditions.is_some() + && let Some(replacement) = self + .shared + .conditional_write_replacements + .lock() + .expect("test replacement lock should not be poisoned") + .pop_front() + { + *self.shared.data.lock().expect("test data lock should not be poisoned") = replacement; + self.shared.etag_revision.fetch_add(1, Ordering::SeqCst); + } + let current_etag = if self + .shared + .data + .lock() + .expect("test data lock should not be poisoned") + .is_empty() + { + None + } else { + Some(format!("mrf-{}", self.shared.etag_revision.load(Ordering::SeqCst))) + }; + if opts.http_preconditions.as_ref().is_some_and(|preconditions| { + preconditions.if_none_match_value() == Some("*") && current_etag.is_some() + || preconditions + .if_match_value() + .is_some_and(|expected| current_etag.as_deref() != Some(expected)) + }) { + return Err(EcstoreError::PreconditionFailed); + } + *self + .shared + .last_put_preconditions + .lock() + .expect("test preconditions lock should not be poisoned") = opts.http_preconditions.clone(); + self.shared.last_put_no_lock.store(opts.no_lock, Ordering::SeqCst); if self.shared.fail_next_write.swap(false, Ordering::SeqCst) { return Err(EcstoreError::Unexpected); } @@ -2644,6 +2756,7 @@ mod tests { .expect("test writes lock should not be poisoned") .push((object.to_string(), encoded.clone())); *self.shared.data.lock().expect("test data lock should not be poisoned") = encoded; + self.shared.etag_revision.fetch_add(1, Ordering::SeqCst); self.shared.write_count.fetch_add(1, Ordering::SeqCst); Ok(ObjectInfo::default()) } @@ -3096,6 +3209,11 @@ mod tests { fn empty_resync_shared_state() -> Arc { Arc::new(LoadResyncSharedState { data: StdMutex::new(Vec::new()), + etag_revision: AtomicUsize::new(0), + last_put_preconditions: StdMutex::new(None), + last_put_no_lock: AtomicBool::new(false), + omit_etag: AtomicBool::new(false), + conditional_write_replacements: StdMutex::new(VecDeque::new()), writes: StdMutex::new(Vec::new()), lock_manager: Arc::new(rustfs_lock::GlobalLockManager::new()), first_read_started: Notify::new(), @@ -3743,6 +3861,11 @@ mod tests { temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("1"))], async { let shared = Arc::new(LoadResyncSharedState { data: StdMutex::new(load_resync_test_metadata()), + etag_revision: AtomicUsize::new(1), + last_put_preconditions: StdMutex::new(None), + last_put_no_lock: AtomicBool::new(false), + omit_etag: AtomicBool::new(false), + conditional_write_replacements: StdMutex::new(VecDeque::new()), writes: StdMutex::new(Vec::new()), lock_manager: Arc::new(rustfs_lock::GlobalLockManager::new()), first_read_started: Notify::new(), @@ -4393,7 +4516,7 @@ mod tests { #[tokio::test] async fn force_delete_intent_append_commit_and_cleanup_are_idempotent() { let shared = empty_resync_shared_state(); - let storage = Arc::new(LoadResyncNodeStore::new("force-delete-journal", shared)); + let storage = Arc::new(LoadResyncNodeStore::new("force-delete-journal", shared.clone())); let operation_id = Uuid::new_v4(); let entry = MrfReplicateEntry { bucket: "source".to_string(), @@ -4408,6 +4531,15 @@ mod tests { persist_force_delete_intent(storage.clone(), entry.clone()) .await .expect("first journal append should succeed"); + let preconditions = shared + .last_put_preconditions + .lock() + .expect("test preconditions lock should not be poisoned") + .clone() + .expect("first journal append should be conditional"); + assert_eq!(preconditions.if_none_match_value(), Some("*")); + assert_eq!(preconditions.if_match_value(), None); + assert!(!shared.last_put_no_lock.load(Ordering::SeqCst)); persist_force_delete_intent(storage.clone(), entry) .await .expect("duplicate journal append should be a no-op"); @@ -4423,6 +4555,15 @@ mod tests { commit_force_delete_intent(storage.clone(), operation_id) .await .expect("commit marker should persist"); + let preconditions = shared + .last_put_preconditions + .lock() + .expect("test preconditions lock should not be poisoned") + .clone() + .expect("commit marker should be conditional"); + assert_eq!(preconditions.if_none_match_value(), None); + assert_eq!(preconditions.if_match_value(), Some("mrf-1")); + assert!(!shared.last_put_no_lock.load(Ordering::SeqCst)); commit_force_delete_intent(storage.clone(), operation_id) .await .expect("duplicate commit marker should be a no-op"); @@ -4440,4 +4581,182 @@ mod tests { .await .expect("duplicate journal cleanup should be a no-op"); } + + #[tokio::test] + async fn force_delete_intent_cleanup_retries_after_a_stale_journal_snapshot() { + let shared = empty_resync_shared_state(); + let storage = Arc::new(LoadResyncNodeStore::new("force-delete-journal", shared.clone())); + let operation_id = Uuid::new_v4(); + let mut entry = MrfReplicateEntry { + bucket: "source".to_string(), + object: "original".to_string(), + force_delete_id: Some(operation_id), + op: MrfOpKind::Delete, + ..Default::default() + }; + persist_force_delete_intent(storage.clone(), entry.clone()) + .await + .expect("journal append should succeed"); + commit_force_delete_intent(storage.clone(), operation_id) + .await + .expect("journal commit should succeed"); + entry.force_delete_local_commit = true; + let concurrent = MrfReplicateEntry { + bucket: "source".to_string(), + object: "concurrent".to_string(), + force_delete_id: Some(Uuid::new_v4()), + op: MrfOpKind::Delete, + ..Default::default() + }; + shared + .conditional_write_replacements + .lock() + .expect("test replacement lock should not be poisoned") + .push_back(encode_mrf_file(&[entry, concurrent.clone()]).expect("concurrent journal entries should encode")); + + complete_force_delete_intent(storage.clone(), operation_id) + .await + .expect("cleanup should retry after a concurrent journal update"); + + let data = ReplicationConfigStore::read(storage, ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE) + .await + .expect("journal should remain readable"); + let entries = decode_mrf_file(&data).expect("journal should decode"); + assert_eq!(entries.len(), 1, "cleanup must preserve only the concurrent journal entry"); + assert_eq!(entries[0].force_delete_id, concurrent.force_delete_id); + assert_eq!(entries[0].object, concurrent.object); + } + + #[tokio::test] + async fn force_delete_intent_commit_retries_past_the_bounded_cas_conflict_limit() { + let shared = empty_resync_shared_state(); + let storage = Arc::new(LoadResyncNodeStore::new("force-delete-journal", shared.clone())); + let operation_id = Uuid::new_v4(); + let entry = MrfReplicateEntry { + bucket: "source".to_string(), + object: "original".to_string(), + force_delete_id: Some(operation_id), + op: MrfOpKind::Delete, + ..Default::default() + }; + persist_force_delete_intent(storage.clone(), entry.clone()) + .await + .expect("journal append should succeed"); + { + let mut replacements = shared + .conditional_write_replacements + .lock() + .expect("test replacement lock should not be poisoned"); + for object in ["first", "second", "third"] { + let mut replacement = entry.clone(); + replacement.object = object.to_string(); + replacements.push_back(encode_mrf_file(&[replacement]).expect("concurrent journal entry should encode")); + } + } + + commit_force_delete_intent(storage.clone(), operation_id) + .await + .expect("commit marker must retry until it is durable"); + + let data = ReplicationConfigStore::read(storage, ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE) + .await + .expect("journal should remain readable"); + let entries = decode_mrf_file(&data).expect("journal should decode"); + assert_eq!(entries.len(), 1); + assert!(entries[0].force_delete_local_commit); + assert_eq!(entries[0].force_delete_id, Some(operation_id)); + } + + #[tokio::test] + async fn force_delete_intent_rejects_existing_journal_without_an_etag() { + let shared = empty_resync_shared_state(); + let storage = Arc::new(LoadResyncNodeStore::new("force-delete-journal", shared.clone())); + let operation_id = Uuid::new_v4(); + let entry = MrfReplicateEntry { + bucket: "source".to_string(), + object: "original".to_string(), + force_delete_id: Some(operation_id), + op: MrfOpKind::Delete, + ..Default::default() + }; + persist_force_delete_intent(storage.clone(), entry) + .await + .expect("journal append should succeed"); + let writes_before = shared.write_count.load(Ordering::SeqCst); + shared.omit_etag.store(true, Ordering::SeqCst); + + let err = commit_force_delete_intent(storage.clone(), operation_id) + .await + .expect_err("missing ETag must reject journal mutation"); + assert!(err.to_string().contains("no ETag")); + assert_eq!(shared.write_count.load(Ordering::SeqCst), writes_before); + let data = ReplicationConfigStore::read(storage, ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE) + .await + .expect("journal should remain readable"); + let entries = decode_mrf_file(&data).expect("journal should decode"); + assert!(!entries[0].force_delete_local_commit); + } + + #[test] + fn force_delete_journal_rejects_a_lost_transaction_lease() { + let err = ensure_force_delete_journal_lock_held(true).expect_err("lost transaction lease must fence the journal write"); + + assert!(err.to_string().contains("lock lost")); + } + + #[tokio::test] + async fn force_delete_journal_rejects_a_stale_conditional_write() { + let shared = empty_resync_shared_state(); + let storage = Arc::new(LoadResyncNodeStore::new("force-delete-journal", shared)); + let file = ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE; + let original = MrfReplicateEntry { + bucket: "source".to_string(), + object: "original".to_string(), + force_delete_id: Some(Uuid::new_v4()), + op: MrfOpKind::Delete, + ..Default::default() + }; + ReplicationConfigStore::save( + storage.clone(), + file, + encode_mrf_file(&[original]).expect("initial journal entry should encode"), + ) + .await + .expect("initial journal write should succeed"); + + let (_, object_info) = ReplicationConfigStore::read_no_lock_with_metadata(storage.clone(), file) + .await + .expect("journal snapshot should include an ETag"); + let stale_preconditions = HTTPPreconditions { + if_match: object_info.etag, + ..Default::default() + }; + let replacement = MrfReplicateEntry { + bucket: "source".to_string(), + object: "replacement".to_string(), + force_delete_id: Some(Uuid::new_v4()), + op: MrfOpKind::Delete, + ..Default::default() + }; + let replacement_data = encode_mrf_file(&[replacement]).expect("replacement journal entry should encode"); + ReplicationConfigStore::save(storage.clone(), file, replacement_data.clone()) + .await + .expect("concurrent journal write should succeed"); + + let err = ReplicationConfigStore::save_conditional( + storage.clone(), + file, + encode_mrf_file(&[]).expect("empty journal should encode"), + stale_preconditions, + ) + .await + .expect_err("stale journal snapshot must not overwrite newer data"); + assert_eq!(err, EcstoreError::PreconditionFailed); + assert_eq!( + ReplicationConfigStore::read(storage, file) + .await + .expect("newer journal data should remain readable"), + replacement_data + ); + } } diff --git a/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs b/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs index 315b88e15..9126f3563 100644 --- a/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs @@ -25,7 +25,7 @@ pub(crate) use crate::storage_api_contracts::list::{ }; pub(crate) use crate::storage_api_contracts::namespace::NamespaceLocking as StorageNamespaceLocking; pub(crate) use crate::storage_api_contracts::object::{ - DeletedObject, EcstoreObjectOperations, ObjectIO, ObjectOperations, ObjectToDelete, + DeletedObject, EcstoreObjectOperations, HTTPPreconditions, ObjectIO, ObjectOperations, ObjectToDelete, }; pub(crate) use crate::storage_api_contracts::range::HTTPRangeSpec; pub(crate) use rustfs_replication::{DeletedObject as ReplicationDeletedObject, ObjectToDelete as ReplicationObjectToDelete}; diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index c62aa74d8..057de9d0b 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -586,10 +586,47 @@ where PutObjectReader = PutObjReader, >, { - save_config_with_opts_and_metadata(api, file, data, opts).await.map(|_| ()) + save_config_with_opts_inner(api, file, data, opts, true).await.map(|_| ()) +} + +/// Saves a configuration object without logging an error for a retryable caller-owned failure. +pub async fn save_config_with_opts_quiet(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_inner(api, file, data, opts, false).await.map(|_| ()) } async fn save_config_with_opts_and_metadata(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_inner(api, file, data, opts, true).await +} + +async fn save_config_with_opts_inner( + api: Arc, + file: &str, + data: Vec, + opts: &ObjectOptions, + log_error: bool, +) -> Result where S: ObjectIO< Error = Error, @@ -605,7 +642,9 @@ where 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); + if log_error { + error!("save_config_with_opts: err: {:?}, file: {}", err, file); + } Err(err) } }