diff --git a/crates/e2e_test/src/fake_s3_target/mod.rs b/crates/e2e_test/src/fake_s3_target/mod.rs index 0fa3c67ef..ba47db6f0 100644 --- a/crates/e2e_test/src/fake_s3_target/mod.rs +++ b/crates/e2e_test/src/fake_s3_target/mod.rs @@ -134,6 +134,7 @@ pub struct RequestRecord { #[derive(Default)] struct ControlState { scripts: HashMap>, + keyed_scripts: HashMap<(Operation, String), VecDeque>, requests: VecDeque, next_sequence: u64, } @@ -352,25 +353,44 @@ impl FakeS3Target { state.buckets.entry(bucket).or_default(); } + /// Remove all retained object versions while preserving the bucket. + pub fn clear_bucket_objects(&self, bucket: &str) { + let mut state = lock(&self.backend.store); + let (removed_versions, removed_bytes) = state + .buckets + .get_mut(bucket) + .expect("fake target bucket must exist") + .objects + .drain() + .flat_map(|(_, versions)| versions) + .fold((0usize, 0usize), |(count, bytes), version| (count + 1, bytes + version.body.len())); + state.total_versions = state + .total_versions + .checked_sub(removed_versions) + .expect("fake target version accounting must not underflow"); + state.total_bytes = state + .total_bytes + .checked_sub(removed_bytes) + .expect("fake target byte accounting must not underflow"); + } + + pub fn has_object(&self, bucket: &str, key: &str) -> bool { + lock(&self.backend.store) + .buckets + .get(bucket) + .and_then(|bucket| bucket.objects.get(key)) + .and_then(|versions| versions.last()) + .is_some_and(|version| !version.delete_marker) + } + /// Queue `times` copies of a fault for one operation. pub fn inject(&self, operation: Operation, action: FaultAction, times: usize) { if times == 0 { return; } - if let FaultAction::SlowDrain { chunk_bytes: 0, .. } = action { - panic!("slow-drain chunk size must be non-zero"); - } - match &action { - FaultAction::Delay(duration) if *duration > MAX_FAULT_DURATION => { - panic!("fault delay must not exceed 30 seconds"); - } - FaultAction::SlowDrain { delay, .. } if *delay >= MAX_FAULT_DURATION => { - panic!("slow-drain slice delay must be below 30 seconds"); - } - _ => {} - } + validate_fault_action(&action); let mut state = lock(&self.control); - let queued = state.scripts.values().map(VecDeque::len).sum::(); + let queued = queued_fault_count(&state); if queued.checked_add(times).is_none_or(|total| total > MAX_SCRIPTED_FAULTS) { panic!("fake target queues at most 4096 scripted faults"); } @@ -381,8 +401,28 @@ impl FakeS3Target { .extend(std::iter::repeat_n(action, times)); } + /// Queue faults for one exact object key without affecting concurrent requests. + pub fn inject_for_key(&self, operation: Operation, key: impl Into, action: FaultAction, times: usize) { + if times == 0 { + return; + } + validate_fault_action(&action); + let mut state = lock(&self.control); + let queued = queued_fault_count(&state); + if queued.checked_add(times).is_none_or(|total| total > MAX_SCRIPTED_FAULTS) { + panic!("fake target queues at most 4096 scripted faults"); + } + state + .keyed_scripts + .entry((operation, key.into())) + .or_default() + .extend(std::iter::repeat_n(action, times)); + } + pub fn clear_faults(&self) { - lock(&self.control).scripts.clear(); + let mut state = lock(&self.control); + state.scripts.clear(); + state.keyed_scripts.clear(); } pub fn requests(&self) -> Vec { @@ -420,6 +460,25 @@ fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { mutex.lock().unwrap_or_else(std::sync::PoisonError::into_inner) } +fn validate_fault_action(action: &FaultAction) { + if let FaultAction::SlowDrain { chunk_bytes: 0, .. } = action { + panic!("slow-drain chunk size must be non-zero"); + } + match action { + FaultAction::Delay(duration) if *duration > MAX_FAULT_DURATION => { + panic!("fault delay must not exceed 30 seconds"); + } + FaultAction::SlowDrain { delay, .. } if *delay >= MAX_FAULT_DURATION => { + panic!("slow-drain slice delay must be below 30 seconds"); + } + _ => {} + } +} + +fn queued_fault_count(state: &ControlState) -> usize { + state.scripts.values().map(VecDeque::len).sum::() + state.keyed_scripts.values().map(VecDeque::len).sum::() +} + #[async_trait] impl S3Access for FaultAccess { async fn check(&self, context: &mut S3AccessContext<'_>) -> S3Result<()> { @@ -492,7 +551,12 @@ fn record_request( content_length: Option, ) -> Option { let mut state = lock(control); - let action = state.scripts.get_mut(&operation).and_then(VecDeque::pop_front); + let action = parsed + .key + .as_ref() + .and_then(|key| state.keyed_scripts.get_mut(&(operation, key.clone()))) + .and_then(VecDeque::pop_front) + .or_else(|| state.scripts.get_mut(&operation).and_then(VecDeque::pop_front)); state.next_sequence += 1; let sequence = state.next_sequence; if state.requests.len() == MAX_REQUEST_RECORDS { diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 5690c1a9e..5873f822c 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -16,7 +16,9 @@ use crate::common::{ RustFSTestEnvironment, awscurl_available, awscurl_post_sts_form_urlencoded, init_logging, local_http_client, replication_fast_env, rustfs_binary_path, }; -use crate::fake_s3_target::{FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, Operation as FakeTargetOperation}; +use crate::fake_s3_target::{ + FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FaultAction as FakeTargetFault, Operation as FakeTargetOperation, +}; use crate::kms::common::{create_key_with_specific_id, sse_customer_key_md5_base64}; use crate::storage_api::replication_extension::BucketTargetSys; use aws_sdk_s3::config::{Credentials, Region}; @@ -379,6 +381,10 @@ struct ReplicationResetStatusTarget { reset_id: String, #[serde(rename = "resyncStatus", default)] status: String, + #[serde(rename = "replicationCount", default)] + replicated_count: i64, + #[serde(rename = "object", default)] + object: String, } async fn signed_request( @@ -5368,6 +5374,7 @@ async fn test_site_replication_resync_lifecycle_survives_real_server_restart() - init_logging(); let resync_process_env = [ ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", "true"), + ("RUSTFS_REPL_RESYNC_POLL_MAX_MS", "100"), // Verbose server logging can block startup when this focused test is run // through a captured test process rather than nextest. ("RUST_LOG", "error"), @@ -5384,6 +5391,7 @@ async fn test_site_replication_resync_lifecycle_survives_real_server_restart() - .await?; let source_bucket = "site-repl-resync-src"; + const RESYNC_OBJECT_COUNT: usize = 128; let source_client = source_env.create_s3_client(); let target_client = target_env.create_s3_client(); @@ -5428,12 +5436,13 @@ async fn test_site_replication_resync_lifecycle_survives_real_server_restart() - wait_for_bucket_on_target(&source_client, source_bucket).await?; let target_arn = wait_for_remote_target_arn(&source_env, source_bucket).await?; - for idx in 0..96 { + for idx in 0..RESYNC_OBJECT_COUNT { + let size = if idx == 0 { 8 * 1024 * 1024 } else { 8 * 1024 }; source_client .put_object() .bucket(source_bucket) .key(format!("resync-object-{idx:02}")) - .body(ByteStream::from(vec![b'x'; 512 * 1024])) + .body(ByteStream::from(vec![b'x'; size])) .send() .await?; } @@ -5492,6 +5501,22 @@ async fn test_site_replication_resync_lifecycle_survives_real_server_restart() - ); let restarted_reset_id = restarted.resync_id.clone(); + let partial_deadline = tokio::time::Instant::now() + Duration::from_secs(10); + loop { + let status = site_replication_resync_op(&source_env, "status", &remote_peer).await?; + let replicated = status.replicated_objects; + if status.resync_id == restarted_reset_id + && replicated > 0 + && replicated < u64::try_from(RESYNC_OBJECT_COUNT).expect("test object count should fit u64") + { + break; + } + if status.state == "completed" || tokio::time::Instant::now() >= partial_deadline { + return Err(format!("resync did not expose a partial durable checkpoint before restart: {status:?}").into()); + } + sleep(Duration::from_millis(10)).await; + } + source_env.restart_server_preserving_data(vec![], &resync_process_env).await?; wait_for_site_replication_enabled(&source_env, 2).await?; @@ -5525,6 +5550,41 @@ async fn test_site_replication_resync_lifecycle_survives_real_server_restart() - })?; assert_eq!(restarted_target.reset_id, restarted_reset_id); + let completion_deadline = tokio::time::Instant::now() + Duration::from_secs(60); + let completed = loop { + let status = site_replication_resync_op(&source_env, "status", &remote_peer).await?; + match status.state.as_str() { + "completed" => break status, + "failed" => return Err(format!("recovered resync failed: {status:?}").into()), + _ if tokio::time::Instant::now() < completion_deadline => sleep(Duration::from_millis(500)).await, + _ => return Err(format!("recovered resync did not complete in time: {status:?}").into()), + } + }; + assert_eq!( + completed.replicated_objects, + u64::try_from(RESYNC_OBJECT_COUNT).expect("test object count should fit u64") + ); + + let replicated = futures::stream::iter(0..RESYNC_OBJECT_COUNT) + .map(|idx| { + let target_client = target_client.clone(); + async move { + let key = format!("resync-object-{idx:02}"); + let body = wait_for_object_on_target(&target_client, source_bucket, &key).await?; + let expected_size = if idx == 0 { 8 * 1024 * 1024 } else { 8 * 1024 }; + if body != vec![b'x'; expected_size] { + return Err(format!("recovered resync object body mismatch for {key}").into()); + } + Ok::<(), Box>(()) + } + }) + .buffer_unordered(16) + .collect::>() + .await; + for result in replicated { + result?; + } + Ok(()) } @@ -7027,6 +7087,119 @@ async fn wait_for_target_request_version_id( } } +#[tokio::test] +#[serial] +async fn test_bucket_resync_restart_revisits_objects_before_out_of_order_checkpoint() -> TestResult { + init_logging(); + + const OBJECT_COUNT: usize = 128; + let target = FakeS3Target::start().await?; + let target_bucket = "resync-checkpoint-dst"; + target.create_bucket(target_bucket); + + let mut source_env = RustFSTestEnvironment::new().await?; + let mut process_env = replication_fast_env(); + process_env.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); + process_env.extend_from_slice(&[ + ("NO_PROXY", "127.0.0.1,localhost"), + ("HTTP_PROXY", ""), + ("HTTPS_PROXY", ""), + ("RUST_LOG", "error"), + ]); + source_env.start_rustfs_server_with_env(vec![], &process_env).await?; + + let source_bucket = "resync-checkpoint-src"; + let source_client = source_env.create_s3_client(); + source_client.create_bucket().bucket(source_bucket).send().await?; + enable_bucket_versioning(&source_env, source_bucket).await?; + let target_arn = set_replication_target_with_options( + &source_env, + source_bucket, + ReplicationTargetOptions { + endpoint: &target.address(), + access_key: FAKE_ACCESS_KEY, + secret_key: FAKE_SECRET_KEY, + target_bucket, + secure: false, + skip_tls_verify: false, + ca_cert_pem: None, + }, + ) + .await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + + for idx in 0..OBJECT_COUNT { + source_client + .put_object() + .bucket(source_bucket) + .key(format!("checkpoint-{idx:03}")) + .body(ByteStream::from(format!("checkpoint payload {idx}").into_bytes())) + .send() + .await?; + } + wait_for_source_replication_status(&source_client, source_bucket, "checkpoint-127", "COMPLETED", false).await?; + let initial_replication_deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + let replicated = target + .requests() + .into_iter() + .filter(|request| request.operation == FakeTargetOperation::PutObject) + .count(); + if replicated >= OBJECT_COUNT { + break; + } + if tokio::time::Instant::now() >= initial_replication_deadline { + return Err(format!("initial replication only sent {replicated}/{OBJECT_COUNT} objects").into()); + } + sleep(Duration::from_millis(100)).await; + } + target.clear_bucket_objects(target_bucket); + target.take_requests(); + target.inject_for_key( + FakeTargetOperation::PutObject, + "checkpoint-000", + FakeTargetFault::Delay(Duration::from_secs(30)), + 100, + ); + + let (reset_arn, reset_id) = start_bucket_replication_reset(&source_env, source_bucket).await?; + assert_eq!(reset_arn, target_arn); + let partial = wait_for_replication_reset_target(&source_env, source_bucket, &target_arn, |status| { + status.reset_id == reset_id + && status.replicated_count > 0 + && status.replicated_count < i64::try_from(OBJECT_COUNT).expect("test object count should fit i64") + && status.object.as_str() > "checkpoint-000" + }) + .await + .map_err(|err| format!("{err}; target journal: {:?}", target.requests()))?; + assert!(target.requests().iter().any(|request| { + request.operation == FakeTargetOperation::PutObject + && request.key.as_deref() == Some("checkpoint-000") + && request.fault == Some(FakeTargetFault::Delay(Duration::from_secs(30))) + })); + assert!(!target.has_object(target_bucket, "checkpoint-000")); + source_env.stop_server(); + target.clear_faults(); + + source_env.start_rustfs_server_without_cleanup_with_env(&process_env).await?; + let completed = wait_for_replication_reset_target(&source_env, source_bucket, &target_arn, |status| { + status.reset_id == reset_id && status.status == "Completed" + }) + .await?; + assert_eq!( + completed.replicated_count, + i64::try_from(OBJECT_COUNT).expect("test object count should fit i64") + ); + assert!( + target.has_object(target_bucket, "checkpoint-000"), + "restart skipped failed object checkpoint-000 before persisted checkpoint {}", + partial.object + ); + + target.shutdown().await; + Ok(()) +} + /// P0-5: MinIO derives the replicated version exclusively from the `versionId` /// query parameter (`putOptsFromReq`); the internal x-*-source-version-id /// headers do not exist there. Without the query, a MinIO target mints fresh diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 9edc4fe2d..07bbb1515 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -710,8 +710,8 @@ pub struct ReplicationPool { mrf_save_tx: Sender, mrf_save_rx: Mutex>>, - // Control channels - mrf_worker_kill_tx: Sender<()>, + // MRF worker lifecycle + mrf_worker_cancellations: Mutex>, mrf_stop_tx: Sender<()>, // Worker size tracking @@ -734,7 +734,6 @@ impl ReplicationPool { // Create MRF channels let (mrf_replica_tx, mrf_replica_rx) = mpsc::channel(100000); let (mrf_save_tx, mrf_save_rx) = mpsc::channel(100000); - let (mrf_worker_kill_tx, _mrf_worker_kill_rx) = mpsc::channel(worker_counts.mrf_workers); let (mrf_stop_tx, _mrf_stop_rx) = mpsc::channel(1); let pool = Arc::new(Self { @@ -752,7 +751,7 @@ impl ReplicationPool { mrf_replica_rx: Arc::new(Mutex::new(mrf_replica_rx)), mrf_save_tx, mrf_save_rx: Mutex::new(Some(mrf_save_rx)), - mrf_worker_kill_tx, + mrf_worker_cancellations: Mutex::new(Vec::with_capacity(worker_counts.mrf_workers)), mrf_stop_tx, mrf_worker_size: AtomicI32::new(0), task_handles: Mutex::new(Vec::new()), @@ -896,12 +895,12 @@ impl ReplicationPool { /// Resizes the failed workers pool pub async fn resize_failed_workers(&self, n: i32) { - // Spawn workers up to n. Each worker shares the receiver via Arc>. - // The mutex is held only while calling recv() — released before processing — so - // all workers process entries concurrently (the dequeue step is serialised but - // the replication I/O is not). - while self.mrf_worker_size.load(Ordering::SeqCst) < n { - self.mrf_worker_size.fetch_add(1, Ordering::SeqCst); + let target = mrf_worker_size_to_count(n); + let mut cancellations = self.mrf_worker_cancellations.lock().await; + + while cancellations.len() < target { + let cancellation = CancellationToken::new(); + cancellations.push(cancellation.clone()); let active_counter = self.active_mrf_workers.clone(); let stats = self.stats.clone(); @@ -910,7 +909,18 @@ impl ReplicationPool { let handle = tokio::spawn(async move { loop { - let operation = { mrf_rx.lock().await.recv().await }; + let operation = tokio::select! { + biased; + operation = async { + let mut receiver = mrf_rx.lock().await; + tokio::select! { + biased; + operation = receiver.recv() => operation, + _ = cancellation.cancelled() => None, + } + } => operation, + _ = cancellation.cancelled() => break, + }; let Some(operation) = operation else { break }; let _active = ActiveWorkerGuard::new(active_counter.clone()); @@ -920,11 +930,13 @@ impl ReplicationPool { self.task_handles.lock().await.push(handle); } - // Remove workers if needed - while self.mrf_worker_size.load(Ordering::SeqCst) > n { - self.mrf_worker_size.fetch_sub(1, Ordering::SeqCst); - let _ = self.mrf_worker_kill_tx.try_send(()); + while cancellations.len() > target { + if let Some(cancellation) = cancellations.pop() { + cancellation.cancel(); + } } + + self.mrf_worker_size.store(n.max(0), Ordering::SeqCst); } /// Resizes worker priority and counts @@ -3350,7 +3362,6 @@ mod tests { ) -> Arc> { let (mrf_replica_tx, mrf_replica_rx) = mpsc::channel(1); let (mrf_save_tx, mrf_save_rx) = mpsc::channel(mrf_save_capacity); - let (mrf_worker_kill_tx, _) = mpsc::channel(1); let (mrf_stop_tx, _) = mpsc::channel(1); Arc::new(ReplicationPool { @@ -3368,7 +3379,7 @@ mod tests { mrf_replica_rx: Arc::new(Mutex::new(mrf_replica_rx)), mrf_save_tx, mrf_save_rx: Mutex::new(Some(mrf_save_rx)), - mrf_worker_kill_tx, + mrf_worker_cancellations: Mutex::new(Vec::new()), mrf_stop_tx, mrf_worker_size: AtomicI32::new(0), task_handles: Mutex::new(Vec::new()), @@ -3971,6 +3982,54 @@ mod tests { ); } + #[tokio::test] + async fn resize_failed_workers_cancels_idle_workers() { + let shared = empty_resync_shared_state(); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("mrf-resize", shared))).await; + + pool.resize_failed_workers(4).await; + assert_eq!(pool.mrf_worker_cancellations.lock().await.len(), 4); + assert_eq!(pool.mrf_worker_size.load(Ordering::SeqCst), 4); + + pool.resize_failed_workers(1).await; + + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let finished = pool + .task_handles + .lock() + .await + .iter() + .filter(|handle| handle.is_finished()) + .count(); + if finished == 3 { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("canceled MRF workers should exit while the shared queue is idle"); + + assert_eq!(pool.mrf_worker_cancellations.lock().await.len(), 1); + assert_eq!(pool.mrf_worker_size.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn resize_failed_workers_is_idempotent_across_growth_and_shrink() { + let shared = empty_resync_shared_state(); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("mrf-resize-repeat", shared))).await; + + for target in [2, 4, 1, 4, 4] { + pool.resize_failed_workers(target).await; + assert_eq!( + pool.mrf_worker_cancellations.lock().await.len(), + usize::try_from(target).expect("test worker count should fit usize") + ); + assert_eq!(pool.mrf_worker_size.load(Ordering::SeqCst), target); + } + } + #[test] fn replicate_object_info_from_object_info_preserves_ssec_checksum() { let checksum = bytes::Bytes::from_static(b"ssec-checksum"); diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 3433aa297..dbdcdd476 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -69,16 +69,17 @@ use rustfs_s3_types::EventName; use rustfs_utils::http::{ AMZ_TAGGING_DIRECTIVE, SUFFIX_REPLICATION_RESET, SUFFIX_REPLICATION_STATUS, has_internal_suffix, insert_str, }; -use rustfs_utils::{DEFAULT_SIP_HASH_KEY, sip_hash}; +use rustfs_utils::{DEFAULT_SIP_HASH_KEY, get_env_usize, sip_hash}; #[cfg(test)] use s3s::dto::ReplicationConfiguration; use std::collections::HashMap; use std::fmt::Display; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; use tokio::io::AsyncRead; -use tokio::sync::RwLock; +use tokio::sync::{OwnedSemaphorePermit, RwLock, Semaphore}; use tokio::task::{JoinHandle, JoinSet}; use tokio::time::Duration as TokioDuration; use tokio_util::io::ReaderStream; @@ -86,6 +87,9 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, error, instrument, trace, warn}; const BACKGROUND_WALKDIR_TIMEOUT: TokioDuration = TokioDuration::from_secs(60); +const ENV_REPL_RESYNC_MAX_JOBS: &str = "RUSTFS_REPL_RESYNC_MAX_JOBS"; +const DEFAULT_REPL_RESYNC_MAX_JOBS: usize = 2; +const MAX_REPL_RESYNC_MAX_JOBS: usize = 32; use uuid::Uuid; const EVENT_RESYNC_STATUS_UPDATE_SKIPPED: &str = "replication_resync_status_update_skipped"; @@ -256,11 +260,20 @@ fn resync_status_duration( type ResyncCancelKey = (String, String, String); +fn configured_resync_max_jobs() -> usize { + bounded_resync_max_jobs(get_env_usize(ENV_REPL_RESYNC_MAX_JOBS, DEFAULT_REPL_RESYNC_MAX_JOBS)) +} + +fn bounded_resync_max_jobs(value: usize) -> usize { + value.clamp(1, MAX_REPL_RESYNC_MAX_JOBS) +} + #[derive(Debug)] pub struct ReplicationResyncer { pub status_map: Arc>>, pub worker_size: usize, pub(crate) cancel_tokens: Arc>>, + resync_admission: Arc, } impl ReplicationResyncer { @@ -269,6 +282,14 @@ impl ReplicationResyncer { status_map: Arc::new(RwLock::new(HashMap::new())), worker_size: RESYNC_WORKER_COUNT, cancel_tokens: Arc::new(RwLock::new(HashMap::new())), + resync_admission: Arc::new(Semaphore::new(configured_resync_max_jobs())), + } + } + + async fn acquire_resync_admission(&self, cancellation_token: &CancellationToken) -> Option { + tokio::select! { + permit = self.resync_admission.clone().acquire_owned() => permit.ok(), + _ = cancellation_token.cancelled() => None, } } @@ -603,6 +624,10 @@ impl ReplicationResyncer { } }; + let Some(_resync_admission_permit) = self.acquire_resync_admission(&cancellation_token).await else { + return; + }; + let cfg = match get_replication_config(&opts.bucket).await { Ok(cfg) => cfg, Err(err) => { @@ -715,55 +740,36 @@ impl ReplicationResyncer { } let (tx, mut rx) = tokio::sync::mpsc::channel(100); - - if let Err(err) = storage - .clone() - .walk( - cancellation_token.clone(), - &opts.bucket, - "", - tx.clone(), - WalkOptions::default().with_walkdir_timeouts(BACKGROUND_WALKDIR_TIMEOUT), - ) - .await - { - error!( - event = EVENT_RESYNC_RUNTIME_SKIPPED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, - bucket = %opts.bucket, - arn = %opts.arn, - reason = "walk_failed", - error = %err, - "Replication resync bucket walk failed" - ); - self.resync_bucket_mark_status(ResyncStatusType::ResyncFailed, opts.clone(), storage.clone()) - .await; - return; - } - drop(tx); - - let status = { - self.status_map - .read() + let walk_failed = Arc::new(AtomicBool::new(false)); + let walk_failed_task = walk_failed.clone(); + let walk_storage = storage.clone(); + let walk_cancellation = cancellation_token.clone(); + let walk_bucket = opts.bucket.clone(); + let walk_arn = opts.arn.clone(); + let walk_task = tokio::spawn(async move { + if let Err(err) = walk_storage + .walk( + walk_cancellation, + &walk_bucket, + "", + tx, + WalkOptions::default().with_walkdir_timeouts(BACKGROUND_WALKDIR_TIMEOUT), + ) .await - .get(&opts.bucket) - .and_then(|status| status.targets_map.get(&opts.arn)) - .cloned() - .unwrap_or_default() - }; - - // An empty checkpoint means no per-object progress was persisted before the - // interruption: resume from the beginning, otherwise `object.name != checkpoint` - // below would skip every object and mark the resync completed without work. - let mut last_checkpoint = if (status.resync_status == ResyncStatusType::ResyncStarted - || status.resync_status == ResyncStatusType::ResyncFailed) - && !status.object.is_empty() - { - Some(status.object) - } else { - None - }; + { + walk_failed_task.store(true, Ordering::Relaxed); + error!( + event = EVENT_RESYNC_RUNTIME_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %walk_bucket, + arn = %walk_arn, + reason = "walk_failed", + error = %err, + "Replication resync bucket walk failed" + ); + } + }); let mut worker_txs = Vec::new(); // mpsc, not broadcast: a lagging broadcast receiver returns Err(Lagged) which @@ -773,7 +779,7 @@ impl ReplicationResyncer { let opts_clone = opts.clone(); let self_clone = self.clone(); - let mut futures = Vec::new(); + let mut futures = vec![walk_task]; let results_fut = tokio::spawn(async move { while let Some(st) = results_rx.recv().await { @@ -962,6 +968,8 @@ impl ReplicationResyncer { error = %err, "Failed to receive resync object info" ); + cancellation_token.cancel(); + drop(rx); let worker_failed = finish_resync_workers(worker_txs, results_tx, futures, false).await; if worker_failed { error!( @@ -980,6 +988,7 @@ impl ReplicationResyncer { } if cancellation_token.is_cancelled() { + drop(rx); finish_resync_workers(worker_txs, results_tx, futures, true).await; self.resync_bucket_mark_status(ResyncStatusType::ResyncCanceled, opts.clone(), storage.clone()) .await; @@ -990,14 +999,6 @@ impl ReplicationResyncer { continue; }; - if heal - && let Some(checkpoint) = &last_checkpoint - && &object.name != checkpoint - { - continue; - } - last_checkpoint = None; - let roi = match get_heal_replicate_object_info(&object, &rcfg).await { Ok(roi) => roi, Err(err) => { @@ -1011,6 +1012,8 @@ impl ReplicationResyncer { error = %err, "Failed to classify object for replication resync" ); + cancellation_token.cancel(); + drop(rx); let worker_failed = finish_resync_workers(worker_txs, results_tx, futures, false).await; if worker_failed { error!( @@ -1033,6 +1036,7 @@ impl ReplicationResyncer { } if cancellation_token.is_cancelled() { + drop(rx); finish_resync_workers(worker_txs, results_tx, futures, true).await; self.resync_bucket_mark_status(ResyncStatusType::ResyncCanceled, opts.clone(), storage.clone()) .await; @@ -1052,6 +1056,8 @@ impl ReplicationResyncer { error = %err, "Failed to send resync object to worker" ); + cancellation_token.cancel(); + drop(rx); let worker_failed = finish_resync_workers(worker_txs, results_tx, futures, false).await; if worker_failed { error!( @@ -1072,7 +1078,7 @@ impl ReplicationResyncer { let worker_failed = finish_resync_workers(worker_txs, results_tx, futures, false).await; let target_failed = self.target_has_resync_failures(&opts).await; - let status = if worker_failed || target_failed { + let status = if walk_failed.load(Ordering::Relaxed) || worker_failed || target_failed { ResyncStatusType::ResyncFailed } else { ResyncStatusType::ResyncCompleted @@ -2355,16 +2361,6 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { ..Default::default() }; - if self.target_replication_status(&tgt_client.arn) == ReplicationStatusType::Completed - && !self.existing_obj_resync.is_empty() - && self.existing_obj_resync.must_resync_target(&tgt_client.arn) - { - rinfo.replication_status = ReplicationStatusType::Completed; - rinfo.replication_resynced = true; - - return rinfo; - } - if ReplicationTargetStore::target_is_offline(&tgt_client).await { debug!( event = EVENT_RESYNC_RUNTIME_SKIPPED, @@ -2723,15 +2719,6 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { rinfo.prev_replication_status = object_info.target_replication_status(&tgt_client.arn); - if rinfo.prev_replication_status == ReplicationStatusType::Completed - && !self.existing_obj_resync.is_empty() - && self.existing_obj_resync.must_resync_target(&tgt_client.arn) - { - rinfo.replication_status = ReplicationStatusType::Completed; - rinfo.replication_resynced = true; - return rinfo; - } - let size = match object_info.get_actual_size() { Ok(size) => size, Err(e) => { @@ -3251,6 +3238,48 @@ mod tests { ReplicationTargetStore::register_test_target(target).await; } + #[test] + fn resync_admission_configuration_is_bounded() { + assert_eq!(ENV_REPL_RESYNC_MAX_JOBS, "RUSTFS_REPL_RESYNC_MAX_JOBS"); + assert_eq!(bounded_resync_max_jobs(0), 1); + assert_eq!(bounded_resync_max_jobs(DEFAULT_REPL_RESYNC_MAX_JOBS), 2); + assert_eq!(bounded_resync_max_jobs(1000), MAX_REPL_RESYNC_MAX_JOBS); + } + + #[tokio::test] + async fn resync_admission_limits_jobs_and_wait_is_cancelable() { + let resyncer = ReplicationResyncer { + resync_admission: Arc::new(Semaphore::new(2)), + ..ReplicationResyncer::new().await + }; + let first = resyncer + .acquire_resync_admission(&CancellationToken::new()) + .await + .expect("first resync should acquire admission"); + let second = resyncer + .acquire_resync_admission(&CancellationToken::new()) + .await + .expect("second resync should acquire admission"); + let cancellation = CancellationToken::new(); + let blocked = resyncer.acquire_resync_admission(&cancellation); + tokio::pin!(blocked); + + assert!( + tokio::time::timeout(TokioDuration::from_millis(25), &mut blocked) + .await + .is_err() + ); + cancellation.cancel(); + assert!( + tokio::time::timeout(TokioDuration::from_secs(1), &mut blocked) + .await + .expect("canceled admission wait should finish") + .is_none() + ); + + drop((first, second)); + } + #[test] fn replication_target_offline_error_classifier_is_network_scoped() { assert!(is_replication_target_offline_error(&"put_object dispatch failure: connector error")); diff --git a/rustfs/src/startup_bucket_metadata.rs b/rustfs/src/startup_bucket_metadata.rs index 0f5f0e239..7c7a39f61 100644 --- a/rustfs/src/startup_bucket_metadata.rs +++ b/rustfs/src/startup_bucket_metadata.rs @@ -12,11 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::startup_runtime_sources; use crate::storage_api::startup::bucket_metadata::contract::bucket::{BucketOperations, BucketOptions}; use crate::storage_api::startup::bucket_metadata::{ - ECStore, init_bucket_metadata_sys, reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, - try_migrate_iam_config, + ECStore, get_global_replication_pool, init_bucket_metadata_sys, reconcile_bucket_resync_target_intents, + try_migrate_bucket_metadata, try_migrate_iam_config, }; use std::{ io::{Error, Result}, @@ -60,7 +59,7 @@ pub(crate) async fn init_bucket_metadata_runtime(store: Arc, ctx: Cance init_bucket_metadata_sys(store, buckets.clone()).await; reconcile_bucket_resync_target_intents(&buckets).await?; - if let Some(pool) = startup_runtime_sources::replication_pool_handle() { + if let Some(pool) = get_global_replication_pool() { pool.init_resync(ctx, buckets.clone()).await?; } diff --git a/rustfs/src/startup_runtime_sources.rs b/rustfs/src/startup_runtime_sources.rs index 3b9a1df4b..8b6e33115 100644 --- a/rustfs/src/startup_runtime_sources.rs +++ b/rustfs/src/startup_runtime_sources.rs @@ -13,10 +13,8 @@ // limitations under the License. use crate::config::RustFSBufferConfig; -use crate::runtime_sources::{ - current_outbound_tls_generation as runtime_current_outbound_tls_generation, current_replication_pool_handle, -}; -use crate::storage_api::startup::runtime_sources::{DynReplicationPool, InstanceContext, set_global_rustfs_port}; +use crate::runtime_sources::current_outbound_tls_generation as runtime_current_outbound_tls_generation; +use crate::storage_api::startup::runtime_sources::{InstanceContext, set_global_rustfs_port}; use rustfs_kms::KmsServiceManager; use rustfs_obs::{GlobalError as ObservabilityError, OtelGuard}; use rustfs_tls_runtime::{OutboundTlsMaterial, TlsGeneration}; @@ -92,10 +90,6 @@ pub(crate) fn init_metrics_runtime(ctx: CancellationToken) { rustfs_obs::init_metrics_runtime(ctx); } -pub(crate) fn replication_pool_handle() -> Option> { - current_replication_pool_handle() -} - pub(crate) fn set_put_stage_metrics_enabled(enabled: bool) { rustfs_io_metrics::set_put_stage_metrics_enabled(enabled); } diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index c7b78db98..492a9f187 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -215,8 +215,8 @@ pub(crate) mod startup { } pub(crate) use crate::storage::storage_api::{ - ECStore, init_bucket_metadata_sys, reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, - try_migrate_iam_config, + ECStore, get_global_replication_pool, init_bucket_metadata_sys, reconcile_bucket_resync_target_intents, + try_migrate_bucket_metadata, try_migrate_iam_config, }; } @@ -263,7 +263,7 @@ pub(crate) mod startup { } pub(crate) mod runtime_sources { - pub(crate) use crate::storage::storage_api::{DynReplicationPool, InstanceContext, set_global_rustfs_port}; + pub(crate) use crate::storage::storage_api::{InstanceContext, set_global_rustfs_port}; } pub(crate) mod services {