diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 437f5c458..aed429ed1 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -204,8 +204,9 @@ pub mod bucket { get_public_access_block_config, get_quota_config, get_replication_config, get_request_payment_config, get_sse_config, get_tagging_config, get_versioning_config, get_website_config, init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata, remove_bucket_metadata, set_bucket_metadata, update, - update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation, update_quota_if_incarnation, - update_under_transaction_lock, + update_bucket_targets_under_transaction_lock, update_config_with, update_if_incarnation, update_if_incarnation_at, + update_quota_if_incarnation, update_quota_if_incarnation_at, update_under_transaction_lock, + update_under_transaction_lock_at, }; } diff --git a/crates/ecstore/src/bucket/metadata.rs b/crates/ecstore/src/bucket/metadata.rs index 41b2afdc5..ddb2bf790 100644 --- a/crates/ecstore/src/bucket/metadata.rs +++ b/crates/ecstore/src/bucket/metadata.rs @@ -802,9 +802,22 @@ impl BucketMetadata { } } + /// Replace one config payload and stamp its `*_config_updated_at` with the + /// local clock. This is the entry for edits that originate here: the + /// local write time is the edit's source time. pub fn update_config(&mut self, config_file: &str, data: Vec) -> Result { - let updated = OffsetDateTime::now_utc(); + self.update_config_at(config_file, data, OffsetDateTime::now_utc()) + } + /// [`Self::update_config`] with an explicit `updated_at` stamp. + /// + /// For a config replicated from another site the edit's source time is + /// the peer's `updated_at`, not the moment it lands here: staleness of + /// the next incoming item is judged against the stored stamp, so stamping + /// the local apply time would reject a newer source edit that was merely + /// delivered late (backlog#2292). Only replication receivers should pass + /// a foreign time; local edits keep [`Self::update_config`]. + pub fn update_config_at(&mut self, config_file: &str, data: Vec, updated: OffsetDateTime) -> Result { match config_file { BUCKET_POLICY_CONFIG => { self.policy_config_json = data; @@ -1543,6 +1556,39 @@ mod test { assert_eq!(metadata.bucket_incarnation_id, incarnation); } + /// backlog#2292: a replicated config is stamped with the source + /// `updated_at` it was given, not the local clock, while the plain + /// `update_config` entry keeps stamping the local clock. + #[test] + fn update_config_at_stamps_the_given_time_and_update_config_stamps_now() { + let source_time = OffsetDateTime::now_utc() - time::Duration::hours(3); + let mut metadata = BucketMetadata::new("source-stamped"); + + let stamped = metadata + .update_config_at(BUCKET_POLICY_CONFIG, br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec(), source_time) + .unwrap(); + assert_eq!(stamped, source_time); + assert_eq!(metadata.policy_config_updated_at, source_time); + + let tagging = b"kv".to_vec(); + let stamped = metadata + .update_config_at(BUCKET_TAGGING_CONFIG, tagging, source_time) + .unwrap(); + assert_eq!(stamped, source_time); + assert_eq!(metadata.tagging_config_updated_at, source_time); + + let before = OffsetDateTime::now_utc(); + let stamped = metadata + .update_config(BUCKET_POLICY_CONFIG, br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec()) + .unwrap(); + assert!(stamped >= before, "a local edit is stamped with the local clock"); + assert_eq!(metadata.policy_config_updated_at, stamped); + assert_eq!( + metadata.tagging_config_updated_at, source_time, + "restamping one config must not move another config's stamp" + ); + } + #[test] fn object_locking_requires_lock_metadata_not_plain_versioning() { use s3s::dto::ObjectLockEnabled; diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 4a53927f9..b04dd9989 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -581,6 +581,32 @@ pub async fn update_if_incarnation( config_file, data, Some(expected_incarnation_id), + None, + )) + .await +} + +/// [`update_if_incarnation`] stamping the config with `updated_at` instead of +/// the local clock. +/// +/// For a site-replication receiver the edit's source time is the peer's +/// `updated_at`; persisting it keeps the stored `*_config_updated_at` on the +/// source clock so the next item's staleness is judged source-time against +/// source-time (backlog#2292). See [`BucketMetadata::update_config_at`]. +pub async fn update_if_incarnation_at( + bucket: &str, + config_file: &str, + data: Vec, + expected_incarnation_id: Uuid, + updated_at: OffsetDateTime, +) -> Result { + Box::pin(update_with_sys_expected( + get_bucket_metadata_sys()?, + bucket, + config_file, + data, + Some(expected_incarnation_id), + Some(updated_at), )) .await } @@ -612,18 +638,22 @@ async fn update_with_sys( config_file: &str, data: Vec, ) -> Result { - update_with_sys_expected(sys, bucket, config_file, data, None).await + update_with_sys_expected(sys, bucket, config_file, data, None, None).await } +/// `updated_at` is the stamp persisted on the config; `None` uses the local +/// clock (the edit originates here), `Some` carries a replicated edit's +/// source time (backlog#2292). async fn update_with_sys_expected( sys: Arc>, bucket: &str, config_file: &str, data: Vec, expected_incarnation_id: Option, + updated_at: Option, ) -> Result { let guard = acquire_config_write_guard_for_incarnation(sys.clone(), bucket, expected_incarnation_id).await?; - update_under_config_write_guard(sys, &guard, config_file, data).await + update_under_config_write_guard(sys, &guard, config_file, data, updated_at).await } /// [`delete`] against an explicitly supplied metadata system. See @@ -786,7 +816,21 @@ pub async fn update_under_transaction_lock( data: Vec, ) -> Result { guard.ensure_valid(bucket)?; - update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data).await + update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data, None).await +} + +/// [`update_under_transaction_lock`] stamping the config with `updated_at` +/// (a replicated edit's source time) instead of the local clock; see +/// [`update_if_incarnation_at`] (backlog#2292). +pub async fn update_under_transaction_lock_at( + guard: &BucketMetadataMutationGuard, + bucket: &str, + config_file: &str, + data: Vec, + updated_at: OffsetDateTime, +) -> Result { + guard.ensure_valid(bucket)?; + update_under_config_write_guard(get_bucket_metadata_sys()?, guard, config_file, data, Some(updated_at)).await } /// Clear one config file while the caller holds this bucket's transaction lock. @@ -804,6 +848,29 @@ pub async fn update_quota_if_incarnation( data: Vec, expected_incarnation_id: Uuid, proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken, +) -> Result { + update_quota_if_incarnation_stamped(bucket, data, expected_incarnation_id, proof, None).await +} + +/// [`update_quota_if_incarnation`] stamping the quota config with +/// `updated_at` (a replicated edit's source time) instead of the local +/// clock; see [`update_if_incarnation_at`] (backlog#2292). +pub async fn update_quota_if_incarnation_at( + bucket: &str, + data: Vec, + expected_incarnation_id: Uuid, + proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken, + updated_at: OffsetDateTime, +) -> Result { + update_quota_if_incarnation_stamped(bucket, data, expected_incarnation_id, proof, Some(updated_at)).await +} + +async fn update_quota_if_incarnation_stamped( + bucket: &str, + data: Vec, + expected_incarnation_id: Uuid, + proof: &crate::services::notification_sys::CrossPoolFenceFleetProofToken, + updated_at: Option, ) -> Result { let sys = get_bucket_metadata_sys()?; let guard = Box::pin(acquire_config_write_guard_for_incarnation( @@ -821,7 +888,7 @@ pub async fn update_quota_if_incarnation( achieved: 0, }); } - update_under_config_write_guard(sys, &guard, rustfs_config::QUOTA_CONFIG_FILE, data).await + update_under_config_write_guard(sys, &guard, rustfs_config::QUOTA_CONFIG_FILE, data, updated_at).await } pub async fn update_bucket_targets_under_transaction_lock( @@ -837,6 +904,7 @@ async fn update_under_config_write_guard( guard: &BucketMetadataMutationGuard, config_file: &str, data: Vec, + updated_at: Option, ) -> Result { guard.ensure_valid(&guard.bucket)?; let metadata_sys = sys.read().await.clone(); @@ -848,7 +916,7 @@ async fn update_under_config_write_guard( Some(&guard.transaction_guard), &guard.bucket, "bucket config transaction", - metadata_sys.update_checked(&guard.bucket, config_file, data, true, guard.incarnation_id), + metadata_sys.update_checked(&guard.bucket, config_file, data, true, guard.incarnation_id, updated_at), ), ) .await?; @@ -871,7 +939,7 @@ async fn delete_under_config_write_guard( Some(&guard.transaction_guard), &guard.bucket, "bucket config deletion transaction", - metadata_sys.update_checked(&guard.bucket, config_file, Vec::new(), false, guard.incarnation_id), + metadata_sys.update_checked(&guard.bucket, config_file, Vec::new(), false, guard.incarnation_id, None), ), ) .await?; @@ -1770,15 +1838,17 @@ impl BucketMetadataSys { /// `update` and the config read alone). Keep these boxed. pub async fn update(&self, bucket: &str, config_file: &str, data: Vec) -> Result { let incarnation_id = Box::pin(self.get_bucket_incarnation_id(bucket)).await?; - Box::pin(self.update_checked(bucket, config_file, data, true, incarnation_id)).await + Box::pin(self.update_checked(bucket, config_file, data, true, incarnation_id, None)).await } pub async fn delete(&self, bucket: &str, config_file: &str) -> Result { let incarnation_id = self.get_bucket_incarnation_id(bucket).await?; - self.update_checked(bucket, config_file, Vec::new(), false, incarnation_id) + self.update_checked(bucket, config_file, Vec::new(), false, incarnation_id, None) .await } + /// `updated_at`: `None` stamps the local clock; `Some` persists a + /// replicated edit's source time (backlog#2292). async fn update_checked( &self, bucket: &str, @@ -1786,6 +1856,7 @@ impl BucketMetadataSys { data: Vec, parse: bool, expected_incarnation_id: Uuid, + updated_at: Option, ) -> Result { // Load through this system's own store, the one `save` persists to // (backlog#1052 S7). Reading from the ambient handle instead made the @@ -1796,7 +1867,10 @@ impl BucketMetadataSys { return Err(Error::BucketNotFound(bucket.to_string())); } - let updated = bm.update_config(config_file, data)?; + let updated = match updated_at { + Some(updated_at) => bm.update_config_at(config_file, data, updated_at)?, + None => bm.update_config(config_file, data)?, + }; Box::pin(self.save(bm)).await?; @@ -3765,6 +3839,57 @@ mod tests { ); } + /// backlog#2292: the explicit-stamp write path persists the given source + /// time as the config's `*_config_updated_at` — through the incarnation + /// path and through an already-held transaction guard — and survives a + /// reload from disk, while the plain path keeps stamping the local clock. + #[tokio::test] + async fn explicit_updated_at_is_persisted_as_the_config_stamp() { + let (dirs, ecstore) = isolated_store_over_temp_disks().await; + let bucket = "source-stamped-config"; + for dir in &dirs { + std::fs::create_dir_all(dir.path().join(bucket)).expect("bucket volume should be created"); + } + let sys = Arc::new(RwLock::new(BucketMetadataSys::new(ecstore))); + let source_time = OffsetDateTime::now_utc() - Duration::from_secs(3 * 3600); + let policy = br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec(); + let tagging = b"kv".to_vec(); + + // Incarnation path (`update_if_incarnation_at` minus the ambient lookup). + let stamped = + update_with_sys_expected(sys.clone(), bucket, BUCKET_POLICY_CONFIG, policy.clone(), None, Some(source_time)) + .await + .expect("source-stamped policy write should persist"); + assert_eq!(stamped, source_time); + + // Held-guard path (`update_under_transaction_lock_at` minus the ambient lookup). + let guard = acquire_config_write_guard(sys.clone(), bucket).await.expect("write guard"); + let stamped = update_under_config_write_guard(sys.clone(), &guard, BUCKET_TAGGING_CONFIG, tagging, Some(source_time)) + .await + .expect("source-stamped tagging write should persist"); + drop(guard); + assert_eq!(stamped, source_time); + + let metadata_sys = sys.read().await.clone(); + metadata_sys.metadata_map.write().await.clear(); + let reloaded = metadata_sys.get_config_from_disk(bucket).await.expect("reload from disk"); + assert_eq!(reloaded.policy_config_updated_at, source_time); + assert_eq!(reloaded.tagging_config_updated_at, source_time); + + // The plain path is unchanged: a local edit is stamped with the local clock. + let before = OffsetDateTime::now_utc(); + let stamped = update_with_sys(sys.clone(), bucket, BUCKET_POLICY_CONFIG, policy) + .await + .expect("locally stamped policy write should persist"); + assert!(stamped >= before, "the plain write path must keep stamping the local clock"); + let reloaded = metadata_sys.get_config_from_disk(bucket).await.expect("reload from disk"); + assert_eq!(reloaded.policy_config_updated_at, stamped); + assert_eq!( + reloaded.tagging_config_updated_at, source_time, + "an unrelated config keeps its source stamp" + ); + } + /// The load and the persisted write share one write guard, so concurrent /// rewrites of the same config compose instead of clobbering each other. /// Moving the load outside that guard loses all but the last tag. @@ -3981,10 +4106,16 @@ mod tests { let new_incarnation = store.bucket_incarnation_id_from_disk(bucket).await.unwrap(); assert_ne!(old_incarnation, new_incarnation); - let err = - update_with_sys_expected(sys.clone(), bucket, BUCKET_TAGGING_CONFIG, b"".to_vec(), Some(old_incarnation)) - .await - .expect_err("a request authorized for the deleted incarnation must fail closed"); + let err = update_with_sys_expected( + sys.clone(), + bucket, + BUCKET_TAGGING_CONFIG, + b"".to_vec(), + Some(old_incarnation), + None, + ) + .await + .expect_err("a request authorized for the deleted incarnation must fail closed"); assert!(matches!(err, Error::BucketNotFound(name) if name == bucket)); let persisted = sys.read().await.get_config_from_disk(bucket).await.unwrap(); @@ -4019,7 +4150,7 @@ mod tests { }], }) .unwrap(); - update_under_config_write_guard(sys, &guard, BUCKET_TAGGING_CONFIG, tagging) + update_under_config_write_guard(sys, &guard, BUCKET_TAGGING_CONFIG, tagging, None) .await .unwrap(); assert!(!delete.is_finished()); diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index de8fe6565..2f1d40fcb 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -5619,6 +5619,17 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> { _ => unreachable!(), }; + // Persist the SOURCE `updated_at` as the stored `*_config_updated_at` + // stamp (backlog#2292). The staleness gate above compares the next item's + // source time against that stamp, so stamping the local apply time would + // reject a newer source edit that was merely delivered after this write + // (two quick edits under delivery delay, or a peer clock ahead of ours). + // Items without a source time keep the local stamp; lc-config keeps it + // too: its staleness axis is the in-document `expiry_updated_at` the merge + // above records, and the whole-config time is only its deletion / legacy + // lower bound. + let source_updated_at = if item.r#type == "lc-config" { None } else { item.updated_at }; + if !skip_config_write { if let Some(data) = data { if item.r#type == "quota-config" { @@ -5637,13 +5648,25 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> { "durable quota capability is not confirmed across the cluster".to_string(), ) })?; - metadata_sys::update_quota_if_incarnation(&item.bucket, data, expected_incarnation_id, &proof) - .await - .map_err(ApiError::from)?; + match source_updated_at { + Some(source_updated_at) => { + metadata_sys::update_quota_if_incarnation_at( + &item.bucket, + data, + expected_incarnation_id, + &proof, + source_updated_at, + ) + .await + } + None => { + metadata_sys::update_quota_if_incarnation(&item.bucket, data, expected_incarnation_id, &proof).await + } + } + .map_err(ApiError::from)?; } else { - metadata_sys::update_if_incarnation(&item.bucket, config_file, data, expected_incarnation_id) - .await - .map_err(ApiError::from)?; + write_replicated_bucket_config(&item.bucket, config_file, data, expected_incarnation_id, source_updated_at) + .await?; } } else { if let Some(guard) = lifecycle_guard.as_ref() { @@ -5651,9 +5674,8 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> { .await .map_err(ApiError::from)?; } else { - metadata_sys::update_if_incarnation(&item.bucket, config_file, data, expected_incarnation_id) - .await - .map_err(ApiError::from)?; + write_replicated_bucket_config(&item.bucket, config_file, data, expected_incarnation_id, source_updated_at) + .await?; } } } else { @@ -5689,6 +5711,26 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> { Ok(()) } +/// Write one replicated bucket config, stamped with the item's source +/// `updated_at` when it carries one and with the local clock otherwise +/// (backlog#2292; see [`apply_bucket_meta_item`]). +async fn write_replicated_bucket_config( + bucket: &str, + config_file: &str, + data: Vec, + expected_incarnation_id: Uuid, + source_updated_at: Option, +) -> S3Result<()> { + match source_updated_at { + Some(source_updated_at) => { + metadata_sys::update_if_incarnation_at(bucket, config_file, data, expected_incarnation_id, source_updated_at).await + } + None => metadata_sys::update_if_incarnation(bucket, config_file, data, expected_incarnation_id).await, + } + .map_err(ApiError::from)?; + Ok(()) +} + fn group_info_requires_upsert(update: &rustfs_madmin::GroupAddRemove) -> bool { !update.is_remove } @@ -14150,4 +14192,72 @@ mod tests { ); } } + + /// backlog#2292: the receiver persists the SOURCE `updated_at` of an + /// applied bucket config and judges the next item's source time against + /// it. Stamping the local apply time instead rejected a source edit that + /// was newer than the applied one but delivered after the local stamp + /// (two quick source edits under delivery delay; a peer clock ahead of + /// ours) and acknowledged it with 200. + #[test] + fn test_bucket_meta_staleness_is_judged_against_the_applied_source_timestamp() { + let apply_wall_clock = OffsetDateTime::now_utc(); + let source_edit_t1 = apply_wall_clock - time::Duration::seconds(30); + let source_edit_t2 = source_edit_t1 + time::Duration::seconds(2); + let source_edit_t0 = source_edit_t1 - time::Duration::seconds(2); + assert!( + source_edit_t2 < apply_wall_clock, + "T2 is newer at the source yet older than the local apply clock" + ); + + // Edit T1 arrives first and is applied the way apply_bucket_meta_item + // persists a replicated config: stamped with its source time. + let mut meta = crate::admin::storage_api::bucket::metadata::BucketMetadata::new("photos"); + meta.update_config_at( + BUCKET_POLICY_CONFIG, + br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec(), + source_edit_t1, + ) + .expect("apply edit T1"); + let local_updated_at = bucket_meta_local_updated_at(&meta, BUCKET_POLICY_CONFIG); + assert_eq!( + local_updated_at, source_edit_t1, + "the stored stamp is the source time, not the apply clock" + ); + + // Edit T2 is newer at the source but delivered late: it must apply. + assert!( + !is_stale_update(local_updated_at, Some(source_edit_t2)), + "edit T2 ({source_edit_t2}) is newer than applied edit T1 ({source_edit_t1}) but is rejected against local stamp {local_updated_at}" + ); + // Edit T0 predates the applied edit: it stays rejected. + assert!( + is_stale_update(local_updated_at, Some(source_edit_t0)), + "edit T0 ({source_edit_t0}) is older than applied edit T1 ({source_edit_t1}) and must be rejected" + ); + // An item without a source time is never judged stale (unchanged). + assert!(!is_stale_update(local_updated_at, None)); + } + + /// backlog#2292: the replicated-config write in `apply_bucket_meta_item` + /// must go through the source-stamped entries; a plain + /// `update_if_incarnation` there would reintroduce local stamping. + #[test] + fn test_apply_bucket_meta_item_writes_through_the_source_stamped_entries() { + let source = include_str!("site_replication.rs"); + let apply = source + .split("async fn apply_bucket_meta_item") + .nth(1) + .and_then(|rest| rest.split("fn group_info_requires_upsert").next()) + .expect("apply_bucket_meta_item source"); + assert!( + apply.contains("update_quota_if_incarnation_at("), + "durable quota must carry the source stamp" + ); + assert!(apply.contains("update_if_incarnation_at("), "bucket configs must carry the source stamp"); + assert!( + !apply.contains("metadata_sys::update_if_incarnation(&item.bucket"), + "no replicated config write may bypass the source stamp" + ); + } } diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 03adac373..6674249e0 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -353,6 +353,25 @@ pub(crate) mod metadata_sys { super::ecstore_bucket::metadata_sys::update_if_incarnation(bucket, config_file, data, expected_incarnation_id).await } + /// [`update_if_incarnation`] stamping the config with a replicated edit's + /// source `updated_at` instead of the local clock (backlog#2292). + pub(crate) async fn update_if_incarnation_at( + bucket: &str, + config_file: &str, + data: Vec, + expected_incarnation_id: uuid::Uuid, + updated_at: OffsetDateTime, + ) -> Result { + super::ecstore_bucket::metadata_sys::update_if_incarnation_at( + bucket, + config_file, + data, + expected_incarnation_id, + updated_at, + ) + .await + } + pub(crate) async fn update_quota_if_incarnation( bucket: &str, data: Vec, @@ -362,6 +381,25 @@ pub(crate) mod metadata_sys { super::ecstore_bucket::metadata_sys::update_quota_if_incarnation(bucket, data, expected_incarnation_id, proof).await } + /// [`update_quota_if_incarnation`] stamping the quota with a replicated + /// edit's source `updated_at` instead of the local clock (backlog#2292). + pub(crate) async fn update_quota_if_incarnation_at( + bucket: &str, + data: Vec, + expected_incarnation_id: uuid::Uuid, + proof: &super::ecstore_notification::CrossPoolFenceFleetProofToken, + updated_at: OffsetDateTime, + ) -> Result { + super::ecstore_bucket::metadata_sys::update_quota_if_incarnation_at( + bucket, + data, + expected_incarnation_id, + proof, + updated_at, + ) + .await + } + pub(crate) async fn capture_bucket_metadata_incarnation(bucket: &str) -> Result { super::ecstore_bucket::metadata_sys::capture_bucket_metadata_incarnation(bucket).await }