diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 1e94e1688..a2ba9b63f 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -4048,17 +4048,23 @@ async fn test_site_replication_allows_private_ca_https_with_ca_cert_pem_real_dua #[tokio::test] #[serial] -async fn test_site_replication_resync_start_cancel_restart_real_dual_node() -> Result<(), Box> { +async fn test_site_replication_resync_lifecycle_survives_real_server_restart() -> Result<(), Box> { init_logging(); + let resync_process_env = [ + ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", "true"), + // Verbose server logging can block startup when this focused test is run + // through a captured test process rather than nextest. + ("RUST_LOG", "error"), + ]; let mut source_env = RustFSTestEnvironment::new().await?; - source_env - .start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV) - .await?; + source_env.capture_log_path = Some(format!("{}/server.log", source_env.temp_dir)); + source_env.start_rustfs_server_with_env(vec![], &resync_process_env).await?; let mut target_env = RustFSTestEnvironment::new().await?; + target_env.capture_log_path = Some(format!("{}/server.log", target_env.temp_dir)); target_env - .start_rustfs_server_without_cleanup_with_env(LOOPBACK_REPLICATION_TARGET_ENV) + .start_rustfs_server_without_cleanup_with_env(&resync_process_env) .await?; let source_bucket = "site-repl-resync-src"; @@ -4106,12 +4112,12 @@ async fn test_site_replication_resync_start_cancel_restart_real_dual_node() -> R 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..32 { + for idx in 0..96 { source_client .put_object() .bucket(source_bucket) .key(format!("resync-object-{idx:02}")) - .body(ByteStream::from(vec![b'x'; 256 * 1024])) + .body(ByteStream::from(vec![b'x'; 512 * 1024])) .send() .await?; } @@ -4119,18 +4125,31 @@ async fn test_site_replication_resync_start_cancel_restart_real_dual_node() -> R let started = site_replication_resync_op(&source_env, "start", &remote_peer).await?; assert_eq!(started.status, "success", "unexpected start result: {:?}", started); assert!( - started - .buckets - .iter() - .any(|bucket| bucket.bucket == source_bucket && matches!(bucket.status.as_str(), "started" | "success")), + started.buckets.iter().any(|bucket| { + bucket.bucket == source_bucket && matches!(bucket.status.as_str(), "started" | "running" | "completed" | "success") + }), "source bucket start status missing: {:?}", started ); assert!(!started.resync_id.is_empty(), "start response omitted the resync id: {:?}", started); let started_reset_id = started.resync_id.clone(); + assert!( + matches!(started.state.as_str(), "pending" | "running"), + "the fixture must keep the first generation active long enough to test duplicate start: {:?}", + started + ); + let duplicate_err = site_replication_resync_op(&source_env, "start", &remote_peer) + .await + .expect_err("duplicate start must be rejected while a generation is active"); + assert!( + duplicate_err.to_string().contains("already active"), + "unexpected duplicate start error: {duplicate_err}" + ); + let canceled = site_replication_resync_op(&source_env, "cancel", &remote_peer).await?; assert_eq!(canceled.status, "success", "unexpected cancel result: {:?}", canceled); + assert_eq!(canceled.state, "canceled"); assert!( canceled .buckets @@ -4139,34 +4158,56 @@ async fn test_site_replication_resync_start_cancel_restart_real_dual_node() -> R "source bucket cancel status missing: {:?}", canceled ); + let canceled_again = site_replication_resync_op(&source_env, "cancel", &remote_peer).await?; + assert_eq!(canceled_again.resync_id, canceled.resync_id, "repeated cancel must be idempotent"); + assert_eq!(canceled_again.state, "canceled"); let canceled_target = wait_for_replication_reset_target(&source_env, source_bucket, &target_arn, |target| target.status == "Canceled").await?; - assert_eq!(canceled_target.status, "Canceled"); assert_eq!(canceled_target.reset_id, started_reset_id); let restarted = site_replication_resync_op(&source_env, "start", &remote_peer).await?; assert_eq!(restarted.status, "success", "unexpected restart result: {:?}", restarted); + assert_ne!(restarted.resync_id, started_reset_id); assert!( + matches!(restarted.state.as_str(), "pending" | "running"), + "the second generation must be active before the process restart: {:?}", restarted - .buckets - .iter() - .any(|bucket| bucket.bucket == source_bucket && matches!(bucket.status.as_str(), "started" | "success")), - "source bucket restart status missing: {:?}", - restarted + ); + let restarted_reset_id = restarted.resync_id.clone(); + + source_env.restart_server_preserving_data(vec![], &resync_process_env).await?; + wait_for_site_replication_enabled(&source_env, 2).await?; + + let after_restart = site_replication_resync_op(&source_env, "status", &remote_peer).await?; + assert_eq!( + after_restart.resync_id, restarted_reset_id, + "server restart changed the durable resync id" + ); + assert_eq!(after_restart.generation, restarted.generation); + assert_eq!(after_restart.created_at, restarted.created_at); + assert!( + matches!(after_restart.state.as_str(), "pending" | "running" | "completed" | "failed"), + "unexpected recovered lifecycle state: {:?}", + after_restart + ); + assert!( + after_restart.buckets.iter().any(|bucket| bucket.bucket == source_bucket), + "durable status lost the source bucket after restart: {:?}", + after_restart ); let restart_snapshot = get_replication_reset_status(&source_env, source_bucket, &target_arn).await?; let restarted_target = wait_for_replication_reset_target(&source_env, source_bucket, &target_arn, |target| { - !target.reset_id.is_empty() && target.reset_id != started_reset_id + target.reset_id == restarted_reset_id }) .await .map_err(|err| { format!( "restart ids: start={} restart={} snapshot={:?}; {err}", - started_reset_id, restarted.resync_id, restart_snapshot.targets + started_reset_id, restarted_reset_id, restart_snapshot.targets ) })?; - assert_ne!(restarted_target.reset_id, started_reset_id); + assert_eq!(restarted_target.reset_id, restarted_reset_id); Ok(()) } diff --git a/crates/madmin/src/site_replication.rs b/crates/madmin/src/site_replication.rs index 1f9313928..79f998325 100644 --- a/crates/madmin/src/site_replication.rs +++ b/crates/madmin/src/site_replication.rs @@ -1156,10 +1156,50 @@ pub struct SRStateEditReq { pub struct ResyncBucketStatus { #[serde(default)] pub bucket: String, + #[serde(rename = "targetArn", default, skip_serializing_if = "String::is_empty")] + pub target_arn: String, #[serde(default)] pub status: String, #[serde(rename = "errorDetail", skip_serializing_if = "String::is_empty", default)] pub err_detail: String, + #[serde( + rename = "createdAt", + default, + with = "time::serde::rfc3339::option", + skip_serializing_if = "Option::is_none" + )] + pub created_at: Option, + #[serde( + rename = "startedAt", + default, + with = "time::serde::rfc3339::option", + skip_serializing_if = "Option::is_none" + )] + pub started_at: Option, + #[serde( + rename = "updatedAt", + default, + with = "time::serde::rfc3339::option", + skip_serializing_if = "Option::is_none" + )] + pub updated_at: Option, + #[serde( + rename = "completedAt", + default, + with = "time::serde::rfc3339::option", + skip_serializing_if = "Option::is_none" + )] + pub completed_at: Option, + #[serde(default, skip_serializing_if = "is_zero_u64")] + pub generation: u64, + #[serde(rename = "replicatedObjects", default, skip_serializing_if = "is_zero_u64")] + pub replicated_objects: u64, + #[serde(rename = "replicatedBytes", default, skip_serializing_if = "is_zero_u64")] + pub replicated_bytes: u64, + #[serde(rename = "failedObjects", default, skip_serializing_if = "is_zero_u64")] + pub failed_objects: u64, + #[serde(rename = "failedBytes", default, skip_serializing_if = "is_zero_u64")] + pub failed_bytes: u64, } #[derive(Debug, Clone, Serialize, Deserialize, Default)] @@ -1170,10 +1210,74 @@ pub struct SRResyncOpStatus { pub resync_id: String, #[serde(default)] pub status: String, + #[serde(default, skip_serializing_if = "String::is_empty")] + pub state: String, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub buckets: Vec, #[serde(rename = "errorDetail", skip_serializing_if = "String::is_empty", default)] pub err_detail: String, + #[serde( + rename = "createdAt", + default, + with = "time::serde::rfc3339::option", + skip_serializing_if = "Option::is_none" + )] + pub created_at: Option, + #[serde( + rename = "startedAt", + default, + with = "time::serde::rfc3339::option", + skip_serializing_if = "Option::is_none" + )] + pub started_at: Option, + #[serde( + rename = "updatedAt", + default, + with = "time::serde::rfc3339::option", + skip_serializing_if = "Option::is_none" + )] + pub updated_at: Option, + #[serde( + rename = "completedAt", + default, + with = "time::serde::rfc3339::option", + skip_serializing_if = "Option::is_none" + )] + pub completed_at: Option, + #[serde(default, skip_serializing_if = "is_zero_u64")] + pub generation: u64, + #[serde(rename = "totalBuckets", default, skip_serializing_if = "is_zero_u64")] + pub total_buckets: u64, + #[serde(rename = "pendingBuckets", default, skip_serializing_if = "is_zero_u64")] + pub pending_buckets: u64, + #[serde(rename = "runningBuckets", default, skip_serializing_if = "is_zero_u64")] + pub running_buckets: u64, + #[serde(rename = "completedBuckets", default, skip_serializing_if = "is_zero_u64")] + pub completed_buckets: u64, + #[serde(rename = "failedBuckets", default, skip_serializing_if = "is_zero_u64")] + pub failed_buckets: u64, + #[serde(rename = "canceledBuckets", default, skip_serializing_if = "is_zero_u64")] + pub canceled_buckets: u64, + #[serde(rename = "replicatedObjects", default, skip_serializing_if = "is_zero_u64")] + pub replicated_objects: u64, + #[serde(rename = "replicatedBytes", default, skip_serializing_if = "is_zero_u64")] + pub replicated_bytes: u64, + #[serde(rename = "failedObjects", default, skip_serializing_if = "is_zero_u64")] + pub failed_objects: u64, + #[serde(rename = "failedBytes", default, skip_serializing_if = "is_zero_u64")] + pub failed_bytes: u64, + #[serde(default, skip_serializing_if = "is_false")] + pub truncated: bool, + #[serde(rename = "nextContinuationToken", default, skip_serializing_if = "String::is_empty")] + pub next_continuation_token: String, +} + +fn is_zero_u64(value: &u64) -> bool { + *value == 0 +} + +fn is_false(value: &bool) -> bool { + !*value } #[derive(Debug, Clone, Serialize, Deserialize, Default)] @@ -1202,7 +1306,7 @@ pub struct SiteNetPerfResult { #[cfg(test)] mod tests { - use super::{PeerInfo, PeerSite}; + use super::{PeerInfo, PeerSite, SRResyncOpStatus}; use serde_json::{Value, json}; const TEST_CA_CERT: &str = "-----BEGIN CERTIFICATE-----\ntest-ca\n-----END CERTIFICATE-----"; @@ -1302,4 +1406,94 @@ mod tests { assert!(peer_debug.contains("skip_tls_verify: false")); assert!(peer_debug.contains("has_custom_ca: true")); } + + #[test] + fn resync_status_legacy_json_defaults_new_lifecycle_fields() { + let legacy_json = json!({ + "op": "start", + "id": "resync-1", + "status": "success", + "buckets": [{ + "bucket": "photos", + "status": "success" + }] + }); + + let status: SRResyncOpStatus = + serde_json::from_value(legacy_json.clone()).expect("legacy resync status should deserialize"); + + assert_eq!(status.generation, 0); + assert!(status.state.is_empty()); + assert!(status.created_at.is_none()); + assert_eq!(status.total_buckets, 0); + assert_eq!(status.replicated_objects, 0); + assert!(!status.truncated); + assert!(status.next_continuation_token.is_empty()); + assert!(status.buckets[0].created_at.is_none()); + assert!(status.buckets[0].target_arn.is_empty()); + assert_eq!(status.buckets[0].generation, 0); + assert_eq!(status.buckets[0].replicated_bytes, 0); + assert_eq!(serde_json::to_value(status).expect("legacy resync status should serialize"), legacy_json); + } + + #[test] + fn resync_status_lifecycle_fields_round_trip_with_exact_json_names() { + let status_json = json!({ + "op": "status", + "id": "resync-2", + "status": "success", + "state": "running", + "createdAt": "2026-07-22T01:00:00Z", + "startedAt": "2026-07-22T01:00:01Z", + "updatedAt": "2026-07-22T01:01:00Z", + "completedAt": "2026-07-22T01:02:00Z", + "generation": 7, + "totalBuckets": 6, + "pendingBuckets": 1, + "runningBuckets": 1, + "completedBuckets": 1, + "failedBuckets": 1, + "canceledBuckets": 2, + "replicatedObjects": 12, + "replicatedBytes": 4096, + "failedObjects": 3, + "failedBytes": 512, + "truncated": true, + "nextContinuationToken": "bucket-page-2", + "buckets": [{ + "bucket": "photos", + "targetArn": "arn:rustfs:replication::peer-a:photos", + "status": "failed", + "errorDetail": "target unavailable", + "createdAt": "2026-07-22T01:00:00Z", + "startedAt": "2026-07-22T01:00:01Z", + "updatedAt": "2026-07-22T01:01:00Z", + "completedAt": "2026-07-22T01:02:00Z", + "generation": 7, + "replicatedObjects": 12, + "replicatedBytes": 4096, + "failedObjects": 3, + "failedBytes": 512 + }] + }); + + let status: SRResyncOpStatus = + serde_json::from_value(status_json.clone()).expect("expanded resync status should deserialize"); + + assert_eq!(status.generation, 7); + assert_eq!(status.state, "running"); + assert_eq!(status.total_buckets, 6); + assert_eq!(status.completed_buckets, 1); + assert_eq!(status.replicated_bytes, 4096); + assert!(status.truncated); + assert_eq!(status.next_continuation_token, "bucket-page-2"); + assert_eq!(status.buckets[0].generation, 7); + assert_eq!(status.buckets[0].target_arn, "arn:rustfs:replication::peer-a:photos"); + assert_eq!(status.buckets[0].failed_objects, 3); + assert!(status.buckets[0].completed_at.is_some()); + assert_eq!( + serde_json::to_value(status).expect("expanded resync status should serialize"), + status_json + ); + } } diff --git a/crates/replication/src/resync.rs b/crates/replication/src/resync.rs index e174d7ed7..f000de47c 100644 --- a/crates/replication/src/resync.rs +++ b/crates/replication/src/resync.rs @@ -154,7 +154,7 @@ impl TargetReplicationResyncStatus { } fn marshal_wire_msg(&self, wr: &mut Vec) -> Result<()> { - rmp::encode::write_map_len(wr, 11)?; + rmp::encode::write_map_len(wr, 12)?; rmp::encode::write_str(wr, "st")?; write_msgp_time(wr, wire_time_or_default(self.start_time))?; rmp::encode::write_str(wr, "lst")?; @@ -177,6 +177,8 @@ impl TargetReplicationResyncStatus { rmp::encode::write_str(wr, &self.bucket)?; rmp::encode::write_str(wr, "obj")?; rmp::encode::write_str(wr, &self.object)?; + rmp::encode::write_str(wr, "err")?; + rmp::encode::write_str(wr, self.error.as_deref().unwrap_or_default())?; Ok(()) } @@ -202,6 +204,10 @@ impl TargetReplicationResyncStatus { "rrc" => out.replicated_count = rmp::decode::read_int(rd)?, "bkt" => out.bucket = read_msgp_str(rd)?, "obj" => out.object = read_msgp_str(rd)?, + "err" => { + let error = read_msgp_str(rd)?; + out.error = (!error.is_empty()).then_some(error); + } _ => skip_msgp_value(rd)?, } } @@ -623,6 +629,7 @@ mod tests { bucket: "bucket-a".to_string(), object: "object-a".to_string(), replicated_count: 7, + error: Some("durable failure".to_string()), ..Default::default() }, ); @@ -634,6 +641,7 @@ mod tests { assert_eq!(got.targets_map["arn:replication:a"].resync_id, "rid-1"); assert_eq!(got.targets_map["arn:replication:a"].resync_status, ResyncStatusType::ResyncStarted); assert_eq!(got.targets_map["arn:replication:a"].replicated_count, 7); + assert_eq!(got.targets_map["arn:replication:a"].error.as_deref(), Some("durable failure")); } #[test] diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index c95c3e76a..3cd0d71a1 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -114,6 +114,8 @@ const SITE_REPL_REMOVE_SUCCESS: &str = "Requested site(s) were removed from clus const SITE_REPL_RESYNC_START: &str = "start"; const SITE_REPL_RESYNC_CANCEL: &str = "cancel"; const SITE_REPL_RESYNC_STATUS: &str = "status"; +const SITE_REPL_RESYNC_DEFAULT_PAGE_SIZE: usize = 100; +const SITE_REPL_RESYNC_MAX_PAGE_SIZE: usize = 1000; const SITE_REPL_MIN_NETPERF_DURATION: Duration = Duration::from_secs(1); const SITE_REPL_MAX_NETPERF_DURATION: Duration = Duration::from_secs(30); const SITE_REPLICATION_PEER_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); @@ -4894,32 +4896,160 @@ fn removed_deployment_ids_for_pending_remove(pending: &PendingRemove, local_peer .collect() } -fn resync_status_for_state( - state: &mut SiteReplicationState, - op_type: &str, - peer: &PeerInfo, - bucket_names: Vec, -) -> SRResyncOpStatus { - let status = SRResyncOpStatus { - op_type: op_type.to_string(), - resync_id: Uuid::new_v4().to_string(), - status: "success".to_string(), - buckets: bucket_names - .into_iter() - .map(|bucket| ResyncBucketStatus { - bucket, - status: if op_type == SITE_REPL_RESYNC_CANCEL { - "canceled".to_string() - } else { - "started".to_string() - }, - ..Default::default() - }) - .collect(), - ..Default::default() +#[derive(Debug, Serialize, Deserialize)] +struct SiteResyncContinuationToken { + id: String, + generation: u64, + offset: usize, +} + +fn site_resync_is_active(status: &SRResyncOpStatus) -> bool { + matches!(status.state.as_str(), "pending" | "running" | "canceling") +} + +fn site_resync_cancel_is_idempotent(status: &SRResyncOpStatus) -> bool { + status.state == "canceled" +} + +fn site_resync_nonnegative(value: i64) -> u64 { + u64::try_from(value.max(0)).unwrap_or_default() +} + +fn site_resync_bucket_state(status: replication::ResyncStatusType) -> &'static str { + match status { + replication::ResyncStatusType::ResyncPending => "pending", + replication::ResyncStatusType::ResyncStarted => "running", + replication::ResyncStatusType::ResyncCompleted => "completed", + replication::ResyncStatusType::ResyncCanceled => "canceled", + replication::ResyncStatusType::ResyncFailed | replication::ResyncStatusType::NoResync => "failed", + } +} + +fn site_bucket_resync_is_active(status: replication::ResyncStatusType) -> bool { + matches!( + status, + replication::ResyncStatusType::ResyncPending | replication::ResyncStatusType::ResyncStarted + ) +} + +fn apply_site_resync_target_status(bucket: &mut ResyncBucketStatus, target: &replication::TargetReplicationResyncStatus) { + bucket.status = site_resync_bucket_state(target.resync_status).to_string(); + bucket.started_at = target.start_time; + bucket.updated_at = target.last_update; + bucket.replicated_objects = site_resync_nonnegative(target.replicated_count); + bucket.replicated_bytes = site_resync_nonnegative(target.replicated_size); + bucket.failed_objects = site_resync_nonnegative(target.failed_count); + bucket.failed_bytes = site_resync_nonnegative(target.failed_size); + bucket.err_detail = target.error.as_deref().map(summarize_peer_error_detail).unwrap_or_default(); + if matches!(bucket.status.as_str(), "completed" | "canceled" | "failed") { + bucket.completed_at = bucket.updated_at; + } +} + +fn summarize_site_resync_status(status: &mut SRResyncOpStatus, now: OffsetDateTime) { + status.total_buckets = status.buckets.len() as u64; + status.pending_buckets = 0; + status.running_buckets = 0; + status.completed_buckets = 0; + status.failed_buckets = 0; + status.canceled_buckets = 0; + status.replicated_objects = 0; + status.replicated_bytes = 0; + status.failed_objects = 0; + status.failed_bytes = 0; + + for bucket in &status.buckets { + match bucket.status.as_str() { + "pending" => status.pending_buckets += 1, + "running" | "started" => status.running_buckets += 1, + "completed" | "success" => status.completed_buckets += 1, + "canceled" => status.canceled_buckets += 1, + _ => status.failed_buckets += 1, + } + status.replicated_objects = status.replicated_objects.saturating_add(bucket.replicated_objects); + status.replicated_bytes = status.replicated_bytes.saturating_add(bucket.replicated_bytes); + status.failed_objects = status.failed_objects.saturating_add(bucket.failed_objects); + status.failed_bytes = status.failed_bytes.saturating_add(bucket.failed_bytes); + } + + status.updated_at = Some(now); + status.status = if status.failed_buckets > 0 { "failed" } else { "success" }.to_string(); + let has_active_buckets = status.pending_buckets > 0 + || status.running_buckets > 0 + || status.buckets.iter().any(|bucket| bucket.status == "conflict"); + status.state = if has_active_buckets { + if status.op_type == SITE_REPL_RESYNC_CANCEL { + "canceling" + } else if status.running_buckets > 0 { + "running" + } else { + "pending" + } + } else if status.failed_buckets > 0 { + "failed" + } else if status.op_type == SITE_REPL_RESYNC_CANCEL || status.canceled_buckets == status.total_buckets { + "canceled" + } else { + "completed" + } + .to_string(); + if matches!(status.state.as_str(), "completed" | "canceled" | "failed") && status.completed_at.is_none() { + status.completed_at = Some(now); + } + status.err_detail = if status.failed_buckets > 0 { + format!("{} of {} buckets failed", status.failed_buckets, status.total_buckets) + } else { + String::new() }; - state.resync_status.insert(peer.deployment_id.clone(), status.clone()); - status +} + +fn site_resync_page(status: &SRResyncOpStatus, limit: usize, offset: usize) -> S3Result { + if offset > status.buckets.len() { + return Err(s3_error!(InvalidRequest, "invalid resync continuation token")); + } + let mut response = status.clone(); + let end = offset.saturating_add(limit).min(status.buckets.len()); + response.buckets = status.buckets[offset..end].to_vec(); + response.truncated = end < status.buckets.len(); + response.next_continuation_token = if response.truncated { + let token = SiteResyncContinuationToken { + id: status.resync_id.clone(), + generation: status.generation, + offset: end, + }; + let encoded = serde_json::to_vec(&token) + .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("encode resync cursor failed: {err}")))?; + URL_SAFE_NO_PAD.encode(encoded) + } else { + String::new() + }; + Ok(response) +} + +fn parse_site_resync_page(query: &HashMap, status: &SRResyncOpStatus) -> S3Result<(usize, usize)> { + let limit = query + .get("limit") + .map(|value| value.parse::()) + .transpose() + .map_err(|_| s3_error!(InvalidRequest, "invalid resync page limit"))? + .unwrap_or(SITE_REPL_RESYNC_DEFAULT_PAGE_SIZE); + if limit == 0 || limit > SITE_REPL_RESYNC_MAX_PAGE_SIZE { + return Err(s3_error!(InvalidRequest, "invalid resync page limit")); + } + let offset = if let Some(value) = query.get("continuationToken") { + let decoded = URL_SAFE_NO_PAD + .decode(value) + .map_err(|_| s3_error!(InvalidRequest, "invalid resync continuation token"))?; + let token: SiteResyncContinuationToken = + serde_json::from_slice(&decoded).map_err(|_| s3_error!(InvalidRequest, "invalid resync continuation token"))?; + if token.id != status.resync_id || token.generation != status.generation { + return Err(s3_error!(InvalidRequest, "stale resync continuation token")); + } + token.offset + } else { + 0 + }; + Ok((limit, offset)) } fn bucket_target_endpoint(target: &BucketTarget) -> String { @@ -4928,8 +5058,10 @@ fn bucket_target_endpoint(target: &BucketTarget) -> String { } fn bucket_target_matches_peer(target: &BucketTarget, peer: &PeerInfo) -> bool { - (!target.deployment_id.is_empty() && target.deployment_id == peer.deployment_id) - || bucket_target_endpoint(target) == canonical_endpoint(&peer.endpoint) + if !target.deployment_id.is_empty() { + return target.deployment_id == peer.deployment_id; + } + bucket_target_endpoint(target) == canonical_endpoint(&peer.endpoint) } fn site_replication_target_arns_by_peer(config: Option<&s3s::dto::ReplicationConfiguration>) -> HashMap { @@ -5580,7 +5712,12 @@ async fn backfill_existing_buckets_after_add( if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) { continue; } - let result = start_site_bucket_resync(name, peer, &resync_id).await; + let manifest = site_bucket_resync_manifest_entry(name, peer, OffsetDateTime::now_utc()).await; + let result = if manifest.target_arn.is_empty() { + manifest + } else { + start_site_bucket_resync(name, &manifest.target_arn, &resync_id).await + }; if result.status == "failed" { warn!( event = EVENT_ADMIN_SITE_REPLICATION_STATE, @@ -5661,10 +5798,60 @@ async fn refresh_bucket_targets_after_endpoint_edit(pending_id: &str, service_ac Ok(()) } -async fn start_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &str) -> ResyncBucketStatus { +async fn site_bucket_resync_manifest_entry(bucket: &str, peer: &PeerInfo, now: OffsetDateTime) -> ResyncBucketStatus { + let mut entry = ResyncBucketStatus { + bucket: bucket.to_string(), + status: "pending".to_string(), + created_at: Some(now), + updated_at: Some(now), + ..Default::default() + }; + let _targets_guard = lock_bucket_targets_metadata(bucket).await; + let (config, _) = match metadata_sys::get_replication_config(bucket).await { + Ok(config) => config, + Err(err) => { + entry.status = "failed".to_string(); + entry.err_detail = summarize_peer_error_detail(&err.to_string()); + return entry; + } + }; + let targets = match metadata_sys::list_bucket_targets(bucket).await { + Ok(targets) => targets, + Err(err) => { + entry.status = "failed".to_string(); + entry.err_detail = summarize_peer_error_detail(&err.to_string()); + return entry; + } + }; + let mut matching = targets + .targets + .iter() + .filter(|target| target.target_type == BucketTargetType::ReplicationService && bucket_target_matches_peer(target, peer)); + let Some(target) = matching.next() else { + entry.status = "failed".to_string(); + entry.err_detail = "no valid remote target found for peer".to_string(); + return entry; + }; + if matching.next().is_some() { + entry.status = "failed".to_string(); + entry.err_detail = "multiple remote targets matched peer".to_string(); + return entry; + } + let (has_arn, existing_object_enabled) = config.has_existing_object_replication(&target.arn); + if !has_arn || !existing_object_enabled { + entry.status = "failed".to_string(); + entry.err_detail = "existing object replication is not enabled for the peer target".to_string(); + return entry; + } + entry.target_arn = target.arn.clone(); + entry +} + +async fn start_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &str) -> ResyncBucketStatus { let mut bucket_status = ResyncBucketStatus { bucket: bucket.to_string(), - status: "started".to_string(), + target_arn: target_arn.to_string(), + status: "running".to_string(), ..Default::default() }; let targets_guard = lock_bucket_targets_metadata(bucket).await; @@ -5687,15 +5874,40 @@ async fn start_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &str } }; + let Some(pool) = current_replication_pool_handle() else { + bucket_status.status = "failed".to_string(); + bucket_status.err_detail = "replication pool is not initialized".to_string(); + return bucket_status; + }; + + let Some(target_index) = targets + .targets + .iter() + .position(|target| target.target_type == BucketTargetType::ReplicationService && target.arn == target_arn) + else { + bucket_status.status = "failed".to_string(); + bucket_status.err_detail = "recorded remote target no longer exists".to_string(); + return bucket_status; + }; + + let existing_reset_id = targets.targets[target_index].reset_id.clone(); + if !existing_reset_id.is_empty() && existing_reset_id != resync_id { + let existing_is_active = pool + .get_bucket_resync_status(bucket) + .await + .ok() + .and_then(|status| status.targets_map.get(target_arn).cloned()) + .is_none_or(|target| target.resync_id != existing_reset_id || site_bucket_resync_is_active(target.resync_status)); + if existing_is_active { + bucket_status.status = "conflict".to_string(); + bucket_status.err_detail = "target belongs to a different active resync operation".to_string(); + return bucket_status; + } + } + let reset_before = Some(OffsetDateTime::now_utc()); let target_arn = { - let Some(target) = targets.targets.iter_mut().find(|target| { - target.target_type == BucketTargetType::ReplicationService && bucket_target_matches_peer(target, peer) - }) else { - bucket_status.status = "failed".to_string(); - bucket_status.err_detail = format!("no valid remote target found for peer {}", peer.deployment_id); - return bucket_status; - }; + let target = &mut targets.targets[target_index]; let (has_arn, existing_object_enabled) = config.has_existing_object_replication(&target.arn); if !has_arn || !existing_object_enabled { @@ -5726,12 +5938,6 @@ async fn start_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &str BucketTargetSys::get().update_all_targets(bucket, Some(&targets)).await; drop(targets_guard); - let Some(pool) = current_replication_pool_handle() else { - bucket_status.status = "failed".to_string(); - bucket_status.err_detail = "replication pool is not initialized".to_string(); - return bucket_status; - }; - if let Err(err) = pool .start_bucket_resync(replication::resync_opts(bucket, target_arn, resync_id, reset_before)) .await @@ -5743,9 +5949,10 @@ async fn start_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &str bucket_status } -async fn cancel_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &str) -> ResyncBucketStatus { +async fn cancel_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &str) -> ResyncBucketStatus { let mut bucket_status = ResyncBucketStatus { bucket: bucket.to_string(), + target_arn: target_arn.to_string(), status: "canceled".to_string(), ..Default::default() }; @@ -5761,18 +5968,32 @@ async fn cancel_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &st }; let Some(target) = targets.targets.iter_mut().find(|target| { - target.target_type == BucketTargetType::ReplicationService - && bucket_target_matches_peer(target, peer) - && target.reset_id == resync_id + target.target_type == BucketTargetType::ReplicationService && target.arn == target_arn && target.reset_id == resync_id }) else { bucket_status.status = "failed".to_string(); - bucket_status.err_detail = format!("no in-progress resync target found for peer {}", peer.deployment_id); + bucket_status.err_detail = "recorded resync target is not in progress".to_string(); return bucket_status; }; + let target_arn = target.arn.clone(); + + let Some(pool) = current_replication_pool_handle() else { + bucket_status.status = "failed".to_string(); + bucket_status.err_detail = "replication pool is not initialized".to_string(); + return bucket_status; + }; + + if let Err(err) = pool + .cancel_bucket_resync(replication::resync_opts(bucket, target_arn, resync_id, None)) + .await + { + bucket_status.status = "failed".to_string(); + bucket_status.err_detail = err.to_string(); + return bucket_status; + } + target.reset_id.clear(); target.reset_before_date = None; - let target_arn = target.arn.clone(); let json_targets = match serde_json::to_vec(&targets) { Ok(json_targets) => json_targets, @@ -5791,23 +6012,91 @@ async fn cancel_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &st BucketTargetSys::get().update_all_targets(bucket, Some(&targets)).await; drop(targets_guard); - let Some(pool) = current_replication_pool_handle() else { - bucket_status.status = "failed".to_string(); - bucket_status.err_detail = "replication pool is not initialized".to_string(); - return bucket_status; - }; - - if let Err(err) = pool - .cancel_bucket_resync(replication::resync_opts(bucket, target_arn, resync_id, None)) - .await - { - bucket_status.status = "failed".to_string(); - bucket_status.err_detail = err.to_string(); - } - bucket_status } +async fn refresh_site_resync_status(mut status: SRResyncOpStatus, peer: &PeerInfo) -> SRResyncOpStatus { + for bucket in &mut status.buckets { + if bucket.target_arn.is_empty() && matches!(bucket.status.as_str(), "pending" | "running" | "started") { + let resolved = site_bucket_resync_manifest_entry(&bucket.bucket, peer, OffsetDateTime::now_utc()).await; + if resolved.target_arn.is_empty() { + bucket.status = "failed".to_string(); + bucket.err_detail = resolved.err_detail; + } else { + bucket.target_arn = resolved.target_arn; + bucket.status = "pending".to_string(); + } + } + } + if let Some(pool) = current_replication_pool_handle() { + for bucket in &mut status.buckets { + if bucket.target_arn.is_empty() || bucket.status == "failed" { + continue; + } + match pool.get_bucket_resync_status(&bucket.bucket).await { + Ok(live) => match live.targets_map.get(&bucket.target_arn) { + Some(target) if target.resync_id == status.resync_id => { + apply_site_resync_target_status(bucket, target); + } + Some(target) if !target.resync_id.is_empty() && site_bucket_resync_is_active(target.resync_status) => { + bucket.status = "conflict".to_string(); + bucket.err_detail = "recorded target belongs to a different resync operation".to_string(); + bucket.updated_at = Some(OffsetDateTime::now_utc()); + } + Some(target) if !target.resync_id.is_empty() => { + bucket.status = "failed".to_string(); + bucket.err_detail = "recorded resync operation was superseded by a terminal bucket resync".to_string(); + bucket.updated_at = Some(OffsetDateTime::now_utc()); + bucket.completed_at = bucket.updated_at; + } + _ if matches!(bucket.status.as_str(), "pending" | "running" | "started") => { + let previous = bucket.clone(); + let mut recovered = + start_site_bucket_resync(&previous.bucket, &previous.target_arn, &status.resync_id).await; + recovered.created_at = previous.created_at; + recovered.started_at = previous.started_at.or(Some(OffsetDateTime::now_utc())); + recovered.updated_at = Some(OffsetDateTime::now_utc()); + recovered.generation = status.generation; + recovered.err_detail = summarize_peer_error_detail(&recovered.err_detail); + *bucket = recovered; + } + _ => {} + }, + Err(err) => { + bucket.err_detail = summarize_peer_error_detail(&err.to_string()); + bucket.updated_at = Some(OffsetDateTime::now_utc()); + } + } + } + } + summarize_site_resync_status(&mut status, OffsetDateTime::now_utc()); + status +} + +async fn persist_site_resync_status(peer_id: &str, status: &SRResyncOpStatus) -> S3Result<()> { + let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let mut state = load_site_replication_state().await?; + if state + .resync_status + .get(peer_id) + .is_some_and(|current| current.resync_id != status.resync_id || current.generation != status.generation) + { + return Err(s3_error!(InvalidRequest, "site replication resync state changed")); + } + state.resync_status.insert(peer_id.to_string(), status.clone()); + save_site_replication_state(&state).await +} + +async fn persist_new_site_resync_status(peer_id: &str, status: &SRResyncOpStatus) -> S3Result<()> { + let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let mut state = load_site_replication_state().await?; + if state.resync_status.get(peer_id).is_some_and(site_resync_is_active) { + return Err(s3_error!(InvalidRequest, "site replication resync is already active")); + } + state.resync_status.insert(peer_id.to_string(), status.clone()); + save_site_replication_state(&state).await +} + fn apply_state_edit_req(mut state: SiteReplicationState, body: SRStateEditReq) -> SiteReplicationState { let Some(incoming_updated_at) = body.updated_at else { return state; @@ -7265,89 +7554,162 @@ pub struct SiteReplicationResyncOpHandler {} impl Operation for SiteReplicationResyncOpHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { validate_site_replication_admin_request(&req, AdminAction::SiteReplicationResyncAction).await?; - let operation = query_pairs(&req.uri).get("operation").cloned().unwrap_or_default(); - // Resolve before the request body is consumed below; the `else` stays - // at its original position so error precedence is unchanged. + let query = query_pairs(&req.uri); + let operation = query.get("operation").cloned().unwrap_or_default(); let resolved_store = object_store_from_req(&req); - let peer: PeerInfo = read_site_replication_json(req, "", false).await?; - let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; - let mut state = load_site_replication_state().await?; - let local_peer = current_local_runtime_peer(&state); - let peer = normalize_peer_info(peer); - if peer.deployment_id == local_peer.deployment_id { - return Err(s3_error!(InvalidRequest, "invalid peer specified - cannot resync to self")); - } - if !state.peers.contains_key(&peer.deployment_id) { - return Err(s3_error!(InvalidRequest, "site replication peer not found")); - } - let Some(store) = resolved_store else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + let requested_peer: PeerInfo = read_site_replication_json(req, "", false).await?; + let _lifecycle_guard = SiteReplicationLifecycleGuard::acquire().await; + let (peer, existing_status) = { + let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let state = load_site_replication_state().await?; + let local_peer = current_local_runtime_peer(&state); + let requested_peer = normalize_peer_info(requested_peer); + if requested_peer.deployment_id == local_peer.deployment_id { + return Err(s3_error!(InvalidRequest, "invalid peer specified - cannot resync to self")); + } + let peer = state + .peers + .get(&requested_peer.deployment_id) + .cloned() + .ok_or_else(|| s3_error!(InvalidRequest, "site replication peer not found"))?; + (peer, state.resync_status.get(&requested_peer.deployment_id).cloned()) }; - let buckets = store.list_bucket(&BucketOptions::default()).await.map_err(ApiError::from)?; - let bucket_names: Vec = buckets.into_iter().map(|bucket| bucket.name).collect(); - let status = match operation.as_str() { + let mut status = match operation.as_str() { SITE_REPL_RESYNC_START => { - let mut status = resync_status_for_state(&mut state, &operation, &peer, vec![]); - let mut bucket_statuses = Vec::new(); + if let Some(existing) = existing_status.as_ref() { + let existing = refresh_site_resync_status(existing.clone(), &peer).await; + persist_site_resync_status(&peer.deployment_id, &existing).await?; + if site_resync_is_active(&existing) { + return Err(s3_error!(InvalidRequest, "site replication resync is already active")); + } + } + let Some(store) = resolved_store else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + let mut bucket_names: Vec = store + .list_bucket(&BucketOptions::default()) + .await + .map_err(ApiError::from)? + .into_iter() + .map(|bucket| bucket.name) + .collect(); + bucket_names.sort(); + let now = OffsetDateTime::now_utc(); + let mut bucket_statuses = Vec::with_capacity(bucket_names.len()); for bucket in bucket_names { - bucket_statuses.push(start_site_bucket_resync(&bucket, &peer, &status.resync_id).await); + bucket_statuses.push(site_bucket_resync_manifest_entry(&bucket, &peer, now).await); } - let failures = bucket_statuses.iter().filter(|bucket| bucket.status == "failed").count(); - if failures == bucket_statuses.len() && !bucket_statuses.is_empty() { - status.status = "failed".to_string(); - status.err_detail = "all buckets resync failed".to_string(); - } else if failures > 0 { - status.err_detail = "partial failure in starting site resync".to_string(); + let mut status = SRResyncOpStatus { + op_type: SITE_REPL_RESYNC_START.to_string(), + resync_id: Uuid::new_v4().to_string(), + status: "success".to_string(), + state: "pending".to_string(), + buckets: bucket_statuses, + created_at: Some(now), + started_at: Some(now), + updated_at: Some(now), + generation: existing_status + .as_ref() + .map_or(1, |existing| existing.generation.saturating_add(1).max(1)), + ..Default::default() + }; + summarize_site_resync_status(&mut status, now); + persist_new_site_resync_status(&peer.deployment_id, &status).await?; + for index in 0..status.buckets.len() { + if status.buckets[index].target_arn.is_empty() || status.buckets[index].status == "failed" { + continue; + } + let previous = status.buckets[index].clone(); + let mut result = start_site_bucket_resync(&previous.bucket, &previous.target_arn, &status.resync_id).await; + result.created_at = previous.created_at; + result.started_at = Some(OffsetDateTime::now_utc()); + result.updated_at = result.started_at; + result.generation = status.generation; + result.err_detail = summarize_peer_error_detail(&result.err_detail); + status.buckets[index] = result; + summarize_site_resync_status(&mut status, OffsetDateTime::now_utc()); + persist_site_resync_status(&peer.deployment_id, &status).await?; } - status.buckets = bucket_statuses; - state.resync_status.insert(peer.deployment_id.clone(), status.clone()); + status = refresh_site_resync_status(status, &peer).await; + persist_site_resync_status(&peer.deployment_id, &status).await?; status } SITE_REPL_RESYNC_CANCEL => { - let Some(existing_status) = state.resync_status.get(&peer.deployment_id).cloned() else { + let Some(existing_status) = existing_status else { return Err(s3_error!(InvalidRequest, "no resync in progress")); }; if existing_status.resync_id.is_empty() { return Err(s3_error!(InvalidRequest, "no resync in progress")); } - let mut status = SRResyncOpStatus { - op_type: operation.clone(), - resync_id: existing_status.resync_id.clone(), - status: "success".to_string(), - ..Default::default() - }; - let mut bucket_statuses = Vec::new(); - for bucket in bucket_names { - bucket_statuses.push(cancel_site_bucket_resync(&bucket, &peer, &existing_status.resync_id).await); + let mut status = refresh_site_resync_status(existing_status, &peer).await; + if status.buckets.iter().any(|bucket| bucket.status == "conflict") { + return Err(s3_error!( + InvalidRequest, + "site replication resync target belongs to a different active operation" + )); } - let failures = bucket_statuses.iter().filter(|bucket| bucket.status == "failed").count(); - if failures == bucket_statuses.len() && !bucket_statuses.is_empty() { - status.status = "failed".to_string(); - status.err_detail = "all buckets resync cancel failed".to_string(); - } else if failures > 0 { - status.err_detail = "partial failure in canceling site resync".to_string(); + if site_resync_cancel_is_idempotent(&status) { + status.op_type = SITE_REPL_RESYNC_CANCEL.to_string(); + for bucket in &status.buckets { + if !bucket.target_arn.is_empty() { + let _ = cancel_site_bucket_resync(&bucket.bucket, &bucket.target_arn, &status.resync_id).await; + } + } + status + } else { + if !site_resync_is_active(&status) { + return Err(s3_error!(InvalidRequest, "no active resync to cancel")); + } + status.op_type = SITE_REPL_RESYNC_CANCEL.to_string(); + status.state = "canceling".to_string(); + status.updated_at = Some(OffsetDateTime::now_utc()); + persist_site_resync_status(&peer.deployment_id, &status).await?; + for index in 0..status.buckets.len() { + if status.buckets[index].target_arn.is_empty() + || matches!(status.buckets[index].status.as_str(), "failed" | "canceled") + { + continue; + } + let previous = status.buckets[index].clone(); + let mut result = + cancel_site_bucket_resync(&previous.bucket, &previous.target_arn, &status.resync_id).await; + result.created_at = previous.created_at; + result.started_at = previous.started_at; + result.updated_at = Some(OffsetDateTime::now_utc()); + result.completed_at = result.updated_at; + result.generation = status.generation; + result.err_detail = summarize_peer_error_detail(&result.err_detail); + status.buckets[index] = result; + summarize_site_resync_status(&mut status, OffsetDateTime::now_utc()); + persist_site_resync_status(&peer.deployment_id, &status).await?; + } + status = refresh_site_resync_status(status, &peer).await; + persist_site_resync_status(&peer.deployment_id, &status).await?; + status } - status.buckets = bucket_statuses; - state.resync_status.insert(peer.deployment_id.clone(), status.clone()); - status } SITE_REPL_RESYNC_STATUS => { - let status = state - .resync_status - .get(&peer.deployment_id) - .cloned() - .unwrap_or_else(|| SRResyncOpStatus { - op_type: SITE_REPL_RESYNC_STATUS.to_string(), - status: "not-found".to_string(), - ..Default::default() - }); - return json_response(&status); + let status = existing_status.unwrap_or_else(|| SRResyncOpStatus { + op_type: SITE_REPL_RESYNC_STATUS.to_string(), + status: "not-found".to_string(), + ..Default::default() + }); + if status.resync_id.is_empty() { + status + } else { + let status = refresh_site_resync_status(status, &peer).await; + persist_site_resync_status(&peer.deployment_id, &status).await?; + status + } } _ => return Err(s3_error!(InvalidRequest, "unsupported resync operation")), }; - save_site_replication_state(&state).await?; - json_response(&status) + status + .buckets + .sort_by(|left, right| left.bucket.cmp(&right.bucket).then(left.target_arn.cmp(&right.target_arn))); + let (limit, offset) = parse_site_resync_page(&query, &status)?; + json_response(&site_resync_page(&status, limit, offset)?) } } @@ -11540,4 +11902,88 @@ mod tests { assert!(rule_ids.contains(&"site-repl-dep-b")); assert!(rule_ids.contains(&"site-repl-dep-c")); } + + #[test] + fn site_resync_summary_reports_partial_failure_and_clamps_counters() { + let now = OffsetDateTime::now_utc(); + let mut running = ResyncBucketStatus { + bucket: "b".to_string(), + target_arn: "arn-b".to_string(), + ..Default::default() + }; + apply_site_resync_target_status( + &mut running, + &replication::TargetReplicationResyncStatus { + resync_status: replication::ResyncStatusType::ResyncStarted, + resync_id: "run-1".to_string(), + replicated_count: 4, + replicated_size: 16, + failed_count: -1, + failed_size: -2, + ..Default::default() + }, + ); + let failed = ResyncBucketStatus { + bucket: "a".to_string(), + target_arn: "arn-a".to_string(), + status: "conflict".to_string(), + err_detail: "durable failure".to_string(), + ..Default::default() + }; + let mut status = SRResyncOpStatus { + op_type: SITE_REPL_RESYNC_START.to_string(), + resync_id: "run-1".to_string(), + buckets: vec![running, failed], + ..Default::default() + }; + + summarize_site_resync_status(&mut status, now); + + assert_eq!(status.status, "failed"); + assert_eq!(status.state, "running"); + assert_eq!(status.running_buckets, 1); + assert_eq!(status.failed_buckets, 1); + assert_eq!(status.replicated_objects, 4); + assert_eq!(status.replicated_bytes, 16); + assert_eq!(status.failed_objects, 0); + assert_eq!(status.failed_bytes, 0); + assert_eq!(status.completed_at, None); + assert!(site_resync_is_active(&status)); + assert!(site_resync_cancel_is_idempotent(&SRResyncOpStatus { + state: "canceled".to_string(), + ..Default::default() + })); + assert!(site_bucket_resync_is_active(replication::ResyncStatusType::ResyncPending)); + assert!(site_bucket_resync_is_active(replication::ResyncStatusType::ResyncStarted)); + assert!(!site_bucket_resync_is_active(replication::ResyncStatusType::ResyncCompleted)); + } + + #[test] + fn site_resync_pagination_is_sorted_and_rejects_stale_cursor() { + let status = SRResyncOpStatus { + resync_id: "run-1".to_string(), + generation: 3, + buckets: ["a", "b", "c"] + .into_iter() + .map(|bucket| ResyncBucketStatus { + bucket: bucket.to_string(), + ..Default::default() + }) + .collect(), + ..Default::default() + }; + let first = site_resync_page(&status, 2, 0).expect("first page should be valid"); + assert!(first.truncated); + assert_eq!(first.buckets.iter().map(|bucket| bucket.bucket.as_str()).collect::>(), ["a", "b"]); + + let query = HashMap::from([ + ("limit".to_string(), "2".to_string()), + ("continuationToken".to_string(), first.next_continuation_token), + ]); + let (_, offset) = parse_site_resync_page(&query, &status).expect("cursor should match operation"); + assert_eq!(offset, 2); + let mut newer = status; + newer.generation += 1; + assert!(parse_site_resync_page(&query, &newer).is_err()); + } } diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index c2ae97208..864fd0bce 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -323,9 +323,7 @@ pub(crate) mod replication { pub(crate) type BucketStats = super::ecstore_bucket::replication::BucketStats; pub(crate) type ReplicationStatusType = super::ecstore_bucket::replication::ReplicationStatusType; pub(crate) type ResyncOpts = super::ecstore_bucket::replication::ResyncOpts; - #[cfg(test)] pub(crate) type ResyncStatusType = super::ecstore_bucket::replication::ResyncStatusType; - #[cfg(test)] pub(crate) type TargetReplicationResyncStatus = super::ecstore_bucket::replication::TargetReplicationResyncStatus; pub(crate) fn resync_opts(