fix(replication): make resync recovery resilient (#5883)

Co-authored-by: houseme <housemecn@gmail.com>

crates/e2e_test/src/fake_s3_target/mod.rs

@@ -134,6 +134,7 @@ pub struct RequestRecord {
 #[derive(Default)]
 struct ControlState {
     scripts: HashMap<Operation, VecDeque<FaultAction>>,
+    keyed_scripts: HashMap<(Operation, String), VecDeque<FaultAction>>,
     requests: VecDeque<RequestRecord>,
     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::<usize>();
+        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<String>, 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<RequestRecord> {
@@ -420,6 +460,25 @@ fn lock<T>(mutex: &Mutex<T>) -> 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::<usize>() + state.keyed_scripts.values().map(VecDeque::len).sum::<usize>()
+}
+
 #[async_trait]
 impl S3Access for FaultAccess {
     async fn check(&self, context: &mut S3AccessContext<'_>) -> S3Result<()> {
@@ -492,7 +551,12 @@ fn record_request(
     content_length: Option<u64>,
 ) -> Option<RequestFault> {
     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 {

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<dyn Error + Send + Sync>>(())
+            }
+        })
+        .buffer_unordered(16)
+        .collect::<Vec<_>>()
+        .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