diff --git a/crates/e2e_test/src/replication_target_matrix_test.rs b/crates/e2e_test/src/replication_target_matrix_test.rs index 8fa32e831..743650e72 100644 --- a/crates/e2e_test/src/replication_target_matrix_test.rs +++ b/crates/e2e_test/src/replication_target_matrix_test.rs @@ -303,14 +303,40 @@ fn expectation(mode: TargetMode, shape: ObjectShape) -> Expectation { /// - 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. +/// +/// A permanent delete of the healed version then has to remove exactly the +/// replica the heal pass located. #[tokio::test] async fn matrix_mint_own_version_ids_redrive_does_not_duplicate() -> TestResult { + run_redrive_does_not_duplicate(RedriveObjects::Plaintext).await +} + +/// The same re-drives for SSE-S3 objects against a target that also computes +/// its own ETags (AWS S3). The source ETag is a keyed fingerprint +/// (docs/operations/encrypted-etags.md) while the replica reports the MD5 of +/// the plaintext it received, so no exact-ETag lookup can ever find the +/// replica: the lost-response heal, the resync and the purge must locate it +/// without comparing ETags. +#[tokio::test] +async fn matrix_mint_own_version_ids_encrypted_redrive_does_not_duplicate() -> TestResult { + run_redrive_does_not_duplicate(RedriveObjects::SseS3OnEtagComputingTarget).await +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum RedriveObjects { + Plaintext, + SseS3OnEtagComputingTarget, +} + +async fn run_redrive_does_not_duplicate(objects: RedriveObjects) -> TestResult { init_logging(); + let encrypted = objects == RedriveObjects::SseS3OnEtagComputingTarget; 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); + target.ignore_source_etags(encrypted); let mut env_vars = replication_fast_env(); env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); @@ -324,6 +350,7 @@ async fn matrix_mint_own_version_ids_redrive_does_not_duplicate() -> TestResult ]); let env = OdmTestEnv::start_with(OdmEnvOptions { env: env_vars, + local_kms: encrypted, ..OdmEnvOptions::default() }) .await?; @@ -352,45 +379,63 @@ async fn matrix_mint_own_version_ids_redrive_does_not_duplicate() -> TestResult }, ) .await?; - put_bucket_replication(source_env, source_bucket, &target_arn).await?; + put_bucket_replication_with_delete_statuses(source_env, source_bucket, &target_arn, "Enabled", Some("Enabled")).await?; + + let put = |key: &'static str, fill: u8, len: usize| { + let request = source_client + .put_object() + .bucket(source_bucket) + .key(key) + .body(ByteStream::from(payload(len, fill))); + let request = if encrypted { + request.server_side_encryption(ServerSideEncryption::Aes256) + } else { + request + }; + async move { request.send().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?; + let probe = put(probe_key, 0x01, 4 * 1024).await?; assert_eq!( wait_for_terminal_replication_status(&source_client, source_bucket, probe_key).await?, "COMPLETED" ); + if encrypted { + // The variant only proves something while the replica's ETag differs + // from the source's encrypted-ETag fingerprint. + let replica = single_target_version(&target, &target_bucket, probe_key)?; + let replica_etag = fake_source_client(&target) + .head_object() + .bucket(&target_bucket) + .key(probe_key) + .version_id(&replica) + .send() + .await? + .e_tag() + .map(str::to_string); + if replica_etag.is_none() || replica_etag.as_deref() == probe.e_tag() { + return Err(format!( + "the replica ETag {replica_etag:?} must differ from the source fingerprint {:?}", + probe.e_tag() + ) + .into()); + } + } // 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?; + let healed = put(heal_key, 0x02, 8 * 1024).await?; + let healed_version = healed.version_id().ok_or("source PUT returned no version id")?.to_string(); 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?; + put(resync_key, 0x03, 8 * 1024).await?; assert_eq!( wait_for_terminal_replication_status(&source_client, source_bucket, resync_key).await?, "COMPLETED" @@ -421,6 +466,20 @@ async fn matrix_mint_own_version_ids_redrive_does_not_duplicate() -> TestResult ); } + // The heal pass recorded the replica it located; a permanent delete of + // that version must remove exactly it. + source_client + .delete_object() + .bucket(source_bucket) + .key(heal_key) + .version_id(&healed_version) + .send() + .await?; + wait_until("permanent delete of the healed replica", || async { + Ok(target.stored_versions(&target_bucket, heal_key).is_empty()) + }) + .await?; + target.shutdown().await; Ok(()) } diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index a8edc849d..a5561717e 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -142,11 +142,11 @@ fn same_replication_service(edited: &BucketTarget, previous: &BucketTarget) -> b && access_key(edited) == access_key(previous) } -/// Page size and page budget for [`TargetClient::locate_replica_by_etag`]. +/// Page size and page budget for [`TargetClient::replica_candidates`]. const FIND_VERSION_BY_ETAG_PAGE_SIZE: i32 = 1000; const FIND_VERSION_BY_ETAG_MAX_PAGES: usize = 8; -/// Candidate cap for [`TargetClient::replica_candidates_by_etag`]: more than -/// this many same-content versions of one key is ambiguity by any measure. +/// Candidate cap for [`TargetClient::replica_candidates`]: more than this +/// many matching versions of one key is ambiguity by any measure. const FIND_VERSION_BY_ETAG_MAX_MATCHES: usize = 16; pub type GetObjectSdkError = Box>; pub type GetObjectTaggingSdkError = Box>; @@ -2030,27 +2030,28 @@ impl TargetClient { /// Candidate replicas by content identity on a target that mints its own /// version ids: page `ListObjectVersions` under the exact key and report - /// the live versions whose ETag matches `source_etag`, newest first. - /// Delete markers and prefix siblings never match. Bounded to + /// the live versions that match `identity`, newest first. Delete markers + /// and prefix siblings never match. Bounded to /// [`FIND_VERSION_BY_ETAG_MAX_PAGES`] pages and - /// [`FIND_VERSION_BY_ETAG_MAX_MATCHES`] candidates 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. + /// [`FIND_VERSION_BY_ETAG_MAX_MATCHES`] candidates per kind 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. /// /// Content identity is not version identity: two source generations with - /// the same bytes have the same ETag. Callers drop the candidates other - /// source versions already claim through their ledgers and refuse an - /// [`ReplicaLocation::Ambiguous`] remainder before mutating or deleting. - pub async fn replica_candidates_by_etag( + /// the same bytes have the same ETag, and two with the same size have the + /// same shape. Callers drop the candidates other source versions already + /// claim through their ledgers and decide through + /// [`ReplicaCandidates::locate`]. + pub async fn replica_candidates( &self, bucket: &str, object: &str, - source_etag: &str, - ) -> Result, Box>> { + identity: &ReplicaIdentity<'_>, + ) -> Result>> { let mut key_marker: Option = None; let mut version_id_marker: Option = None; - let mut matches: Vec = Vec::new(); + let mut candidates = ReplicaCandidates::default(); for _ in 0..FIND_VERSION_BY_ETAG_MAX_PAGES { let page = self .client @@ -2063,19 +2064,27 @@ impl TargetClient { .send() .await .map_err(Box::new)?; - matches.extend( - page.versions() - .iter() - .filter(|version| { - version.key() == Some(object) - && version.version_id().is_some_and(|id| !id.is_empty()) - && replication_etags_match(Some(source_etag), version.e_tag()) - }) - .filter_map(|version| version.version_id().map(str::to_string)), - ); + for version in page.versions() { + let Some(version_id) = version.version_id().filter(|id| !id.is_empty()) else { + continue; + }; + if version.key() != Some(object) { + continue; + } + match identity.evidence(version.e_tag(), version.size(), version.last_modified()) { + Some(ReplicaEvidence::Etag) => candidates.exact.push(version_id.to_string()), + // An unversioned target overwrites the key in place: its + // `null` version holds whichever generation was written + // last, so a shape match there identifies nothing. + Some(ReplicaEvidence::Shape) if version_id != "null" => candidates.shaped.push(version_id.to_string()), + Some(ReplicaEvidence::Shape) => {} + None => {} + } + } // A listing that moved past the exact key (every listed key is >= - // the prefix), ended, or already filled the candidate cap decides. - if matches.len() >= FIND_VERSION_BY_ETAG_MAX_MATCHES + // the prefix), ended, or already filled a candidate cap decides. + if candidates.exact.len() >= FIND_VERSION_BY_ETAG_MAX_MATCHES + || candidates.shaped.len() >= FIND_VERSION_BY_ETAG_MAX_MATCHES || page .versions() .iter() @@ -2090,8 +2099,9 @@ impl TargetClient { break; } } - matches.truncate(FIND_VERSION_BY_ETAG_MAX_MATCHES); - Ok(matches) + candidates.exact.truncate(FIND_VERSION_BY_ETAG_MAX_MATCHES); + candidates.shaped.truncate(FIND_VERSION_BY_ETAG_MAX_MATCHES); + Ok(candidates) } /// PutObjectRetention against a replica version on a target that does not @@ -2614,19 +2624,18 @@ impl TargetClient { } /// Where a replica stands on a target that mints its own version ids, by -/// content identity (exact key + ETag) after the candidates other source -/// versions claim were removed. See -/// [`TargetClient::replica_candidates_by_etag`]. +/// content identity after the candidates other source versions claim were +/// removed. See [`TargetClient::replica_candidates`]. #[derive(Debug, Clone, PartialEq, Eq)] pub enum ReplicaLocation { - /// No live version under the key carries the source ETag. + /// No live version under the key matches the source. Missing, - /// Exactly one live version carries it: safe to address. + /// Exactly one live version matches: safe to address. Unique(String), - /// More than one live version carries it (same bytes replicated for - /// several source generations). `newest` is the most recently listed - /// one — good enough to prove the replica exists, never good enough to - /// pick which one to mutate or delete. + /// More than one live version carries the source ETag (same bytes + /// replicated for several source generations). `newest` is the most + /// recently listed one — good enough to prove the replica exists, never + /// good enough to pick which one to mutate or delete. Ambiguous { newest: String }, } @@ -2652,6 +2661,107 @@ impl ReplicaLocation { } } +/// What a replica of an encrypted source version still has in common with +/// it on a target that computes its own ETags. The source ETag is a keyed +/// fingerprint (docs/operations/encrypted-etags.md) or a legacy plaintext +/// MD5 the target's re-encrypted replica never reports, so the replica is +/// recognised by the byte count the replication PUT sent and by a write time +/// no earlier than the source version's own. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct EncryptedReplicaShape { + /// Bytes on the replication wire: the plaintext size for managed SSE, + /// the stored ciphertext size for an SSE-C passthrough. + pub size: i64, + /// The source version's mtime in whole seconds; listings and HEAD report + /// second-precision times. + pub not_before_secs: i64, +} + +impl EncryptedReplicaShape { + pub fn matches(&self, size: Option, last_modified: Option<&aws_sdk_s3::primitives::DateTime>) -> bool { + size == Some(self.size) && last_modified.is_some_and(|written| written.secs() >= self.not_before_secs) + } +} + +/// How a source version recognises its replica among a key's target +/// versions. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReplicaIdentity<'a> { + /// Unencrypted source: its ETag is the MD5 of the replicated bytes, which + /// every target reports back — exact identity. + Etag(&'a str), + /// Encrypted source: a target that adopts the source ETag (RustFS, + /// MinIO) still reports `etag`; any other target only keeps the shape. + Encrypted { + etag: Option<&'a str>, + shape: EncryptedReplicaShape, + }, +} + +/// Which kind of evidence tied a target version to the source. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReplicaEvidence { + /// The target reports the source ETag. + Etag, + /// Encrypted source on a target that computes its own ETag: same size, + /// written no earlier than the source version. + Shape, +} + +impl ReplicaIdentity<'_> { + pub fn evidence( + &self, + etag: Option<&str>, + size: Option, + last_modified: Option<&aws_sdk_s3::primitives::DateTime>, + ) -> Option { + match *self { + Self::Etag(source_etag) => replication_etags_match(Some(source_etag), etag).then_some(ReplicaEvidence::Etag), + Self::Encrypted { + etag: source_etag, + shape, + } => { + if replication_etags_match(source_etag, etag) { + Some(ReplicaEvidence::Etag) + } else { + shape.matches(size, last_modified).then_some(ReplicaEvidence::Shape) + } + } + } + } +} + +/// Content-identity candidates of one key, newest first, by evidence kind. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ReplicaCandidates { + /// Versions that report the source ETag. + pub exact: Vec, + /// Encrypted sources only: versions without the source ETag whose shape + /// matches. + pub shaped: Vec, +} + +impl ReplicaCandidates { + pub fn is_empty(&self) -> bool { + self.exact.is_empty() && self.shaped.is_empty() + } + + /// Drop the candidates other source versions claim and decide. An ETag + /// match outranks any shape match. A shape is weaker evidence than an + /// ETag — two generations of one size share it — so several shape + /// matches are never resolved by picking one: they read as + /// [`ReplicaLocation::Missing`] for convergence (the caller re-PUTs, a + /// duplicate at worst) and as ambiguous for a purge (it refuses). + pub fn locate(mut self, claimed: &HashSet) -> (ReplicaLocation, ReplicaEvidence) { + self.exact.retain(|candidate| !claimed.contains(candidate)); + if !self.exact.is_empty() { + return (ReplicaLocation::from_matches(self.exact), ReplicaEvidence::Etag); + } + self.shaped.retain(|candidate| !claimed.contains(candidate)); + (ReplicaLocation::from_matches(self.shaped), ReplicaEvidence::Shape) + } +} + #[derive(Debug)] pub enum BucketTargetError { BucketRemoteTargetNotFound { diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 0d5a148a0..35736548d 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -36,9 +36,9 @@ use super::replication_object_decision_boundary::{ MustReplicateOptions, ReplicationMultipartPartInput, delete_marker_purge_mrf_entry, delete_marker_purge_version_id, delete_replication_creates_marker, delete_replication_target_version_id, heal_uses_delete_replication_path, is_object_lock_denied_delete, is_retryable_delete_replication_head_error, is_version_delete_replication, - replicate_delete_outcome, replication_etags_match, replication_multipart_complete_actual_size, - replication_multipart_part_plan, replication_single_put_size_error, resync_existing_delete_replication_info, - should_retry_delete_marker_purge, single_part_replica_etag_mismatch, + replicate_delete_outcome, replication_multipart_complete_actual_size, replication_multipart_part_plan, + replication_single_put_size_error, resync_existing_delete_replication_info, should_retry_delete_marker_purge, + single_part_replica_etag_mismatch, }; use super::replication_queue_boundary::{DeletedObjectReplicationInfo, ReplicationQueueAdmission}; use super::replication_resync_boundary::ResyncStatusType; @@ -60,14 +60,15 @@ use super::replication_storage_boundary::{ use super::replication_storage_boundary::{NamespaceLockFence, NamespaceLockSignalTestFence, ReplicationDeletedObject}; use super::replication_tagging_boundary::ReplicationTagFilter; use super::replication_target_boundary::{ - ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions, - RemotePutObjectResponse, ReplicaLocation, ReplicationTargetStore, S3ClientError, SsecPassthroughCapability, - SsecPassthroughGate, TargetClient, VersionIdentityCapability, is_replication_target_offline_error, - replication_action_for_target_head, replication_complete_multipart_options, 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_source_etag_is_comparable, - replication_target_head_is_newer_null_version, resolve_read_api_version_id, ssec_passthrough_evidence_present, - ssec_passthrough_gate, version_identity_capability_from_put, version_identity_drifted, + ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, EncryptedReplicaShape, HeadObjectSdkError, PutObjectOptions, + PutObjectPartOptions, RemotePutObjectResponse, ReplicaCandidates, ReplicaEvidence, ReplicaIdentity, ReplicaLocation, + ReplicationTargetStore, S3ClientError, SsecPassthroughCapability, SsecPassthroughGate, TargetClient, + VersionIdentityCapability, is_replication_target_offline_error, replication_action_for_target_head, + replication_complete_multipart_options, 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_source_etag_is_comparable, replication_target_head_is_newer_null_version, + resolve_read_api_version_id, ssec_passthrough_evidence_present, ssec_passthrough_gate, version_identity_capability_from_put, + version_identity_drifted, }; use super::replication_versioning_boundary::ReplicationVersioningStore; use super::runtime_boundary as runtime_sources; @@ -490,6 +491,7 @@ async fn head_object_fallback( Ok(head) => Ok(Some(LocatedReplica { version_id: head.version_id.clone(), head, + by: LocatedBy::CurrentVersion, })), Err(e) if head_object_not_found(&e) => Ok(None), Err(e) => Err(e), @@ -503,6 +505,63 @@ async fn head_object_fallback( struct LocatedReplica { head: HeadObjectOutput, version_id: Option, + by: LocatedBy, +} + +/// How [`replica_head_fallback`] reached a replica, which decides what may +/// confirm it (see [`located_replica_matches`]). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum LocatedBy { + /// The target version the source ledger recorded for this version. + Ledger, + /// The one candidate content identity left (`ListObjectVersions`). + ContentIdentity, + /// Whatever version is current under the key (AWS-style fallback): + /// nothing ties it to this source version but an ETag match. + CurrentVersion, +} + +/// How this source version recognises its replica on a target. `None` when +/// the source carries nothing to compare (no ETag on a plaintext source). +fn replica_identity(object_info: &ObjectInfo) -> Option> { + let etag = object_info.etag.as_deref().filter(|etag| !etag.trim().is_empty()); + if replication_source_etag_is_comparable(&object_info.user_defined) { + return etag.map(ReplicaIdentity::Etag); + } + // Same byte count the replication PUT sends: SSE-C passes the stored + // ciphertext through, managed SSE sends the decrypted plaintext. + let size = if replication_object_is_ssec_encrypted(&object_info.user_defined) { + object_info.size + } else { + object_info.get_actual_size().ok()? + }; + match object_info.mod_time { + Some(mod_time) if size >= 0 => Some(ReplicaIdentity::Encrypted { + etag, + shape: EncryptedReplicaShape { + size, + not_before_secs: mod_time.unix_timestamp(), + }, + }), + _ => etag.map(ReplicaIdentity::Etag), + } +} + +/// Whether a replica reached through [`replica_head_fallback`] holds this +/// source version. An ETag match always does. For an encrypted source on a +/// target that computes its own ETag, the shape does too — but only for a +/// replica the ledger or content identity singled out, never for whichever +/// version happens to be current. A plaintext source still needs its ETag. +fn located_replica_matches(identity: Option<&ReplicaIdentity<'_>>, located: &LocatedReplica) -> bool { + let Some(identity) = identity else { + return false; + }; + let head = &located.head; + match identity.evidence(head.e_tag(), head.content_length(), head.last_modified()) { + Some(ReplicaEvidence::Etag) => true, + Some(ReplicaEvidence::Shape) => located.by != LocatedBy::CurrentVersion, + None => false, + } } fn head_object_not_found(err: &HeadObjectSdkError) -> bool { @@ -584,15 +643,20 @@ fn ledger_target_version_id(user_defined: &HashMap, arn: &str) - /// 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. +/// - AWS-style 400/403 (the RustFS uuid is rejected as malformed) and /// - 404 on a target that mints its own version ids (the Wasabi shape, -/// rustfs/backlog#2340): the source id never existed there. HEAD the id -/// the source ledger recorded for this target when it has one; otherwise, -/// once the target is known to mint ids, 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. +/// rustfs/backlog#2340): the source id never existed there. +/// +/// Both HEAD the id the source ledger recorded for this target first. Without +/// one, the AWS shape HEADs the current version without a version id +/// (callers compare ETags); the Wasabi shape, once the target is known to +/// mint ids, locates the replica by exact key and content identity through +/// ListObjectVersions and HEADs the id the target assigned. An encrypted +/// source takes that listing on the AWS shape too: its ETag never matches a +/// target-computed one (docs/operations/encrypted-etags.md), so only a unique +/// shape match can stand in for it. Without this, every heal, MRF retry and +/// existing-object resync re-drive PUTs the object again and mints one more +/// target version. /// /// `content_identity` is the sibling-ledger view used to disown candidates /// other source versions already claim; `None` disables content identity @@ -609,7 +673,7 @@ struct ReplicaSource<'a> { bucket: &'a str, object: &'a str, version_id: Option, - etag: Option<&'a str>, + identity: Option>, ledger_version_id: Option<&'a str>, } @@ -623,13 +687,11 @@ async fn replica_head_fallback( bucket: source_bucket, object, version_id: source_version_id, - etag: source_etag, + identity, ledger_version_id, } = source; - if is_version_id_format_mismatch(err) { - return Some(head_object_fallback(tgt_client, object).await); - } - if !head_object_not_found(err) { + let format_mismatch = is_version_id_format_mismatch(err); + if !format_mismatch && !head_object_not_found(err) { return None; } if let Some(ledger_version_id) = ledger_version_id { @@ -641,24 +703,45 @@ async fn replica_head_fallback( return Some(Ok(Some(LocatedReplica { head, version_id: Some(ledger_version_id.to_string()), + by: LocatedBy::Ledger, }))); } // The recorded version is gone: fall through to content identity // before concluding that the replica is missing. - Err(e) if head_object_not_found(&e) => {} + Err(e) if head_object_not_found(&e) || is_version_id_format_mismatch(&e) => {} Err(e) => return Some(Err(e)), } } - if !ReplicationTargetStore::version_identity_capability(&tgt_client.arn).version_addressing_unreliable() { + let encrypted = matches!(identity, Some(ReplicaIdentity::Encrypted { .. })); + if format_mismatch { + if !(encrypted && content_identity.is_some()) { + return Some(head_object_fallback(tgt_client, object).await); + } + } else if !ReplicationTargetStore::version_identity_capability(&tgt_client.arn).version_addressing_unreliable() { return None; } let siblings = content_identity?; - let etag = source_etag.filter(|etag| !etag.trim().is_empty())?; - Some(match tgt_client.replica_candidates_by_etag(&tgt_client.bucket, object, etag).await { + let identity = identity?; + let candidates = match tgt_client.replica_candidates(&tgt_client.bucket, object, &identity).await { + Ok(candidates) => Ok(candidates), + // A target that refused the version-addressed HEAD (a bodiless 403 + // can be a permission denial) may refuse the listing too: keep the + // current-version fallback, which re-PUTs, instead of stalling. + Err(_) if format_mismatch => return Some(head_object_fallback(tgt_client, object).await), + Err(list_err) => Err(list_err), + }; + Some(match candidates { Ok(candidates) => { - let location = + let (location, evidence) = disowned_replica_location(siblings, &tgt_client.arn, source_bucket, object, source_version_id, candidates).await; - match location.any_version_id() { + // Several same-ETag versions all hold the source bytes, so any of + // them proves the replica exists. Several same-shape versions do + // not: re-PUT rather than adopt the wrong generation. + let assigned_version_id = match (&location, evidence) { + (ReplicaLocation::Ambiguous { .. }, ReplicaEvidence::Shape) => None, + _ => location.any_version_id(), + }; + match assigned_version_id { Some(assigned_version_id) => { debug!( event = EVENT_REPLICATION_DRIFTED_REPLICA_LOCATED, @@ -669,6 +752,7 @@ async fn replica_head_fallback( arn = %tgt_client.arn, assigned_version_id = %assigned_version_id, ambiguous = matches!(location, ReplicaLocation::Ambiguous { .. }), + evidence = ?evidence, "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.to_string())) @@ -677,6 +761,7 @@ async fn replica_head_fallback( Ok(head) => Ok(Some(LocatedReplica { head, version_id: Some(assigned_version_id.to_string()), + by: LocatedBy::ContentIdentity, })), // The located version disappeared between LIST and HEAD. Err(e) if head_object_not_found(&e) => Ok(None), @@ -699,16 +784,15 @@ async fn disowned_replica_location( source_bucket: &str, object: &str, source_version_id: Option, - mut candidates: Vec, -) -> ReplicaLocation { + candidates: ReplicaCandidates, +) -> (ReplicaLocation, ReplicaEvidence) { if candidates.is_empty() { - return ReplicaLocation::Missing; + return (ReplicaLocation::Missing, ReplicaEvidence::Etag); } let claimed = siblings .claimed_target_versions(source_bucket, object, source_version_id, arn) .await; - candidates.retain(|candidate| !claimed.contains(candidate)); - ReplicaLocation::from_matches(candidates) + candidates.locate(&claimed) } /// The id the target assigned to a located replica when it is not the source @@ -1855,10 +1939,14 @@ fn spawn_resync_walk_task( /// Classify the target HEAD verification result for one resynced object, /// updating the per-object status counters and returning the accounted size -/// together with any verification error. +/// together with any verification error. `source` is the source version as +/// stored after the resync replicated it (its ledger and encryption decide +/// how the replica is recognised); without it the queued ETag is all there +/// is to go on. async fn verify_resync_head_result( head_result: std::result::Result, roi: &ReplicateObjectInfo, + source: Option<&ObjectInfo>, st: &mut TargetReplicationResyncStatus, target_client: &Arc, ) -> (i64, Option) { @@ -1900,14 +1988,23 @@ async fn verify_resync_head_result( // 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`). + let ledger_version_id = source.and_then(|info| ledger_target_version_id(&info.user_defined, &target_client.arn)); + let identity = match source { + Some(info) => replica_identity(info), + None => roi + .etag + .as_deref() + .filter(|etag| !etag.trim().is_empty()) + .map(ReplicaIdentity::Etag), + }; match replica_head_fallback( target_client.as_ref(), ReplicaSource { bucket: &roi.bucket, object: &roi.name, version_id: roi.version_id, - etag: roi.etag.as_deref(), - ledger_version_id: None, + identity, + ledger_version_id: ledger_version_id.as_deref(), }, Some(&NoSiblingLedger), &err, @@ -1932,6 +2029,24 @@ async fn verify_resync_head_result( } } +/// The resynced source version as stored now, ledger included, for the +/// verification fallback. `None` when it cannot be read. +async fn resync_source_object_info(storage: &Arc, roi: &ReplicateObjectInfo) -> Option { + storage + .get_object_info( + &roi.bucket, + &roi.name, + &ObjectOptions { + version_id: roi.version_id.map(|version_id| version_id.to_string()), + versioned: ReplicationVersioningStore::prefix_enabled(&roi.bucket, &roi.name).await, + version_suspended: ReplicationVersioningStore::prefix_suspended(&roi.bucket, &roi.name).await, + ..Default::default() + }, + ) + .await + .ok() +} + /// Replicate one existing object (or delete marker / version purge) to the /// resync target, verify the outcome via a target HEAD, and produce the /// per-object resync status update. @@ -1966,7 +2081,12 @@ async fn resync_worker_process_object( roi.version_id.map(|v| v.to_string()), ) .await; - let (size, err) = verify_resync_head_result(head_result, &roi, &mut st, target_client).await; + let source = if head_result.is_err() && !roi.delete_marker { + resync_source_object_info(storage, &roi).await + } else { + None + }; + let (size, err) = verify_resync_head_result(head_result, &roi, source.as_ref(), &mut st, target_client).await; if err.is_some() { debug!( @@ -3363,6 +3483,9 @@ fn abandoned_purge_target_info(bucket: &str, dobj: &DeletedObjectReplicationInfo /// to resolve the replica on a target that mints its own version ids. struct VersionPurgeSource { etag: Option, + /// Encrypted sources: what identifies the replica when the target + /// computes its own ETag (see [`replica_identity`]). + encrypted_shape: Option, ledger: HashMap, ledger_corrupt: bool, /// Target versions other generations of the key claim, per target ARN: @@ -3408,8 +3531,13 @@ async fn version_purge_source( claimed_by_siblings.insert(target.arn.clone(), claimed); } } + let encrypted_shape = match replica_identity(&info) { + Some(ReplicaIdentity::Encrypted { shape, .. }) => Some(shape), + _ => None, + }; Some(VersionPurgeSource { etag: info.etag, + encrypted_shape, ledger, ledger_corrupt, claimed_by_siblings, @@ -3427,7 +3555,8 @@ enum PurgeReplicaResolution { /// goal is already met there. Absent, /// Refuse: no ledger entry and more than one target version carries the - /// source ETag, so any pick could destroy a live generation. + /// source ETag (or, for an encrypted source on a target that computes its + /// own ETag, its shape), so any pick could destroy a live generation. Ambiguous, /// Refuse: the persisted ledger is inconsistent. LedgerCorrupt, @@ -3455,15 +3584,19 @@ async fn resolve_purge_replica( if !force_lookup && !ReplicationTargetStore::version_identity_capability(&tgt_client.arn).version_addressing_unreliable() { return PurgeReplicaResolution::SourceVersion; } - let Some(etag) = source.etag.as_deref().filter(|etag| !etag.trim().is_empty()) else { - return PurgeReplicaResolution::SourceVersion; + let etag = source.etag.as_deref().filter(|etag| !etag.trim().is_empty()); + let identity = match (source.encrypted_shape, etag) { + (Some(shape), etag) => ReplicaIdentity::Encrypted { etag, shape }, + (None, Some(etag)) => ReplicaIdentity::Etag(etag), + (None, None) => return PurgeReplicaResolution::SourceVersion, }; - match tgt_client.replica_candidates_by_etag(&tgt_client.bucket, object, etag).await { - Ok(mut candidates) => { - if let Some(claimed) = source.claimed_by_siblings.get(&tgt_client.arn) { - candidates.retain(|candidate| !claimed.contains(candidate)); - } - match ReplicaLocation::from_matches(candidates) { + match tgt_client.replica_candidates(&tgt_client.bucket, object, &identity).await { + Ok(candidates) => { + let no_claims = HashSet::new(); + let claimed = source.claimed_by_siblings.get(&tgt_client.arn).unwrap_or(&no_claims); + // Several shape matches stay ambiguous here: a purge never picks + // one to delete. + match candidates.locate(claimed).0 { ReplicaLocation::Unique(assigned) => PurgeReplicaResolution::Resolved(assigned), ReplicaLocation::Missing => PurgeReplicaResolution::Absent, ReplicaLocation::Ambiguous { .. } => PurgeReplicaResolution::Ambiguous, @@ -4517,13 +4650,14 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { // generation with the same bytes would be mistaken for one. // Only the ledger (a lost response of this very version) and // the AWS-style format fallback apply here. + let identity = replica_identity(&object_info); if let Some(fallback) = replica_head_fallback( &tgt_client, ReplicaSource { bucket: &bucket, object: &object, version_id: self.version_id, - etag: object_info.etag.as_deref(), + identity, ledger_version_id: ledger_version_id.as_deref(), }, None, @@ -4532,9 +4666,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { .await { match fallback { - Ok(Some(located)) - if replication_etags_match(object_info.etag.as_deref(), located.head.e_tag.as_deref()) => - { + Ok(Some(located)) if located_replica_matches(identity.as_ref(), &located) => { if ssec_audit_required && !settle_ssec_passthrough_evidence(&located.head, &tgt_client, &bucket, &object, &mut rinfo) .await @@ -5182,13 +5314,14 @@ async fn resolve_replicate_all_action( } Err(e) => { let ledger_version_id = ledger_target_version_id(&object_info.user_defined, &tgt_client.arn); + let identity = replica_identity(&object_info); if let Some(fallback) = replica_head_fallback( tgt_client, ReplicaSource { bucket, object, version_id: roi.version_id, - etag: object_info.etag.as_deref(), + identity, ledger_version_id: ledger_version_id.as_deref(), }, Some(siblings), @@ -5199,20 +5332,20 @@ async fn resolve_replicate_all_action( match fallback { Ok(Some(located)) => { let oi = &located.head; - let etags_match = replication_etags_match(object_info.etag.as_deref(), oi.e_tag.as_deref()); - if require_existing_target && !etags_match { + let replica_matches = located_replica_matches(identity.as_ref(), &located); + if require_existing_target && !replica_matches { rinfo.error = Some("replica metadata target does not contain matching object data".to_string()); rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); return None; } - if etags_match + if replica_matches && ssec_audit_required && !settle_ssec_passthrough_evidence(oi, tgt_client, bucket, object, rinfo).await { rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); return None; } - if etags_match && let Some(assigned) = drifted_replica_version_id(&located, roi.version_id) { + if replica_matches && let Some(assigned) = drifted_replica_version_id(&located, roi.version_id) { // The replica lives under an id the target minted: // record it (the ledger every later mutation // resolves through) and apply metadata in place — @@ -5245,7 +5378,7 @@ async fn resolve_replicate_all_action( // records the sync without running the PUT. replication_action = ReplicationAction::None; } else { - replication_action = if etags_match { + replication_action = if replica_matches { ReplicationAction::None } else { ReplicationAction::All @@ -6655,13 +6788,361 @@ mod tests { 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; + let (size, err) = verify_resync_head_result(head_result, &roi, None, &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"); } + /// What a target that computes its own ETag (AWS S3) reports for the + /// replica of an encrypted source: the MD5 of the plaintext it received, + /// never the source's encrypted-ETag fingerprint ([`DRIFTED_ETAG`] here). + const COMPUTED_REPLICA_ETAG: &str = "0f0e0d0c0b0a09080706050403020100"; + /// The replica's write time on the scripted targets below. + const REPLICA_WRITTEN: &str = "2026-09-06T10:00:00.000Z"; + const REPLICA_WRITTEN_HTTP: &str = "Sun, 06 Sep 2026 10:00:00 GMT"; + + /// An SSE-S3 source version: 4 plaintext bytes, stored as ciphertext, its + /// ETag a keyed fingerprint, written at `written` (unix seconds). + fn encrypted_roi_and_object(written: i64) -> (ReplicateObjectInfo, ObjectInfo) { + let (roi, mut object_info) = drifted_roi_and_object(); + object_info.size = 36; + object_info.actual_size = 4; + object_info.mod_time = Some(OffsetDateTime::from_unix_timestamp(written).expect("valid timestamp")); + object_info.user_defined = HashMap::from([("x-amz-server-side-encryption".to_string(), "AES256".to_string())]).into(); + (roi, object_info) + } + + fn replica_written_secs() -> i64 { + OffsetDateTime::parse(REPLICA_WRITTEN, &Rfc3339) + .expect("valid replica time") + .unix_timestamp() + } + + fn computed_etag_versions_response(versions: &[(&str, i64)]) -> String { + let versions = versions + .iter() + .enumerate() + .map(|(index, (version_id, size))| { + format!( + "object{version_id}{}{REPLICA_WRITTEN}"{COMPUTED_REPLICA_ETAG}"{size}STANDARD", + index == 0 + ) + }) + .collect::(); + xml_response( + "200 OK", + format!( + "target-bucketobject1000false{versions}" + ), + ) + } + + fn computed_etag_head_response(version_id: &str, size: i64) -> String { + format!( + "HTTP/1.1 200 OK\r\nETag: \"{COMPUTED_REPLICA_ETAG}\"\r\nContent-Length: {size}\r\nx-amz-version-id: {version_id}\r\nLast-Modified: {REPLICA_WRITTEN_HTTP}\r\nConnection: close\r\n\r\n" + ) + } + + /// A Wasabi/AWS-shaped target that computes its own ETags: 404 on the + /// source uuid, the listed versions, and a HEAD for any listed id. + fn spawn_computed_etag_target( + requests: usize, + versions: &'static [(&'static str, i64)], + ) -> (String, std::thread::JoinHandle>) { + spawn_scripted_target_server(requests, move |line| { + if line.starts_with("GET ") && line.contains("versions") { + return computed_etag_versions_response(versions); + } + if line.starts_with("HEAD ") + && let Some((version_id, size)) = versions + .iter() + .find(|(version_id, _)| line.contains(&format!("versionId={version_id}"))) + { + return computed_etag_head_response(version_id, *size); + } + if line.starts_with("HEAD ") { + return empty_response("404 Not Found"); + } + empty_response("500 Unexpected") + }) + } + + async fn resolve_heal_action( + target: &Arc, + roi: &ReplicateObjectInfo, + object_info: ObjectInfo, + rinfo: &mut ReplicatedTargetInfo, + ) -> Option<(ReplicationAction, ObjectInfo)> { + resolve_replicate_all_action( + ReplicateAllActionContext { + siblings: &NoSiblingLedger, + roi, + tgt_client: target, + bucket: &roi.bucket, + object: &roi.name, + start_time: OffsetDateTime::now_utc(), + ssec_audit_required: false, + }, + object_info, + rinfo, + ) + .await + } + + #[tokio::test] + async fn heal_redrive_locates_encrypted_replica_by_shape_on_target_that_computes_own_etags() { + static VERSIONS: &[(&str, i64)] = &[(DRIFTED_ASSIGNED_VERSION_ID, 4)]; + let (endpoint, server) = spawn_computed_etag_target(3, VERSIONS); + let target = test_target_client(endpoint); + ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn); + let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() - 1); + let mut rinfo = replicate_all_target_info(&roi, &target); + + let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await; + + assert!( + matches!(action, Some((ReplicationAction::None, _))), + "an encrypted replica located by its shape must not be re-driven: {action:?}" + ); + assert!(rinfo.error.is_none(), "{:?}", rinfo.error); + assert_eq!(rinfo.target_version_id.as_deref(), Some(DRIFTED_ASSIGNED_VERSION_ID)); + 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:?}"); + } + + #[tokio::test] + async fn heal_redrive_does_not_adopt_an_encrypted_replica_of_another_size_or_an_older_write() { + // Size differs from the 4 plaintext bytes. + static OTHER_SIZE: &[(&str, i64)] = &[(DRIFTED_ASSIGNED_VERSION_ID, 36)]; + let (endpoint, server) = spawn_computed_etag_target(2, OTHER_SIZE); + let target = test_target_client(endpoint); + ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn); + let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() - 1); + let mut rinfo = replicate_all_target_info(&roi, &target); + let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await; + assert!(matches!(action, Some((ReplicationAction::All, _))), "{action:?}"); + server.join().expect("test HTTP server should finish"); + + // Same size, but written before the source version existed. + static SAME_SIZE: &[(&str, i64)] = &[(DRIFTED_ASSIGNED_VERSION_ID, 4)]; + let (endpoint, server) = spawn_computed_etag_target(2, SAME_SIZE); + let target = test_target_client(endpoint); + ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn); + let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() + 1); + let mut rinfo = replicate_all_target_info(&roi, &target); + let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await; + assert!(matches!(action, Some((ReplicationAction::All, _))), "{action:?}"); + server.join().expect("test HTTP server should finish"); + } + + #[tokio::test] + async fn heal_redrive_does_not_pick_among_several_encrypted_shape_matches() { + static VERSIONS: &[(&str, i64)] = &[(DRIFTED_ASSIGNED_VERSION_ID, 4), ("another-minted-id", 4)]; + let (endpoint, server) = spawn_computed_etag_target(2, VERSIONS); + let target = test_target_client(endpoint); + ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn); + let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() - 1); + let mut rinfo = replicate_all_target_info(&roi, &target); + + let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await; + + assert!( + matches!(action, Some((ReplicationAction::All, _))), + "two same-shape versions are not evidence of either: {action:?}" + ); + assert!(rinfo.target_version_id.is_none(), "{:?}", rinfo.target_version_id); + let seen = server.join().expect("test HTTP server should finish"); + assert_eq!(seen.len(), 2, "HEAD by source id, ListObjectVersions, no HEAD of a guess: {seen:?}"); + } + + #[tokio::test] + async fn heal_redrive_keeps_exact_etag_identity_for_plaintext_sources() { + // A same-size replica whose ETag differs is not this plaintext + // version's replica: the shape criterion is for encrypted sources only. + static VERSIONS: &[(&str, i64)] = &[(DRIFTED_ASSIGNED_VERSION_ID, 4)]; + let (endpoint, server) = spawn_computed_etag_target(2, VERSIONS); + let target = test_target_client(endpoint); + ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn); + let (roi, mut object_info) = drifted_roi_and_object(); + object_info.mod_time = Some(OffsetDateTime::from_unix_timestamp(replica_written_secs() - 1).expect("valid timestamp")); + let mut rinfo = replicate_all_target_info(&roi, &target); + + let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await; + + assert!(matches!(action, Some((ReplicationAction::All, _))), "{action:?}"); + server.join().expect("test HTTP server should finish"); + } + + #[tokio::test] + async fn heal_redrive_never_adopts_an_unversioned_null_version_by_shape() { + // An unversioned target's `null` version holds whatever generation + // was written last; a same-size newer generation must not be claimed. + static VERSIONS: &[(&str, i64)] = &[("null", 4)]; + let (endpoint, server) = spawn_computed_etag_target(2, VERSIONS); + let target = test_target_client(endpoint); + ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn); + let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() - 1); + let mut rinfo = replicate_all_target_info(&roi, &target); + + let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await; + + assert!(matches!(action, Some((ReplicationAction::All, _))), "{action:?}"); + assert!(rinfo.target_version_id.is_none(), "{:?}", rinfo.target_version_id); + let seen = server.join().expect("test HTTP server should finish"); + assert_eq!(seen.len(), 2, "HEAD by source id, ListObjectVersions, nothing adopted: {seen:?}"); + } + + #[tokio::test] + async fn encrypted_lookup_keeps_the_current_version_fallback_when_listing_is_denied() { + // A bodiless 403 on the version-addressed HEAD may be a permission + // denial; the same credentials then refuse ListObjectVersions. The + // object must still be re-driven, as before content identity existed. + let (endpoint, server) = spawn_scripted_target_server(3, |line| { + let listing = line.starts_with("GET ") && line.contains("versions"); + let version_addressed_head = line.starts_with("HEAD ") && line.contains("versionId="); + if listing || version_addressed_head { + empty_response("403 Forbidden") + } else if line.starts_with("HEAD ") { + computed_etag_head_response(DRIFTED_ASSIGNED_VERSION_ID, 4) + } else { + empty_response("500 Unexpected") + } + }); + let target = test_target_client(endpoint); + let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() - 1); + let mut rinfo = replicate_all_target_info(&roi, &target); + + let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await; + + assert!( + matches!(action, Some((ReplicationAction::All, _))), + "a denied listing must fall back to the re-PUT, not stall: {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 of the current version: {seen:?}" + ); + assert!(seen[2].starts_with("HEAD ") && !seen[2].contains("versionId="), "{}", seen[2]); + } + + #[tokio::test] + async fn encrypted_ledger_replica_converges_on_an_aws_style_target() { + // AWS rejects the RustFS uuid as malformed (400). The ledger names the + // replica; its computed ETag cannot confirm it, its shape does. + let (endpoint, server) = spawn_scripted_target_server(2, |line| { + if line.starts_with("HEAD ") && line.contains(&format!("versionId={DRIFTED_ASSIGNED_VERSION_ID}")) { + computed_etag_head_response(DRIFTED_ASSIGNED_VERSION_ID, 4) + } else if line.starts_with("HEAD ") { + empty_response("400 Bad Request") + } else { + empty_response("500 Unexpected") + } + }); + let target = test_target_client(endpoint); + let (roi, mut object_info) = encrypted_roi_and_object(replica_written_secs() - 1); + let mut user_defined = (*object_info.user_defined).clone(); + insert_str( + &mut user_defined, + &format!("{SUFFIX_REPLICATION_TARGET_VERSION_ARN_PREFIX}{}", target.arn), + DRIFTED_ASSIGNED_VERSION_ID.to_string(), + ); + object_info.user_defined = Arc::new(user_defined); + let mut rinfo = replicate_all_target_info(&roi, &target); + + let action = resolve_heal_action(&target, &roi, object_info, &mut rinfo).await; + + assert!(matches!(action, Some((ReplicationAction::None, _))), "{action:?}"); + let seen = server.join().expect("test HTTP server should finish"); + assert_eq!(seen.len(), 2, "HEAD by source id, HEAD by the ledger id: {seen:?}"); + assert!(seen[1].contains(DRIFTED_ASSIGNED_VERSION_ID), "{}", seen[1]); + } + + #[tokio::test] + async fn resync_verification_counts_encrypted_replica_by_shape() { + static VERSIONS: &[(&str, i64)] = &[(DRIFTED_ASSIGNED_VERSION_ID, 4)]; + let (endpoint, server) = spawn_computed_etag_target(3, VERSIONS); + let target = test_target_client(endpoint); + ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn); + let (roi, object_info) = encrypted_roi_and_object(replica_written_secs() - 1); + 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 (_, err) = verify_resync_head_result(head_result, &roi, Some(&object_info), &mut st, &target).await; + + assert!(err.is_none(), "{err:?}"); + assert_eq!((st.replicated_count, st.failed_count), (1, 0)); + server.join().expect("test HTTP server should finish"); + } + + fn encrypted_purge_source() -> Arc { + let (_, object_info) = encrypted_roi_and_object(replica_written_secs() - 1); + let Some(ReplicaIdentity::Encrypted { shape, .. }) = replica_identity(&object_info) else { + panic!("an SSE-S3 source must be identified by its shape"); + }; + Arc::new(VersionPurgeSource { + etag: object_info.etag, + encrypted_shape: Some(shape), + ledger: HashMap::new(), + ledger_corrupt: false, + claimed_by_siblings: HashMap::new(), + }) + } + + #[tokio::test] + async fn encrypted_version_purge_without_ledger_deletes_the_unique_shape_match() { + let (endpoint, server) = spawn_scripted_target_server(2, |line| { + if line.starts_with("GET ") && line.contains("versions") { + computed_etag_versions_response(&[(DRIFTED_ASSIGNED_VERSION_ID, 4)]) + } else if line.starts_with("DELETE ") && line.contains(&format!("versionId={DRIFTED_ASSIGNED_VERSION_ID}")) { + empty_response("204 No Content") + } else { + empty_response("500 Unexpected") + } + }); + let target = test_target_client(endpoint); + register_test_target(&target).await; + ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn); + let dobj = version_purge_dobj(&target.arn); + + let rinfo = replicate_delete_to_target(&dobj, target.clone(), Some(encrypted_purge_source())).await; + + assert_eq!(rinfo.version_purge_status, VersionPurgeStatusType::Complete, "{:?}", rinfo.error); + let seen = server.join().expect("test HTTP server should finish"); + assert_eq!(seen.len(), 2, "ListObjectVersions, then DELETE by the located id: {seen:?}"); + assert!( + seen[1].starts_with("DELETE ") && seen[1].contains(DRIFTED_ASSIGNED_VERSION_ID), + "{}", + seen[1] + ); + } + + #[tokio::test] + async fn encrypted_version_purge_refuses_several_shape_matches() { + let (endpoint, server) = spawn_scripted_target_server(1, |line| { + if line.starts_with("GET ") && line.contains("versions") { + computed_etag_versions_response(&[(DRIFTED_ASSIGNED_VERSION_ID, 4), ("another-minted-id", 4)]) + } else { + empty_response("500 Unexpected") + } + }); + let target = test_target_client(endpoint); + register_test_target(&target).await; + ReplicationTargetStore::record_version_identity_capability(&target.arn, VersionIdentityCapability::MintsOwn); + let dobj = version_purge_dobj(&target.arn); + + let rinfo = replicate_delete_to_target(&dobj, target.clone(), Some(encrypted_purge_source())).await; + + assert_eq!(rinfo.version_purge_status, VersionPurgeStatusType::Failed); + let seen = server.join().expect("test HTTP server should finish"); + assert_eq!(seen.len(), 1, "no DELETE of a guessed replica: {seen:?}"); + } + fn delete_marker_roi() -> ReplicateObjectInfo { ReplicateObjectInfo { bucket: "source".to_string(), @@ -6886,7 +7367,7 @@ mod tests { None, "the fixture must reproduce the codeless 405 the SDK yields for a bodiless HEAD" ); - let (size, err) = verify_resync_head_result(head_result, &roi, &mut st, &target).await; + let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await; assert!(err.is_none(), "{err:?}"); assert_eq!((size, st.replicated_count, st.failed_count), (0, 1, 0)); @@ -6913,7 +7394,7 @@ mod tests { 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; + let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await; assert!(err.is_some(), "a 405 must not count as a completed purge ({status:?})"); assert_eq!((size, st.replicated_count, st.failed_count), (0, 0, 1), "{status:?}"); @@ -6931,7 +7412,7 @@ mod tests { 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; + let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await; assert!(err.is_none(), "{err:?}"); assert_eq!((size, st.replicated_count, st.failed_count), (0, 1, 0)); @@ -6948,7 +7429,7 @@ mod tests { 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; + let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await; assert!(err.is_some(), "a 503 must not count as a propagated delete marker"); assert_eq!((size, st.replicated_count, st.failed_count), (0, 0, 1)); @@ -6974,7 +7455,7 @@ mod tests { 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; + let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await; assert!(err.is_none(), "{err:?}"); assert_eq!((size, st.replicated_count, st.failed_count), (0, 1, 0)); @@ -6999,7 +7480,7 @@ mod tests { 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; + let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await; assert!(err.is_some(), "a live latest version must not verify the marker"); assert_eq!((size, st.replicated_count, st.failed_count), (0, 0, 1)); @@ -7017,7 +7498,7 @@ mod tests { 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; + let (size, err) = verify_resync_head_result(head_result, &roi, None, &mut st, &target).await; assert!(err.is_some(), "a 403 must not count as a propagated delete marker"); assert_eq!((size, st.replicated_count, st.failed_count), (0, 0, 1)); @@ -7134,6 +7615,7 @@ mod tests { fn purge_source(ledger: &[(&str, &str)]) -> Arc { Arc::new(VersionPurgeSource { etag: Some(DRIFTED_ETAG.to_string()), + encrypted_shape: None, ledger: ledger .iter() .map(|(arn, version_id)| (arn.to_string(), version_id.to_string())) @@ -7338,6 +7820,7 @@ mod tests { let dobj = version_purge_dobj(&target.arn); let source = Arc::new(VersionPurgeSource { etag: Some(DRIFTED_ETAG.to_string()), + encrypted_shape: None, ledger: HashMap::new(), ledger_corrupt: false, claimed_by_siblings: HashMap::from([(target.arn.clone(), HashSet::from([DRIFTED_ASSIGNED_VERSION_ID.to_string()]))]), @@ -7425,6 +7908,7 @@ mod tests { let dobj = version_purge_dobj(&target.arn); let source = Arc::new(VersionPurgeSource { etag: Some(DRIFTED_ETAG.to_string()), + encrypted_shape: None, ledger: HashMap::new(), ledger_corrupt: true, claimed_by_siblings: HashMap::new(), diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index e425bad99..2d7fba2dd 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -37,8 +37,9 @@ use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; pub(crate) use crate::bucket::bucket_target_sys::{ - AdvancedPutOptions, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions, RemotePutObjectResponse, RemoveObjectOptions, - ReplicaLocation, S3ClientError, TargetClient, resolve_read_api_version_id, + AdvancedPutOptions, EncryptedReplicaShape, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions, + RemotePutObjectResponse, RemoveObjectOptions, ReplicaCandidates, ReplicaEvidence, ReplicaIdentity, ReplicaLocation, + S3ClientError, TargetClient, resolve_read_api_version_id, }; #[cfg(test)] pub(crate) use crate::bucket::target::BucketTarget; diff --git a/docs/operations/encrypted-etags.md b/docs/operations/encrypted-etags.md index 71d807a2c..854ee3740 100644 --- a/docs/operations/encrypted-etags.md +++ b/docs/operations/encrypted-etags.md @@ -36,6 +36,32 @@ the replication PUT either, so a target that transforms their payload is not detected. See `docs/operations/replication-outbound-transport.md`. +The same mismatch affects finding an existing replica on a target that mints +its own version ids (AWS S3, Wasabi), which a heal, MRF retry or +existing-object resync does before writing again: + +- A replica recorded in the source's target-version ledger is addressed by + that id, on a 404 and on an AWS-style 400 for the source id alike. For an + encrypted source the replica counts when its size matches the bytes + replication sent (plaintext for SSE-S3 / SSE-KMS, ciphertext for SSE-C) and + it was written no earlier than the source version. +- Without a ledger entry (for example a PUT whose response was lost), the + replica is located through `ListObjectVersions` by the same size and + write-time rule, after removing versions that other generations of the key + already claim. Only a single remaining match is adopted and recorded in the + ledger. Several matches are never resolved by picking one: convergence writes + the replica again, and a version purge is refused with the usual backoff. + The `null` version of an unversioned target is never matched this way, and + a target that refuses the listing keeps the earlier current-version check, + which writes the replica again. +- Plaintext sources keep exact ETag identity. A target that adopts the source + ETag still matches encrypted sources by ETag first. + +Two same-size generations of a key whose own ledgers were both lost can still +be confused, and a version purge with no ledger and no shape match reports the +replica as already gone. Delete a leftover replica on the target by hand if a +`replication-check` VersionFidelity report shows one. + ## Existing data and migration The upgrade does not rewrite existing object metadata or erase fingerprints diff --git a/docs/operations/replication-outbound-transport.md b/docs/operations/replication-outbound-transport.md index 47f0a9de6..b99b4bd24 100644 --- a/docs/operations/replication-outbound-transport.md +++ b/docs/operations/replication-outbound-transport.md @@ -18,7 +18,7 @@ | Rejects or mis-stores `aws-chunked` bodies (SeaweedFS 3.97) | Handled by the plain-payload default above. | Outbound target matrix, `RejectAwsChunked` mode | | Requires `Content-MD5` or `x-amz-checksum-*` on a PutObject with Object Lock parameters (AWS S3, MinIO, Impossible Cloud, most compatible stores) | Satisfied: a locked single PUT carries `Content-MD5` derived from the source ETag (plaintext objects whose ETag is the MD5 of the wire bytes) or an SDK CRC32 checksum (multipart-layout ETags, managed SSE, SSE-C passthrough — this one is an `aws-chunked` trailer, so a target that also rejects that framing cannot take such objects). Releases before this fix (`1.0.0-rc.5`) need `RUSTFS_REPLICATION_STREAMING_CHECKSUMS=true` as a workaround. | Outbound target matrix, `RequireChecksumWithObjectLock` mode | | Stores `x-amz-checksum-*` from a PutObject and returns it on `HEAD ?ChecksumMode=ENABLED` (AWS S3, Wasabi, RustFS) | Satisfied for single-part objects: the replica answers with the source's checksum. Before this fix (`1.0.0-rc.5`) the checksum left the source as `x-amz-meta-` user metadata and no replica carried it (rustfs/backlog#2340). | Outbound target matrix, `Checksummed` shape | -| Mints its own version ids (AWS S3, Wasabi, Impossible Cloud) | Data lands, and later mutations converge through the target-version ledger: the id the target assigned is recorded on the source version (internal key `replication-target-version-`) and version deletes, tag and Object Lock updates address it (tag/retention/legal-hold through the metadata-only APIs, never a re-PUT). A replica written before the ledger existed is located by exact key + ETag, minus the candidates other generations of the key already claim; an ambiguous remainder is refused with a one-hour backoff (`replication_purge_replica_unresolved`). `NoSuchVersion` on a version-addressed DELETE counts as purged. See rustfs/backlog#2340, rustfs/backlog#2085 and `docs/operations/replication-check.md` (VersionFidelity). | `replication-check`, outbound target matrix, `MintOwnVersionIds` mode (also models Wasabi's 404 `NoSuchVersion` on unknown ids) | +| Mints its own version ids (AWS S3, Wasabi, Impossible Cloud) | Data lands, and later mutations converge through the target-version ledger: the id the target assigned is recorded on the source version (internal key `replication-target-version-`) and version deletes, tag and Object Lock updates address it (tag/retention/legal-hold through the metadata-only APIs, never a re-PUT). A replica the ledger does not name (written before the ledger existed, or a PUT whose response was lost) is located by exact key + ETag, minus the candidates other generations of the key already claim; an ambiguous remainder is refused with a one-hour backoff (`replication_purge_replica_unresolved`). An encrypted source whose ETag the target does not report is matched by size and write time instead, and only a unique match is adopted (`docs/operations/encrypted-etags.md`). `NoSuchVersion` on a version-addressed DELETE counts as purged. See rustfs/backlog#2340, rustfs/backlog#2085 and `docs/operations/replication-check.md` (VersionFidelity). | `replication-check`, outbound target matrix, `MintOwnVersionIds` mode (also models Wasabi's 404 `NoSuchVersion` on unknown ids); `matrix_mint_own_version_ids_encrypted_redrive_does_not_duplicate` for SSE-S3 on an ETag-computing target | | Ignores `x-rustfs-source-etag` / `x-minio-source-etag` and reports the MD5 of the received plaintext as the ETag (AWS S3, generic S3 services) | Plaintext single-part replicas are verified against the source ETag. Encrypted sources skip the comparison (see above). In `1.0.2`, the first release with encrypted-ETag fingerprints, every single-part SSE-S3 object (and SSE-KMS where the target does not echo `aws:kms`) failed with `replica etag mismatch` and was retried indefinitely; the only workaround there is `RUSTFS_REPLICATION_REPLICA_ETAG_VERIFY=false`, which disables the check for every object. RustFS and MinIO targets adopt the source ETag and were not affected. | Outbound target matrix, `ComputesOwnEtags` mode, `SseS3` shape | | Returns an ETag that is not the content MD5 without announcing SSE | Every single-part object fails ETag verification. Set `RUSTFS_REPLICATION_REPLICA_ETAG_VERIFY=false`. | Replication status FAILED with `replica etag mismatch` |