diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 305e6f43a..001ff2bb3 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -368,23 +368,23 @@ impl Drop for SlowReplicationTargetGuard { // Mirrors madmin-go `ResyncTargetsInfo`/`ResyncTarget` json tags — the same // shape `mc replicate resync status` decodes. #[derive(Debug, Clone, serde::Deserialize)] -struct ReplicationResetStatusResponse { +pub(crate) struct ReplicationResetStatusResponse { #[serde(rename = "target", default)] - targets: Vec, + pub(crate) targets: Vec, } #[derive(Debug, Clone, serde::Deserialize)] -struct ReplicationResetStatusTarget { +pub(crate) struct ReplicationResetStatusTarget { #[serde(rename = "arn", default)] - arn: String, + pub(crate) arn: String, #[serde(rename = "resetid", default)] - reset_id: String, + pub(crate) reset_id: String, #[serde(rename = "resyncStatus", default)] - status: String, + pub(crate) status: String, #[serde(rename = "replicationCount", default)] - replicated_count: i64, + pub(crate) replicated_count: i64, #[serde(rename = "object", default)] - object: String, + pub(crate) object: String, } fn extract_xml_tag(xml: &str, tag: &str) -> Option { @@ -2294,7 +2294,7 @@ async fn site_replication_state_edit( /// return the target `(arn, reset_id)`, asserting the response carries the /// madmin `ResyncTargetsInfo` shape (`target[0].arn` / `target[0].resetid`) /// that `mc replicate resync start` decodes. -async fn start_bucket_replication_reset( +pub(crate) async fn start_bucket_replication_reset( env: &RustFSTestEnvironment, bucket: &str, ) -> Result<(String, String), Box> { @@ -2314,7 +2314,7 @@ async fn start_bucket_replication_reset( Ok((arn, reset_id)) } -async fn get_replication_reset_status( +pub(crate) async fn get_replication_reset_status( env: &RustFSTestEnvironment, bucket: &str, arn: &str, diff --git a/crates/e2e_test/src/replication_target_matrix_test.rs b/crates/e2e_test/src/replication_target_matrix_test.rs index 02f82034a..aa688f592 100644 --- a/crates/e2e_test/src/replication_target_matrix_test.rs +++ b/crates/e2e_test/src/replication_target_matrix_test.rs @@ -33,11 +33,11 @@ use crate::common::{init_logging, replication_fast_env}; use crate::fake_s3_target::{BucketMode, FAKE_ACCESS_KEY, FAKE_SECRET_KEY}; -use crate::fake_s3_target::{FakeS3Target, Operation as FakeTargetOperation, RequestRecord}; +use crate::fake_s3_target::{FakeS3Target, FaultAction as FakeTargetFault, Operation as FakeTargetOperation, RequestRecord}; use crate::on_demand_migration::common::{OdmEnvOptions, OdmTestEnv, fake_source_client}; use crate::replication_extension_test::{ - LOOPBACK_REPLICATION_TARGET_ENV, ReplicationTargetOptions, enable_bucket_versioning, put_bucket_replication, - set_replication_target_with_options, + LOOPBACK_REPLICATION_TARGET_ENV, ReplicationTargetOptions, enable_bucket_versioning, get_replication_reset_status, + put_bucket_replication, set_replication_target_with_options, start_bucket_replication_reset, }; use aws_sdk_s3::Client; use aws_sdk_s3::primitives::{ByteStream, DateTime}; @@ -227,6 +227,170 @@ fn expectation(mode: TargetMode, shape: ObjectShape) -> Expectation { .unwrap_or(Expectation::Completed) } +/// rustfs/backlog#2340: a target that mints its own version ids (Wasabi, +/// AWS S3) answers 404 to a HEAD by the source uuid, which the worker used to +/// read as "replica missing" and re-drive the PUT — one more target version +/// per heal, MRF retry or resync. Two re-drive shapes, both must converge on +/// the single version the first PUT created: +/// - the first PUT lands but its response is lost, so the object is FAILED +/// and the scanner heal pass re-drives it; +/// - an existing-object resync re-drives a COMPLETED object unconditionally. +#[tokio::test] +async fn matrix_mint_own_version_ids_redrive_does_not_duplicate() -> TestResult { + init_logging(); + + let target = FakeS3Target::start().await?; + let target_bucket = "matrix-mint-own-redrive-dst".to_string(); + target.create_bucket_with_object_lock(target_bucket.clone()); + target.assign_own_version_ids(true); + + let mut env_vars = replication_fast_env(); + env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); + env_vars.extend_from_slice(&[ + ("NO_PROXY", "127.0.0.1,localhost"), + ("HTTP_PROXY", ""), + ("HTTPS_PROXY", ""), + // The scanner heal pass is what re-drives a FAILED object. + ("RUSTFS_SCANNER_CYCLE", "1"), + ("RUSTFS_SCANNER_START_DELAY_SECS", "1"), + ]); + let env = OdmTestEnv::start_with(OdmEnvOptions { + env: env_vars, + ..OdmEnvOptions::default() + }) + .await?; + let source_env = &env.rustfs; + + let source_bucket = "matrix-mint-own-redrive-src"; + let source_client = source_env.create_s3_client(); + source_client + .create_bucket() + .bucket(source_bucket) + .object_lock_enabled_for_bucket(true) + .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: &target_bucket, + secure: false, + skip_tls_verify: false, + ca_cert_pem: None, + }, + ) + .await?; + put_bucket_replication(source_env, source_bucket, &target_arn).await?; + + // Teach the worker the target's identity contract with one ordinary + // write, exactly as production learns it (the PUT response carries the + // minted id). + let probe_key = "redrive/identity-probe.bin"; + source_client + .put_object() + .bucket(source_bucket) + .key(probe_key) + .body(ByteStream::from(payload(4 * 1024, 0x01))) + .send() + .await?; + assert_eq!( + wait_for_terminal_replication_status(&source_client, source_bucket, probe_key).await?, + "COMPLETED" + ); + + // Shape 1: the PUT is stored, its response never arrives, heal re-drives. + let heal_key = "redrive/heal.bin"; + target.inject_for_key(FakeTargetOperation::PutObject, heal_key, FakeTargetFault::DisconnectAfterResponse, 1); + source_client + .put_object() + .bucket(source_bucket) + .key(heal_key) + .body(ByteStream::from(payload(8 * 1024, 0x02))) + .send() + .await?; + wait_for_replication_status_and_single_version(&source_client, source_bucket, &target, &target_bucket, heal_key).await?; + + // Shape 2: an existing-object resync re-drives a COMPLETED object. + let resync_key = "redrive/resync.bin"; + source_client + .put_object() + .bucket(source_bucket) + .key(resync_key) + .body(ByteStream::from(payload(8 * 1024, 0x03))) + .send() + .await?; + assert_eq!( + wait_for_terminal_replication_status(&source_client, source_bucket, resync_key).await?, + "COMPLETED" + ); + let (reset_arn, _reset_id) = start_bucket_replication_reset(source_env, source_bucket).await?; + assert_eq!(reset_arn, target_arn); + let resync = async { + loop { + let status = get_replication_reset_status(source_env, source_bucket, &target_arn).await?; + if let Some(entry) = status.targets.iter().find(|entry| entry.arn == target_arn) + && entry.status == "Completed" + { + return Ok::<_, Box>(entry.replicated_count); + } + sleep(Duration::from_millis(250)).await; + } + }; + let replicated = timeout(Duration::from_secs(90), resync) + .await + .map_err(|_| "existing-object resync did not complete within 90 seconds")??; + assert!(replicated >= 3, "resync must count the located replicas as replicated, got {replicated}"); + for key in [probe_key, heal_key, resync_key] { + let versions = target.stored_versions(&target_bucket, key); + assert_eq!( + versions.len(), + 1, + "{key}: a re-drive against a target that mints its own version ids must not mint another one: {versions:?}" + ); + } + + target.shutdown().await; + Ok(()) +} + +/// Wait until `key` is COMPLETED on the source and, for the observation +/// window after that, the target still holds exactly one live version of it. +async fn wait_for_replication_status_and_single_version( + source_client: &Client, + source_bucket: &str, + target: &FakeS3Target, + target_bucket: &str, + key: &str, +) -> TestResult { + // The lost PUT response first settles the object FAILED; only the next + // scanner heal pass can turn that into COMPLETED, so FAILED is transient + // here and the wait is for COMPLETED alone. + let converged = async { + loop { + let head = source_client.head_object().bucket(source_bucket).key(key).send().await?; + if head.replication_status().is_some_and(|status| status.as_str() == "COMPLETED") { + return Ok::<_, Box>(()); + } + sleep(Duration::from_millis(250)).await; + } + }; + timeout(Duration::from_secs(90), converged) + .await + .map_err(|_| format!("{key}: heal re-drive did not converge to COMPLETED within 90 seconds"))??; + // The heal pass keeps visiting the key for a few scanner cycles; a + // duplicate would show up here as a second stored version. + for _ in 0..12 { + let versions = target.stored_versions(target_bucket, key); + assert_eq!(versions.len(), 1, "{key}: target minted another version on re-drive: {versions:?}"); + sleep(Duration::from_millis(500)).await; + } + Ok(()) +} + #[tokio::test] async fn matrix_baseline_target() -> TestResult { run_row(TargetMode::Baseline).await diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index acdd95ee5..5c21c4704 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -32,7 +32,7 @@ pub mod bucket { pub mod bucket_target_sys { pub use crate::bucket::bucket_target_sys::{ AdvancedPutOptions, BucketTargetError, BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, - SsecPassthroughCapability, TargetClient, append_version_id_query, + SsecPassthroughCapability, TargetClient, VersionIdentityCapability, append_version_id_query, }; } diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 9cf4ed902..9aa5afbd0 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -18,7 +18,7 @@ use crate::bucket::metadata_sys::get_replication_config; use crate::bucket::remote_s3_client::{ PathStyle, REPLICATION_TARGET_RETRY_POLICY, RemoteCredentials, RemoteS3EndpointSpec, build_remote_s3_client, }; -use crate::bucket::replication::{ObjectLockIntegrity, object_lock_put_integrity}; +use crate::bucket::replication::{ObjectLockIntegrity, object_lock_put_integrity, replication_etags_match}; use crate::bucket::replication::{ReplicationStatusType, ReplicationTargetConfigBridge}; use crate::bucket::target::ARN; use crate::bucket::target::BucketTargetType; @@ -126,6 +126,22 @@ impl From<&BucketTarget> for RemoteS3EndpointSpec { } pub type HeadObjectSdkError = Box>; + +/// Whether an edited bucket target still addresses the same remote service +/// (endpoint, bucket, path style, TLS and identity), so a verdict learned +/// about that service stays valid across the edit. +fn same_replication_service(edited: &BucketTarget, previous: &BucketTarget) -> bool { + let access_key = |target: &BucketTarget| target.credentials.as_ref().map(|credentials| credentials.access_key.clone()); + edited.endpoint == previous.endpoint + && edited.target_bucket == previous.target_bucket + && edited.secure == previous.secure + && edited.path == previous.path + && access_key(edited) == access_key(previous) +} + +/// Page size and page budget for [`TargetClient::find_version_by_etag`]. +const FIND_VERSION_BY_ETAG_PAGE_SIZE: i32 = 1000; +const FIND_VERSION_BY_ETAG_MAX_PAGES: usize = 8; pub type GetObjectSdkError = Box>; pub type GetObjectTaggingSdkError = Box>; pub type PutObjectTaggingSdkError = Box>; @@ -349,6 +365,13 @@ struct TargetClientBuildProbe { /// their import path while the verdict vocabulary lives with the /// replication decision logic. pub use crate::bucket::replication::SsecPassthroughCapability; +/// Version-identity verdicts (see the enum's own docs in +/// `rustfs-replication`) are cached here per target ARN and follow the same +/// `arn_remotes_map` lifecycle. They carry no TTL: the verdict is refreshed +/// by every replication write's response, so it can only go stale on a +/// target that receives no writes — and a stale `MintsOwn` costs one extra +/// content-identity lookup before a PUT, never a lost replica. +pub use crate::bucket::replication::VersionIdentityCapability; /// How long an audited SSE-C passthrough verdict stays authoritative. /// @@ -375,6 +398,11 @@ pub struct BucketTargetSys { /// SSE-C passthrough capability verdicts keyed by target ARN. See /// [`SsecPassthroughCapability`]; reset alongside `arn_remotes_map`. ssec_passthrough_map: Arc>>, + /// Version-identity verdicts keyed by target ARN. See + /// [`VersionIdentityCapability`]; reset alongside `arn_remotes_map`. A std + /// lock (never held across an await) so the replication worker can record + /// a verdict from inside its synchronous PUT-response audit. + version_identity_map: Arc>>, pub targets_map: Arc>>>, /// Buckets whose persisted `bucket-targets.json` exists but cannot be /// decoded (rustfs/backlog#2282). Written under the bucket's update mutex @@ -423,6 +451,7 @@ impl BucketTargetSys { Self { arn_remotes_map: Arc::new(RwLock::new(HashMap::new())), ssec_passthrough_map: Arc::new(RwLock::new(HashMap::new())), + version_identity_map: Arc::new(std::sync::RwLock::new(HashMap::new())), targets_map: Arc::new(RwLock::new(HashMap::new())), unreadable_targets: Arc::new(RwLock::new(HashSet::new())), h_mutex: Arc::new(RwLock::new(HashMap::new())), @@ -746,10 +775,40 @@ impl BucketTargetSys { arn_remotes_map.remove(&target.arn); health_map.remove(&target.arn); ssec_map.remove(&target.arn); + self.forget_version_identity_capability(&target.arn); } } } + /// Cached version-identity verdict for a target ARN; `Unknown` until a + /// replication write or a replication-check VersionFidelity probe judged + /// it since the target was built. + pub fn version_identity_capability(&self, arn: &str) -> VersionIdentityCapability { + self.version_identity_map + .read() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .get(arn) + .copied() + .unwrap_or_default() + } + + /// Record a version-identity verdict for a target ARN. Written by the + /// replication worker after every PutObject / CompleteMultipartUpload + /// response and by the replication-check VersionFidelity phase. + pub fn record_version_identity_capability(&self, arn: &str, capability: VersionIdentityCapability) { + self.version_identity_map + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .insert(arn.to_string(), capability); + } + + fn forget_version_identity_capability(&self, arn: &str) { + self.version_identity_map + .write() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .remove(arn); + } + /// Cached SSE-C passthrough capability for a target ARN, plus whether the /// verdict is older than [`SSEC_PASSTHROUGH_CAPABILITY_TTL`]. `(Unknown, /// false)` when no verdict has been recorded since the target was built. @@ -1162,12 +1221,32 @@ impl BucketTargetSys { // Remove existing targets if let Some(existing_targets) = targets_map.remove(bucket) { let mut ssec_map = self.ssec_passthrough_map.write().await; + let unchanged_service: HashMap<&str, &BucketTarget> = targets + .map(|new_targets| { + new_targets + .targets + .iter() + .map(|target| (target.arn.as_str(), target)) + .collect() + }) + .unwrap_or_default(); for target in existing_targets { arn_remotes_map.remove(&target.arn); health_map.remove(&target.arn); // A rebuilt/edited target may point at a different service: // the SSE-C passthrough verdict must be re-audited from Unknown. ssec_map.remove(&target.arn); + // The version-identity verdict survives an edit that keeps the + // same remote service (a resync start or a bandwidth change + // rewrites the entry in place): forgetting it there would make + // the very resync that follows re-drive every object as a + // duplicate on a target that mints its own version ids. + if unchanged_service + .get(target.arn.as_str()) + .is_none_or(|edited| !same_replication_service(edited, &target)) + { + self.forget_version_identity_capability(&target.arn); + } self.update_bandwidth_limit(bucket, &target.arn, 0); } } @@ -1892,6 +1971,62 @@ impl TargetClient { .map_err(Box::new) } + /// Locate a replica by content identity on a target that mints its own + /// version ids: page `ListObjectVersions` under the exact key and return + /// the newest live version whose ETag matches `source_etag`. Delete + /// markers and prefix siblings never match. Bounded to + /// [`FIND_VERSION_BY_ETAG_MAX_PAGES`] pages so a key with a very deep + /// history cannot turn one convergence check into an unbounded scan; a + /// replica beyond that window reads as missing, which only costs a + /// re-PUT (today's behaviour), never a lost object. + pub async fn find_version_by_etag( + &self, + bucket: &str, + object: &str, + source_etag: &str, + ) -> Result, Box>> { + let mut key_marker: Option = None; + let mut version_id_marker: Option = None; + for _ in 0..FIND_VERSION_BY_ETAG_MAX_PAGES { + let page = self + .client + .list_object_versions() + .bucket(bucket) + .prefix(object) + .max_keys(FIND_VERSION_BY_ETAG_PAGE_SIZE) + .set_key_marker(key_marker.take()) + .set_version_id_marker(version_id_marker.take()) + .send() + .await + .map_err(Box::new)?; + if let Some(version) = page.versions().iter().find(|version| { + version.key() == Some(object) + && version.version_id().is_some_and(|id| !id.is_empty()) + && replication_etags_match(Some(source_etag), version.e_tag()) + }) { + return Ok(version.version_id().map(str::to_string)); + } + // Every listed key is >= the prefix; once the listing moved past + // the exact key there is nothing left to find. + if page + .versions() + .iter() + .any(|version| version.key().is_some_and(|key| key > object)) + { + return Ok(None); + } + if !page.is_truncated().unwrap_or(false) { + return Ok(None); + } + key_marker = page.next_key_marker().map(str::to_string); + version_id_marker = page.next_version_id_marker().map(str::to_string); + if key_marker.is_none() { + return Ok(None); + } + } + Ok(None) + } + /// HEAD used by the read-proxy path (GET/HEAD of an object not yet /// replicated locally, MinIO `proxyHeadToRepTarget`). /// @@ -3221,6 +3356,64 @@ mod tests { assert!(message.contains("connection refused")); } + #[test] + fn same_replication_service_ignores_resync_and_bandwidth_edits() { + let base = BucketTarget { + endpoint: "target.example:9000".to_string(), + target_bucket: "replica".to_string(), + secure: true, + path: "on".to_string(), + arn: "arn:rustfs:replication:us-east-1:bucket:same".to_string(), + credentials: Some(Credentials { + access_key: "access".to_string(), + ..Default::default() + }), + ..Default::default() + }; + let resync_edit = BucketTarget { + reset_id: "reset-1".to_string(), + bandwidth_limit: 1024, + ..base.clone() + }; + assert!(same_replication_service(&resync_edit, &base)); + for moved in [ + BucketTarget { + endpoint: "other.example:9000".to_string(), + ..base.clone() + }, + BucketTarget { + target_bucket: "other".to_string(), + ..base.clone() + }, + BucketTarget { + secure: false, + ..base.clone() + }, + BucketTarget { + credentials: Some(Credentials { + access_key: "rotated".to_string(), + ..Default::default() + }), + ..base.clone() + }, + ] { + assert!(!same_replication_service(&moved, &base)); + } + } + + #[test] + fn version_identity_verdict_is_per_arn_and_forgotten_with_the_target() { + let sys = BucketTargetSys::default(); + let arn = "arn:rustfs:replication:us-east-1:bucket:identity"; + assert_eq!(sys.version_identity_capability(arn), VersionIdentityCapability::Unknown); + sys.record_version_identity_capability(arn, VersionIdentityCapability::MintsOwn); + assert_eq!(sys.version_identity_capability(arn), VersionIdentityCapability::MintsOwn); + assert_eq!(sys.version_identity_capability("other"), VersionIdentityCapability::Unknown); + // A rebuilt target may point at a different service. + sys.forget_version_identity_capability(arn); + assert_eq!(sys.version_identity_capability(arn), VersionIdentityCapability::Unknown); + } + #[test] fn endpoint_health_key_preserves_explicit_port() { let url = Url::parse("https://remote.example:9443").expect("url should parse"); diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index df68abea5..bb9e002d0 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -66,6 +66,7 @@ pub(crate) use replication_lifecycle_bridge::ReplicationLifecycleBridge; pub(crate) use replication_migration_bridge::ReplicationMigrationBridge; pub use replication_object_bridge::ReplicationObjectBridge; pub use replication_object_config::{DeleteReplicationConfigSnapshot, ReplicationConfig}; +pub(crate) use replication_object_decision_boundary::replication_etags_match; pub use replication_object_decision_boundary::{ MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, delete_replication_state_from_config, delete_replication_version_id, should_schedule_delete_replication, should_use_existing_delete_replication_info, @@ -88,5 +89,6 @@ pub use replication_state::{ReplicationStats, RuntimeReplicationTargetBacklog}; pub use replication_stats_boundary::{BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, XferStats}; pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage}; pub use replication_target_boundary::SsecPassthroughCapability; +pub use replication_target_boundary::VersionIdentityCapability; pub use replication_target_boundary::{ObjectLockIntegrity, object_lock_put_integrity}; pub(crate) use replication_target_config_bridge::ReplicationTargetConfigBridge; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 271662095..40aa4c513 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -56,6 +56,8 @@ use super::replication_storage_boundary::{ }; #[cfg(test)] use super::replication_storage_boundary::{NamespaceLockFence, NamespaceLockSignalTestFence, ReplicationDeletedObject}; +#[cfg(test)] +use super::replication_target_boundary::VersionIdentityCapability; use super::replication_target_boundary::{ ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions, RemotePutObjectResponse, ReplicationTargetStore, S3ClientError, SsecPassthroughCapability, SsecPassthroughGate, TargetClient, @@ -63,7 +65,7 @@ use super::replication_target_boundary::{ replication_delete_marker_purge_remove_options, replication_delete_remove_options, replication_force_delete_remove_options, replication_object_is_ssec_encrypted, replication_put_object_header_size, replication_put_object_options, replication_target_head_is_newer_null_version, resolve_read_api_version_id, ssec_passthrough_evidence_present, - ssec_passthrough_gate, version_identity_drifted, + ssec_passthrough_gate, version_identity_capability_from_put, version_identity_drifted, }; use super::replication_versioning_boundary::ReplicationVersioningStore; use super::runtime_boundary as runtime_sources; @@ -123,6 +125,7 @@ const EVENT_DELETE_MARKER_PURGE_FAILED: &str = "replication_delete_marker_purge_ const EVENT_DELETE_MARKER_PURGE_MRF: &str = "replication_delete_marker_purge_mrf"; const METRIC_DELETE_MARKER_PURGE_TOTAL: &str = "rustfs_replication_delete_marker_purge_total"; const EVENT_REPLICATION_VERSION_IDENTITY_DRIFT: &str = "replication_version_identity_drift"; +const EVENT_REPLICATION_DRIFTED_REPLICA_LOCATED: &str = "replication_drifted_replica_located"; const EVENT_REPLICATION_OBJECT_FAILED: &str = "replication_object_failed"; const EVENT_REPLICATION_PURGE_OBJECT_LOCK_DENIED: &str = "replication_purge_object_lock_denied"; @@ -332,6 +335,12 @@ fn verify_single_part_replica( const REPLICA_ETAG_MISMATCH_ERROR: &str = "replica etag mismatch: the target persisted different bytes than were sent"; fn audit_target_version_identity(tgt_client: &TargetClient, source_version_id: &str, assigned_version_id: Option<&str>) { + // Every write refreshes the cached verdict, so the convergence fallback + // below (`replica_head_fallback`) knows whether a 404 on a + // version-addressed HEAD can mean "replica missing" on this target. + if let Some(capability) = version_identity_capability_from_put(source_version_id, assigned_version_id) { + ReplicationTargetStore::record_version_identity_capability(&tgt_client.arn, capability); + } if !version_identity_drifted(source_version_id, assigned_version_id) { return; } @@ -404,11 +413,70 @@ async fn head_object_fallback( ) -> std::result::Result, HeadObjectSdkError> { match head_object_for_worker(tgt_client, &tgt_client.bucket, object, None).await { Ok(oi) => Ok(Some(oi)), - Err(e) if e.as_service_error().is_some_and(|se| se.is_not_found()) || has_raw_status(&e, 404) => Ok(None), + Err(e) if head_object_not_found(&e) => Ok(None), Err(e) => Err(e), } } +fn head_object_not_found(err: &HeadObjectSdkError) -> bool { + err.as_service_error().is_some_and(|se| se.is_not_found()) || has_raw_status(err, 404) +} + +/// Second look at a replica whose version-addressed HEAD failed, for the two +/// target shapes where that failure is not a verdict on the replica: +/// +/// - AWS-style 400/403 (the RustFS uuid is rejected as malformed): HEAD the +/// current version without a version id; callers compare ETags. +/// - 404 on a target known to mint its own version ids (the Wasabi shape, +/// rustfs/backlog#2340): the source id never existed there, so locate the +/// replica by exact key and ETag through ListObjectVersions and HEAD the id +/// the target assigned. Without this, every heal, MRF retry and +/// existing-object resync re-drive PUTs the object again and mints one +/// more target version. +/// +/// `None` when the error stands as-is: a real miss on an adopting target, or +/// a target whose identity contract is still unknown. A failed lookup is +/// returned as a HEAD-shaped error so callers keep their "target operation +/// failed" handling (retry later) instead of re-driving the PUT. +async fn replica_head_fallback( + tgt_client: &TargetClient, + object: &str, + source_etag: Option<&str>, + err: &HeadObjectSdkError, +) -> Option, HeadObjectSdkError>> { + if is_version_id_format_mismatch(err) { + return Some(head_object_fallback(tgt_client, object).await); + } + if !head_object_not_found(err) + || !ReplicationTargetStore::version_identity_capability(&tgt_client.arn).version_addressing_unreliable() + { + return None; + } + let etag = source_etag.filter(|etag| !etag.trim().is_empty())?; + Some(match tgt_client.find_version_by_etag(&tgt_client.bucket, object, etag).await { + Ok(Some(assigned_version_id)) => { + debug!( + event = EVENT_REPLICATION_DRIFTED_REPLICA_LOCATED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %tgt_client.bucket, + object = %object, + arn = %tgt_client.arn, + assigned_version_id = %assigned_version_id, + "Located replica by content identity on a target that mints its own version ids" + ); + match head_object_for_worker(tgt_client, &tgt_client.bucket, object, Some(assigned_version_id)).await { + Ok(oi) => Ok(Some(oi)), + // The located version disappeared between LIST and HEAD. + Err(e) if head_object_not_found(&e) => Ok(None), + Err(e) => Err(e), + } + } + Ok(None) => Ok(None), + Err(list_err) => Err(Box::new(SdkError::construction_failure(*list_err))), + }) +} + /// Resolve the N2 fail-closed gate for an SSE-C passthrough attempt against /// this target. Returns `Some(audit_required)` when replication may proceed; /// on a freshly-flagged header-dropping target it settles `rinfo` as FAILED @@ -1400,31 +1468,26 @@ async fn verify_resync_head_result( (0, None) } } - Err(err) if is_version_id_format_mismatch(&err) => { - // AWS-style target rejects the RustFS UUID versionId - // (400). Re-verify without the versionId before - // concluding the object failed to replicate, instead - // of counting a well-replicated object as failed. - match head_object_fallback(target_client.as_ref(), &roi.name).await { - Ok(Some(_)) => { + Err(err) => { + // A version-addressed HEAD is not the last word on every target: + // re-verify through the fallback before counting a well-replicated + // object as failed (see `replica_head_fallback`). + match replica_head_fallback(target_client.as_ref(), &roi.name, roi.etag.as_deref(), &err).await { + Some(Ok(Some(_))) => { st.replicated_count += 1; st.replicated_size += roi.size; (roi.size, None) } - Ok(None) => { + Some(Ok(None)) | None => { st.failed_count += 1; (0, Some(err)) } - Err(e2) => { + Some(Err(e2)) => { st.failed_count += 1; (0, Some(e2)) } } } - Err(err) => { - st.failed_count += 1; - (0, Some(err)) - } } } @@ -3597,11 +3660,8 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { } } Err(e) => { - if e.as_service_error().is_some_and(|se| se.is_not_found()) || has_raw_status(&e, 404) { - // Object not on target yet → fall through to PUT. - } else if is_version_id_format_mismatch(&e) { - // Version-ID format mismatch: retry without versionId and compare ETags. - match head_object_fallback(&tgt_client, &object).await { + if let Some(fallback) = replica_head_fallback(&tgt_client, &object, object_info.etag.as_deref(), &e).await { + match fallback { Ok(Some(oi)) if replication_etags_match(object_info.etag.as_deref(), oi.e_tag.as_deref()) => { if ssec_audit_required && !settle_ssec_passthrough_evidence(&oi, &tgt_client, &bucket, &object, &mut rinfo).await @@ -3631,6 +3691,8 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } } + } else if head_object_not_found(&e) { + // Object not on target yet → fall through to PUT. } else { rinfo.error = Some(e.to_string()); warn!( @@ -4230,9 +4292,8 @@ async fn resolve_replicate_all_action( } } Err(e) => { - if is_version_id_format_mismatch(&e) { - // Version-ID format mismatch: retry without versionId and compare ETags. - match head_object_fallback(tgt_client, object).await { + if let Some(fallback) = replica_head_fallback(tgt_client, object, object_info.etag.as_deref(), &e).await { + match fallback { Ok(Some(oi)) => { let etags_match = replication_etags_match(object_info.etag.as_deref(), oi.e_tag.as_deref()); if require_existing_target && !etags_match { @@ -4284,7 +4345,7 @@ async fn resolve_replicate_all_action( return None; } } - } else if e.as_service_error().is_some_and(|se| se.is_not_found()) || has_raw_status(&e, 404) { + } else if head_object_not_found(&e) { if require_existing_target { rinfo.error = Some("replica metadata target does not contain this object version".to_string()); rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); @@ -5383,6 +5444,178 @@ mod tests { ReplicationTargetStore::register_test_target(target).await; } + const DRIFTED_ASSIGNED_VERSION_ID: &str = "001788697733811332140-fR6j6uXKV-"; + const DRIFTED_ETAG: &str = "9a0364b9e99bb480dd25e1f0284c8555"; + + /// The Wasabi shape (rustfs/backlog#2340): a version-addressed HEAD with + /// the source uuid answers 404 (not the AWS 400), ListObjectVersions shows + /// the id the target minted, and a HEAD by that id succeeds. Serves exactly + /// `requests` connections and returns the request lines it saw. + fn spawn_drifted_target_server(requests: usize) -> (String, std::thread::JoinHandle>) { + use std::io::{Read, Write}; + + let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).expect("test HTTP listener should bind"); + let endpoint = format!("http://{}", listener.local_addr().expect("test HTTP listener should have an address")); + let handle = std::thread::spawn(move || { + let mut seen = Vec::new(); + for _ in 0..requests { + let (mut stream, _) = listener.accept().expect("test HTTP client should connect"); + let mut request = [0_u8; 8192]; + let bytes_read = stream.read(&mut request).expect("test HTTP request should be read"); + let text = String::from_utf8_lossy(&request[..bytes_read]).to_string(); + let request_line = text.lines().next().unwrap_or_default().to_string(); + let response = if request_line.starts_with("HEAD ") { + if request_line.contains(&format!("versionId={DRIFTED_ASSIGNED_VERSION_ID}")) { + format!( + "HTTP/1.1 200 OK\r\nETag: \"{DRIFTED_ETAG}\"\r\nContent-Length: 4\r\nLast-Modified: Sun, 06 Sep 2026 10:00:00 GMT\r\nConnection: close\r\n\r\n" + ) + } else { + "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_string() + } + } else if request_line.starts_with("GET ") && request_line.contains("versions") { + let body = format!( + "target-bucketobject1000falseobject{DRIFTED_ASSIGNED_VERSION_ID}true2026-09-06T10:00:00.000Z"{DRIFTED_ETAG}"4STANDARD" + ); + format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/xml\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ) + } else { + "HTTP/1.1 500 Unexpected\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_string() + }; + stream + .write_all(response.as_bytes()) + .expect("test HTTP response should be written"); + seen.push(request_line); + } + seen + }); + (endpoint, handle) + } + + fn drifted_roi_and_object() -> (ReplicateObjectInfo, ObjectInfo) { + let roi = ReplicateObjectInfo { + bucket: "source".to_string(), + name: "object".to_string(), + version_id: Some(Uuid::new_v4()), + op_type: ReplicationType::Heal, + replication_status: ReplicationStatusType::Pending, + etag: Some(DRIFTED_ETAG.to_string()), + size: 4, + ..Default::default() + }; + let object_info = ObjectInfo { + bucket: roi.bucket.clone(), + name: roi.name.clone(), + version_id: roi.version_id, + etag: Some(DRIFTED_ETAG.to_string()), + size: 4, + ..Default::default() + }; + (roi, object_info) + } + + #[tokio::test] + async fn heal_redrive_locates_replica_by_etag_on_target_that_mints_own_version_ids() { + let (endpoint, server) = spawn_drifted_target_server(3); + let target = test_target_client(endpoint); + ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn); + let (roi, object_info) = drifted_roi_and_object(); + let mut rinfo = replicate_all_target_info(&roi, &target); + + let action = resolve_replicate_all_action( + ReplicateAllActionContext { + roi: &roi, + tgt_client: &target, + bucket: &roi.bucket, + object: &roi.name, + start_time: OffsetDateTime::now_utc(), + ssec_audit_required: false, + }, + object_info, + &mut rinfo, + ) + .await; + + assert!( + matches!(action, Some((ReplicationAction::None, _))), + "a replica located by content identity must not be re-driven: {action:?}" + ); + assert!(rinfo.error.is_none(), "{:?}", rinfo.error); + let seen = server.join().expect("test HTTP server should finish"); + assert_eq!(seen.len(), 3, "HEAD by source id, ListObjectVersions, HEAD by assigned id: {seen:?}"); + assert!(seen[0].starts_with("HEAD ") && seen[0].contains(&roi.version_id.unwrap().to_string())); + assert!(seen[1].starts_with("GET ") && seen[1].contains("prefix=object"), "{}", seen[1]); + assert!(seen[2].starts_with("HEAD ") && seen[2].contains(DRIFTED_ASSIGNED_VERSION_ID)); + } + + #[tokio::test] + async fn head_not_found_still_replicates_when_identity_contract_is_unknown() { + // Same 404, but the target never revealed whether it adopts version + // ids: a 404 keeps meaning "replica missing" (adopting targets, e.g. + // RustFS/MinIO peers, must not skip a genuinely missing version). + let (endpoint, server) = spawn_head_status_server(404); + let target = test_target_client(endpoint); + let (roi, object_info) = drifted_roi_and_object(); + let mut rinfo = replicate_all_target_info(&roi, &target); + + let action = resolve_replicate_all_action( + ReplicateAllActionContext { + roi: &roi, + tgt_client: &target, + bucket: &roi.bucket, + object: &roi.name, + start_time: OffsetDateTime::now_utc(), + ssec_audit_required: false, + }, + object_info, + &mut rinfo, + ) + .await; + + assert!(matches!(action, Some((ReplicationAction::All, _)))); + server.join().expect("test HTTP server should finish"); + } + + #[tokio::test] + async fn resync_verification_counts_drifted_replica_as_replicated() { + let (endpoint, server) = spawn_drifted_target_server(3); + let target = test_target_client(endpoint); + ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn); + let (roi, _) = drifted_roi_and_object(); + let mut st = TargetReplicationResyncStatus::default(); + + let head_result = + head_object_for_worker(target.as_ref(), &target.bucket, &roi.name, roi.version_id.map(|v| v.to_string())).await; + let (size, err) = verify_resync_head_result(head_result, &roi, &mut st, &target).await; + + assert!(err.is_none(), "{err:?}"); + assert_eq!((size, st.replicated_count, st.failed_count), (4, 1, 0)); + server.join().expect("test HTTP server should finish"); + } + + #[test] + fn put_response_audit_records_identity_verdict() { + let target = test_target_client("http://127.0.0.1:1".to_string()); + let source = Uuid::new_v4().to_string(); + audit_target_version_identity(&target, &source, Some(DRIFTED_ASSIGNED_VERSION_ID)); + assert_eq!( + ReplicationTargetStore::version_identity_capability(&target.arn), + VersionIdentityCapability::MintsOwn + ); + audit_target_version_identity(&target, &source, Some(&source)); + assert_eq!( + ReplicationTargetStore::version_identity_capability(&target.arn), + VersionIdentityCapability::Adopts + ); + // An unversioned write carries no contract and must not overwrite it. + audit_target_version_identity(&target, "null", None); + assert_eq!( + ReplicationTargetStore::version_identity_capability(&target.arn), + VersionIdentityCapability::Adopts + ); + } + #[test] fn resync_admission_configuration_is_bounded() { assert_eq!(ENV_REPL_RESYNC_MAX_JOBS, "RUSTFS_REPL_RESYNC_MAX_JOBS"); diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index ac98affa7..93b6e1768 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -48,6 +48,7 @@ pub use rustfs_replication::{ObjectLockIntegrity, object_lock_put_integrity}; pub(crate) use rustfs_replication::{ SsecPassthroughGate, is_replication_target_offline_error, ssec_passthrough_gate, version_identity_drifted, }; +pub use rustfs_replication::{VersionIdentityCapability, version_identity_capability_from_put}; use super::replication_config_store::ReplicationConfigStore; use super::replication_error_boundary::{Error, Result}; @@ -192,6 +193,14 @@ impl ReplicationTargetStore { .await } + pub(crate) fn version_identity_capability(arn: &str) -> VersionIdentityCapability { + BucketTargetSys::get().version_identity_capability(arn) + } + + pub(crate) fn record_version_identity_capability(arn: &str, capability: VersionIdentityCapability) { + BucketTargetSys::get().record_version_identity_capability(arn, capability) + } + #[cfg(test)] pub(crate) async fn register_test_target(target_client: &Arc) { BucketTargetSys::get().arn_remotes_map.write().await.insert( diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 295340026..3e046b4bd 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -65,9 +65,10 @@ pub use multipart::{ }; pub use object::{ ObjectLockIntegrity, ReplicationSourceObject, ReplicationTargetObject, SsecPassthroughCapability, SsecPassthroughGate, - content_matches_by_etag, is_replication_target_offline_error, object_lock_put_integrity, replication_action_for_target, - replication_etags_match, single_part_replica_etag_mismatch, ssec_passthrough_evidence_present, ssec_passthrough_gate, - target_is_newer_than_source_null_version, version_identity_drifted, + VersionIdentityCapability, content_matches_by_etag, is_replication_target_offline_error, object_lock_put_integrity, + replication_action_for_target, replication_etags_match, single_part_replica_etag_mismatch, ssec_passthrough_evidence_present, + ssec_passthrough_gate, target_is_newer_than_source_null_version, version_identity_capability_from_put, + version_identity_drifted, }; pub use operation::{ MustReplicateOptions, ReplicationDeleteScheduleInput, ReplicationDeleteSource, ReplicationDeleteStateSource, diff --git a/crates/replication/src/object.rs b/crates/replication/src/object.rs index 490821706..5d12f8c5a 100644 --- a/crates/replication/src/object.rs +++ b/crates/replication/src/object.rs @@ -234,18 +234,62 @@ fn comparable_metadata(metadata: Option<&HashMap>) -> HashMap) -> bool { - if source_version_id.is_empty() { - return false; + version_identity_capability_from_put(source_version_id, assigned_version_id) == Some(VersionIdentityCapability::MintsOwn) +} + +/// Whether a replication target adopts the source version id it is handed on +/// PutObject / CompleteMultipartUpload, or mints its own. +/// +/// A target that mints its own ids (AWS S3, Wasabi, Impossible Cloud) still +/// stores the bytes, but every later version-addressed request from the +/// source names an id the target never had. Its HEAD then answers 404 — +/// indistinguishable from a replica that is really missing — so a heal, MRF +/// retry or existing-object resync re-drive would PUT the object again and +/// mint yet another target version (rustfs/backlog#2340). The replication +/// worker learns the verdict from each PUT response (and replication-check's +/// VersionFidelity phase) and, once `MintsOwn` is known, locates a replica by +/// exact key and ETag before concluding that it is missing. The verdict cache +/// is owned by the runtime's bucket target system; this crate owns only the +/// vocabulary and the judgment. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum VersionIdentityCapability { + #[default] + Unknown, + Adopts, + MintsOwn, +} + +impl VersionIdentityCapability { + /// True when a 404 from a version-addressed HEAD on this target cannot be + /// read as "replica missing": the source-side id was never the target's. + pub fn version_addressing_unreliable(self) -> bool { + self == VersionIdentityCapability::MintsOwn + } +} + +/// Judge the identity contract from one replication write: `None` when no +/// contract applies (the source addressed no real version — an empty or nil +/// uuid travels as the literal "null", unversioned-source semantics), +/// otherwise whether the target echoed the source id or answered with +/// anything else — including nothing at all. +pub fn version_identity_capability_from_put( + source_version_id: &str, + assigned_version_id: Option<&str>, +) -> Option { + if source_version_id.is_empty() { + return None; } - // A nil source uuid travels as the literal "null" (unversioned-source - // semantics); no identity contract applies to it. if uuid::Uuid::parse_str(source_version_id) .map(|uuid| uuid.is_nil()) .unwrap_or(true) { - return false; + return None; } - assigned_version_id != Some(source_version_id) + Some(if assigned_version_id == Some(source_version_id) { + VersionIdentityCapability::Adopts + } else { + VersionIdentityCapability::MintsOwn + }) } const REPLICATION_TARGET_OFFLINE_ERROR_MARKERS: &[&str] = &[ @@ -377,9 +421,9 @@ mod tests { use super::{ ReplicationSourceObject, ReplicationTargetObject, SsecPassthroughCapability, SsecPassthroughGate, - content_matches_by_etag, is_replication_target_offline_error, replication_action_for_target, replication_etags_match, - single_part_replica_etag_mismatch, ssec_passthrough_evidence_present, ssec_passthrough_gate, - target_is_newer_than_source_null_version, version_identity_drifted, + VersionIdentityCapability, content_matches_by_etag, is_replication_target_offline_error, replication_action_for_target, + replication_etags_match, single_part_replica_etag_mismatch, ssec_passthrough_evidence_present, ssec_passthrough_gate, + target_is_newer_than_source_null_version, version_identity_capability_from_put, version_identity_drifted, }; use crate::filemeta::{ReplicationAction, ReplicationType}; use crate::http::AMZ_OBJECT_LOCK_MODE; @@ -508,6 +552,32 @@ mod tests { } } + #[test] + fn version_identity_capability_is_judged_only_for_real_source_versions() { + let source = "8e4d2f4c-2d5c-4f1b-9d0a-9c8b7a6f5e4d"; + assert_eq!( + version_identity_capability_from_put(source, Some(source)), + Some(VersionIdentityCapability::Adopts) + ); + // Wasabi / AWS shape: a minted id, or no id at all, both mean the + // source-side id is not addressable on the target. + assert_eq!( + version_identity_capability_from_put(source, Some("001788697733811332140-fR6j6uXKV-")), + Some(VersionIdentityCapability::MintsOwn) + ); + assert_eq!( + version_identity_capability_from_put(source, None), + Some(VersionIdentityCapability::MintsOwn) + ); + // No contract for an unversioned source write. + assert_eq!(version_identity_capability_from_put("", Some("anything")), None); + assert_eq!(version_identity_capability_from_put("00000000-0000-0000-0000-000000000000", None), None); + assert_eq!(version_identity_capability_from_put("null", Some("null")), None); + assert!(VersionIdentityCapability::MintsOwn.version_addressing_unreliable()); + assert!(!VersionIdentityCapability::Adopts.version_addressing_unreliable()); + assert!(!VersionIdentityCapability::Unknown.version_addressing_unreliable()); + } + #[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/admin/router.rs b/rustfs/src/admin/router.rs index 9c69babc5..bb0e06f72 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -18,7 +18,7 @@ use super::storage_api::bucket::replication::{self, BucketReplicationResyncStatu use super::storage_api::bucket::target::{BucketTarget, BucketTargetType, BucketTargets}; use super::storage_api::bucket::target_sys::{ BucketTargetSys, PutObjectOptions, RemoveObjectOptions, S3ClientError, SsecPassthroughCapability, TargetClient, - append_version_id_query, + VersionIdentityCapability, append_version_id_query, }; use super::storage_api::bucket::versioning_sys::BucketVersioningSys; use super::storage_api::bucket::{AdminReplicationConfigExt as _, AdminVersioningConfigExt as _}; @@ -2100,6 +2100,18 @@ async fn check_replication_target( } _ => {} } + // Same for the identity verdict: once a target is known to mint its own + // version ids, the worker locates replicas by content identity instead of + // re-driving PUTs whenever a version-addressed HEAD answers 404. + match (result.phases.version_fidelity.status, result.phases.version_fidelity.code) { + ("OK", _) => { + BucketTargetSys::get().record_version_identity_capability(&target.arn, VersionIdentityCapability::Adopts); + } + ("FAILED", Some(REPLICATION_CHECK_CODE_VERSION_MISMATCH)) => { + BucketTargetSys::get().record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn); + } + _ => {} + } result } diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index c29159a47..f41918068 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -211,6 +211,7 @@ pub(crate) mod bucket_target_sys { pub(crate) type RemoveObjectOptions = super::ecstore_bucket::bucket_target_sys::RemoveObjectOptions; pub(crate) type S3ClientError = super::ecstore_bucket::bucket_target_sys::S3ClientError; pub(crate) type SsecPassthroughCapability = super::ecstore_bucket::bucket_target_sys::SsecPassthroughCapability; + pub(crate) type VersionIdentityCapability = super::ecstore_bucket::bucket_target_sys::VersionIdentityCapability; pub(crate) type TargetClient = super::ecstore_bucket::bucket_target_sys::TargetClient; }