fix(site-replication): stamp replicated bucket configs with the source updated_at (backlog#2292)

The bucket-meta receiver judged an incoming item stale by comparing its
source `updated_at` with the `*_config_updated_at` stamp of the config on
disk, but that stamp was the receiver's local clock at apply time
(`BucketMetadata::update_config`). A source edit newer than the applied one
but delivered after the local stamp was judged stale and acknowledged with
200: two quick source edits under delivery delay lose the second, and a peer
clock ahead of ours loses every follow-up edit inside the skew.

Add explicit-timestamp write entries, expanding rather than changing the
existing ones:

- `BucketMetadata::update_config_at`; `update_config` delegates to it with
  the local clock.
- `metadata_sys::update_if_incarnation_at`,
  `update_under_transaction_lock_at`, `update_quota_if_incarnation_at`,
  threaded through the shared write-guard path as `Option<OffsetDateTime>`
  (`None` keeps local stamping for every existing caller and for deletes).
- Re-exported through the ecstore `api` facade and the rustfs admin
  `storage_api::metadata_sys` facade.

`apply_bucket_meta_item` now persists policy, tags, versioning, object-lock,
sse, replication, quota and cors configs with the item's source time, so the
stored stamp equals the source `updatedAt` and staleness is judged source
time against source time. Items without `updated_at` keep the local stamp.
lc-config stays on the local stamp: its staleness axis is the in-document
`expiry_updated_at` the merge records, and the whole-config time only serves
as its deletion / legacy lower bound. Local (non-replicated) edits keep
stamping the local clock — they are the source.

(cherry picked from commit c1009c018b217ef9edc7773c8e56667ea7e77335)

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<u8>,
+    expected_incarnation_id: Uuid,
+    source_updated_at: Option<OffsetDateTime>,
+) -> 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"
+        );
+    }
 }

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<u8>,
+        expected_incarnation_id: uuid::Uuid,
+        updated_at: OffsetDateTime,
+    ) -> Result<OffsetDateTime> {
+        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<u8>,
@@ -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<u8>,
+        expected_incarnation_id: uuid::Uuid,
+        proof: &super::ecstore_notification::CrossPoolFenceFleetProofToken,
+        updated_at: OffsetDateTime,
+    ) -> Result<OffsetDateTime> {
+        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<uuid::Uuid> {
         super::ecstore_bucket::metadata_sys::capture_bucket_metadata_incarnation(bucket).await
     }