fix(site-replication): keep reverse direction after config broadcast (#5292)

* fix(site-replication): keep reverse direction after config broadcast

`site-repl-*` rules encode the sender's outbound direction: their
destination ARN names the receiver. `apply_bucket_meta_item` wrote an
incoming rule set verbatim over the receiver's, leaving the receiver with
a rule whose ARN is its own deployment ID. `reconcile_site_replication_bucket_targets`
skips the local peer, so no bucket target can back that ARN and every
object was dropped; the follow-up call reconciled targets only, so
nothing rebuilt the lost reverse rule. Replication went one-directional
after any PutBucketReplication broadcast — the console's Save button,
`mc replicate import`, a metadata import, or `/site-replication/repair`.

Only operator-authored rules now travel between sites; each site owns its
`site-repl-*` rules and rebuilds them from the current peer set.

Four defects kept that invisible or unrecoverable:

- `update_all_targets` discarded target-client build errors silently, and
  `replicate_object` logged the resulting missing-target drop at debug
  while every other failure there logs at error. Both now report.
- `site_replication_rule_complete` never checked that a rule's
  destination named a remote site, so two sites holding identical
  configs — the post-clobber state — passed as in sync.
- `update_service_account` cannot rewrite `parent_user`, and IAM records
  encrypted with a previous root secret decode as "no such account".
  Startup now reconciles the account, reseeding from the secret every
  site-replication bucket target already stores, and refuses the
  delete-then-create sequence when the parent cannot back an account.
  Bucket rules are reconciled too, so an already-broken site heals on
  upgrade.
- A joined site never verified it could reach the initiator, whose
  endpoint is derived from the Host header of the admin request that
  created the topology. The join now probes each peer and reports through
  `initial_sync_error_message`.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(site-replication): report unreachable targets and reconcile on a timer

Rule-shape checking cannot see an unreachable peer. A `site-repl-*` rule
can be perfectly formed while the endpoint recorded for its peer is one
this site cannot reach: `update_all_targets` then builds no client and
`replicate_object` drops every object against that ARN, yet the rule set
still reads as correct and the bucket reports in sync.

Each site now reports whether all of its `site-repl-*` rules resolve to a
live target (`SRBucketInfo.replicationTargetsOnline`, read from the
already-resolved client map so the status path stays cheap), and the
status aggregation treats an offline report as a mismatch. The field is
additive and optional: peers that omit it are "unknown", never a fault,
so a mixed-version topology does not flip every bucket to out of sync.

The reconcilers also run on a 10-minute timer instead of at startup only,
so drift is repaired without waiting for a restart. Both are no-ops when
the wiring already matches — the bucket pass compares serialized targets
and the rule set before writing. The tick takes the site-replication
lifecycle lock with a non-blocking try_acquire and skips the round when
an add/remove/endpoint-refresh holds it: those run in phases, and
rebuilding rules between two of them would resurrect what the operation
just tore down.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* refactor(site-replication): invert reconcile dependency to satisfy layers

`startup_services.rs` sits in the infra layer and was calling the
reconcilers in `admin::handlers::site_replication`, which is interface —
a reverse dependency that check_layer_dependencies.sh rejects.

Moving the reconcilers down is not viable in this change: they rest on
the site-replication state core (`SiteReplicationState` alone has 107
in-file uses, `load_site_replication_state` 38, the state lock 33), so
relocating it would move ~2000 lines and ~200 call sites through a
bug-fix PR.

Invert the direction instead. A new infra module owns the contract and
the schedule; the admin layer registers its reconciler from
`register_site_replication_route`, which runs while the admin router is
built — `init_startup_http_servers` awaits that before
`init_startup_runtime_services` reconciles, so the hook is always
installed in time. No logic moves and no baseline entry is added: the
dependency genuinely reverses.

The lifecycle guard now wraps both reconcilers in one round rather than
each separately, closing the window between them.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(site-replication): harden reconcile per review feedback

Addresses the automated review on #5292.

Security: secret recovery from bucket targets accepted any target carrying
the `site-replicator-0` access key. Bucket targets are writable by anyone
holding `admin:SetBucketTarget`, so such a principal could plant a secret
and have reconciliation recreate the broadly privileged replication
account with it. A target must now name a peer in the persisted state and
point at that peer's recorded endpoint, disagreeing targets abort the
recovery, and only missing/unreadable-account errors may trigger it at
all — a transient store failure no longer rewrites a live account.

Durability: the repair no longer deletes before creating. A readable
account is rebound in place through a new `parent_user` field on
`UpdateServiceAccountOpts`, gated to `site-replicator-0` under
`allow_site_replicator_account` exactly as the account itself is. The
parent also lives in the session-token claims, and
`prepare_service_account_auth` denies the account when the two disagree,
so both move together.

Availability: the reconcile scheduler no longer requires an inline IAM
bootstrap. Deferred IAM recovers in the background with no callback into
the scheduler, which left a recovered node with self-pointing rules until
the next restart. It now starts unconditionally and returns early while
IAM or the object store are unavailable. Its first pass runs inside the
task, so walking every bucket no longer delays startup.

Correctness: an endpoint refresh commits bucket targets and peer state in
separate steps without holding the lifecycle lock, so a tick landing
between them rewrote targets from the stale endpoint; the reconciler now
also skips while any pending marker is set. Rule repair preserves an
operator-authored `role` and clears only sender-owned site-replication
ARNs, matching the merge path.

Hot path: the per-object missing-target message returns to debug. The
condition is reported once per bucket per reconcile pass instead.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: houseme <housemecn@gmail.com>

crates/e2e_test/src/replication_extension_test.rs

@@ -4803,6 +4803,153 @@ async fn test_site_replication_replicates_object_with_bucket_versioning_real_dua
     Ok(())
 }
 
+/// Re-applying a site's own replication config must not disable the peer's reverse direction.
+///
+/// `PutBucketReplication` broadcasts the config to every peer — the console's replication
+/// Save button, `mc replicate import`, and a bucket-metadata import all go through it. The
+/// receiver used to overwrite its rules with the sender's, whose destination ARN names the
+/// receiver itself. No bucket target can satisfy that ARN, so every object written on the
+/// receiver was dropped with only a debug line, while `replicate status` still reported
+/// "1/1 Buckets in sync" because both configs were byte-identical.
+#[tokio::test]
+#[serial]
+async fn test_site_replication_config_broadcast_keeps_reverse_direction_real_dual_node() -> TestResult {
+    init_logging();
+
+    let mut source_env = RustFSTestEnvironment::new().await?;
+    source_env
+        .start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
+        .await?;
+
+    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env
+        .start_rustfs_server_without_cleanup_with_env(LOOPBACK_REPLICATION_TARGET_ENV)
+        .await?;
+
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+    let bucket = "site-repl-config-broadcast";
+
+    let add_status = site_replication_add(
+        &source_env,
+        &[
+            PeerSite {
+                name: "broadcast-source".to_string(),
+                endpoint: source_env.url.clone(),
+                access_key: source_env.access_key.clone(),
+                secret_key: source_env.secret_key.clone(),
+                ..Default::default()
+            },
+            PeerSite {
+                name: "broadcast-target".to_string(),
+                endpoint: target_env.url.clone(),
+                access_key: target_env.access_key.clone(),
+                secret_key: target_env.secret_key.clone(),
+                ..Default::default()
+            },
+        ],
+    )
+    .await?;
+    assert!(add_status.success, "unexpected site add result: {add_status:?}");
+    wait_for_site_replication_enabled(&source_env, 2).await?;
+    wait_for_site_replication_enabled(&target_env, 2).await?;
+
+    source_client.create_bucket().bucket(bucket).send().await?;
+    wait_for_bucket_on_target(&target_client, bucket).await?;
+
+    // Both directions work before the broadcast.
+    source_client
+        .put_object()
+        .bucket(bucket)
+        .key("from-source.txt")
+        .body(ByteStream::from_static(b"written on the initiating site"))
+        .send()
+        .await?;
+    assert_eq!(
+        wait_for_object_on_target(&target_client, bucket, "from-source.txt").await?,
+        b"written on the initiating site".to_vec(),
+    );
+    target_client
+        .put_object()
+        .bucket(bucket)
+        .key("from-target.txt")
+        .body(ByteStream::from_static(b"written on the joined site"))
+        .send()
+        .await?;
+    assert_eq!(
+        wait_for_object_on_target(&source_client, bucket, "from-target.txt").await?,
+        b"written on the joined site".to_vec(),
+    );
+
+    // Round-trip the source's own config through PutBucketReplication, exactly what the
+    // console does when an operator opens the bucket's replication page and saves it.
+    let source_config = source_client
+        .get_bucket_replication()
+        .bucket(bucket)
+        .send()
+        .await?
+        .replication_configuration
+        .ok_or("source bucket has no replication configuration")?;
+    source_client
+        .put_bucket_replication()
+        .bucket(bucket)
+        .replication_configuration(source_config)
+        .send()
+        .await?;
+
+    let target_config = wait_for_site_replication_rule(&target_client, bucket).await?;
+    let target_deployment_id = site_replication_info(&target_env)
+        .await?
+        .sites
+        .iter()
+        .find(|peer| peer.endpoint == target_env.url)
+        .map(|peer| peer.deployment_id.clone())
+        .ok_or("joined site missing from its own replication info")?;
+    for rule in &target_config.rules {
+        let destination = rule
+            .destination
+            .as_ref()
+            .map(|destination| destination.bucket.as_str())
+            .unwrap_or_default();
+        assert!(
+            !destination.contains(&target_deployment_id),
+            "joined site adopted a rule pointing at itself: {destination}"
+        );
+    }
+
+    target_client
+        .put_object()
+        .bucket(bucket)
+        .key("from-target-after-broadcast.txt")
+        .body(ByteStream::from_static(b"written after the config broadcast"))
+        .send()
+        .await?;
+    assert_eq!(
+        wait_for_object_on_target(&source_client, bucket, "from-target-after-broadcast.txt").await?,
+        b"written after the config broadcast".to_vec(),
+        "config broadcast made replication one-directional"
+    );
+
+    Ok(())
+}
+
+async fn wait_for_site_replication_rule(
+    client: &aws_sdk_s3::Client,
+    bucket: &str,
+) -> Result<aws_sdk_s3::types::ReplicationConfiguration, Box<dyn Error + Send + Sync>> {
+    for _ in 0..40 {
+        if let Ok(response) = client.get_bucket_replication().bucket(bucket).send().await
+            && let Some(config) = response.replication_configuration
+            && !config.rules.is_empty()
+        {
+            return Ok(config);
+        }
+        sleep(Duration::from_millis(250)).await;
+    }
+
+    Err(format!("bucket {bucket} never reported a replication rule").into())
+}
+
 #[tokio::test]
 #[serial]
 async fn test_site_replication_active_active_converges_without_loops_real_dual_node() -> TestResult {

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -807,15 +807,29 @@ impl BucketTargetSys {
             && !new_targets.is_empty()
         {
             for target in &new_targets.targets {
-                if let Ok(client) = self.get_remote_target_client_internal(target).await {
-                    arn_remotes_map.insert(
-                        target.arn.clone(),
-                        ArnTarget {
-                            client: Some(Arc::new(client)),
-                            last_refresh: OffsetDateTime::now_utc(),
-                        },
-                    );
-                    self.update_bandwidth_limit(bucket, &target.arn, target.bandwidth_limit);
+                match self.get_remote_target_client_internal(target).await {
+                    Ok(client) => {
+                        arn_remotes_map.insert(
+                            target.arn.clone(),
+                            ArnTarget {
+                                client: Some(Arc::new(client)),
+                                last_refresh: OffsetDateTime::now_utc(),
+                            },
+                        );
+                        self.update_bandwidth_limit(bucket, &target.arn, target.bandwidth_limit);
+                    }
+                    // The target stays in `targets_map`, so it keeps showing up in
+                    // `bucket remote ls` while no client exists to replicate through it —
+                    // replication then drops every object for this ARN. Without this the
+                    // rejection (loopback endpoint, bad CA, unparseable URL) left no trace
+                    // anywhere.
+                    Err(err) => warn!(
+                        bucket = %bucket,
+                        arn = %target.arn,
+                        endpoint = %target.endpoint,
+                        error = %err,
+                        "replication target client unavailable; objects for this ARN will not replicate"
+                    ),
                 }
             }
             targets_map.insert(bucket.to_string(), new_targets.targets.clone());

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -2098,14 +2098,20 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
 
     for arn in tgt_arns {
         let Some(tgt_client) = ReplicationTargetStore::remote_target_client(&bucket, &arn).await else {
+            // Deliberately debug: this fires once per object per ARN, so a target that
+            // stays unreachable would flood the log from the replication hot path. The
+            // condition is reported once per pass by the site-replication reconciler and
+            // once per rebuild by `update_all_targets`, which is where an operator can act
+            // on it; the per-object event below still records each dropped object.
             debug!(
                 event = EVENT_RESYNC_RUNTIME_SKIPPED,
                 component = LOG_COMPONENT_ECSTORE,
                 subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
                 bucket = %bucket,
+                object = %object,
                 arn = %arn,
                 reason = "target_client_missing",
-                "Skipping replication object target"
+                "Replication rule has no bucket target for its destination ARN; object not replicated"
             );
             send_local_event(EventArgs {
                 event_name: EventName::ObjectReplicationNotTracked.to_string(),

crates/iam/src/manager.rs

@@ -822,6 +822,15 @@ where
             cr.secret_key = secret;
         }
 
+        if let Some(parent_user) = opts.parent_user {
+            // Same gate as the account itself: a rebind grants the account whatever the new
+            // parent can do, so only the site-replication repair path may ask for one.
+            if parent_user.is_empty() || !opts.allow_site_replicator_account || name != SITE_REPLICATOR_SERVICE_ACCOUNT {
+                return Err(Error::IAMActionNotAllowed);
+            }
+            cr.parent_user = parent_user;
+        }
+
         if opts.name.is_some() {
             cr.name = opts.name;
         }
@@ -889,6 +898,10 @@ where
         if name == SITE_REPLICATOR_SERVICE_ACCOUNT && opts.allow_site_replicator_account {
             m.insert(SITE_REPLICATOR_CLAIM.to_owned(), Value::Bool(true));
         }
+        // The parent lives in the token as well, and `prepare_service_account_auth` denies the
+        // request when the two disagree — a rebind that updated only the credential would
+        // lock the account out.
+        m.insert("parent".to_owned(), Value::String(cr.parent_user.clone()));
 
         cr.session_token = jwt_sign(&m, &cr.secret_key)?;
 
@@ -2735,6 +2748,7 @@ mod tests {
                             description: Some("new".to_string()),
                             expiration: None,
                             status: None,
+                            parent_user: None,
                             allow_site_replicator_account: false,
                         },
                     )
@@ -3097,6 +3111,7 @@ mod tests {
             description: Some("Updated service account".to_string()),
             expiration: None,
             session_policy: Some(policy),
+            parent_user: None,
             allow_site_replicator_account: false,
         };
 

crates/iam/src/sys.rs

@@ -1634,6 +1634,13 @@ pub struct UpdateServiceAccountOpts {
     pub description: Option<String>,
     pub expiration: Option<OffsetDateTime>,
     pub status: Option<String>,
+    /// Rebind the account to a different parent.
+    ///
+    /// Only site replication sets this, and only to repair an account left pointing at a
+    /// parent that a root-credential change invalidated. Rebinding an arbitrary service
+    /// account would let it inherit another user's policies, so it is gated behind
+    /// `allow_site_replicator_account` in the same way the account itself is.
+    pub parent_user: Option<String>,
     pub allow_site_replicator_account: bool,
 }
 
@@ -2061,6 +2068,7 @@ mod tests {
                     description: None,
                     expiration: None,
                     status: None,
+                    parent_user: None,
                     allow_site_replicator_account: false,
                 },
             )
@@ -2090,6 +2098,7 @@ mod tests {
                     description: None,
                     expiration: None,
                     status: None,
+                    parent_user: None,
                     allow_site_replicator_account: false,
                 },
             )
@@ -2286,6 +2295,7 @@ mod tests {
                     description: None,
                     expiration: Some(updated_expiration),
                     status: None,
+                    parent_user: None,
                     allow_site_replicator_account: false,
                 },
             )
@@ -2332,6 +2342,7 @@ mod tests {
                     description: None,
                     expiration: Some(updated_expiration),
                     status: None,
+                    parent_user: None,
                     allow_site_replicator_account: false,
                 },
             )
@@ -2383,6 +2394,7 @@ mod tests {
                         description: None,
                         expiration: None,
                         status: Some(STATUS_ENABLED.to_string()),
+                        parent_user: None,
                         allow_site_replicator_account: false,
                     },
                 )
@@ -2401,6 +2413,7 @@ mod tests {
                     description: None,
                     expiration: None,
                     status: Some(STATUS_ENABLED.to_string()),
+                    parent_user: None,
                     allow_site_replicator_account: true,
                 },
             )
@@ -2416,6 +2429,116 @@ mod tests {
         );
     }
 
+    /// A root-credential change can leave `site-replicator-0` bound to a parent that no
+    /// longer exists, and the repair must rebind in place: deleting first would leave the
+    /// site with no replication account at all if the recreate failed. The parent also lives
+    /// in the session token, so both copies have to move together or authorization denies
+    /// the account.
+    #[tokio::test]
+    async fn test_site_replicator_parent_rebind_updates_credential_and_claim() {
+        ensure_test_global_credentials();
+
+        let store = StsTestMockStore::new(false);
+        let cache_manager = IamCache::new(store).await.unwrap();
+        let iam_sys = IamSys::new(cache_manager);
+
+        let (cred, _) = iam_sys
+            .new_service_account(
+                "stale-parent-user",
+                None,
+                NewServiceAccountOpts {
+                    access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
+                    secret_key: "siteReplicatorSecretKeyForTest1234567890".to_string(),
+                    allow_site_replicator_account: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("site replicator account should be created");
+
+        iam_sys
+            .update_service_account(
+                &cred.access_key,
+                UpdateServiceAccountOpts {
+                    session_policy: None,
+                    secret_key: None,
+                    name: None,
+                    description: None,
+                    expiration: None,
+                    status: None,
+                    parent_user: Some("current-parent-user".to_string()),
+                    allow_site_replicator_account: true,
+                },
+            )
+            .await
+            .expect("site replication repair may rebind the parent");
+
+        let (identity, claims) = iam_sys
+            .get_account_with_claims_allow_missing_exp(&cred.access_key)
+            .await
+            .expect("rebound account should still be readable");
+        assert_eq!(identity.credentials.parent_user, "current-parent-user");
+        assert_eq!(
+            claims.get("parent").and_then(Value::as_str),
+            Some("current-parent-user"),
+            "the token claim must follow the credential or authorization denies the account"
+        );
+        assert_eq!(
+            iam_sys
+                .get_site_replicator_service_account_secret(&cred.access_key)
+                .await
+                .expect("secret survives a rebind"),
+            cred.secret_key,
+            "peers keep using the same secret, so a rebind must not rotate it"
+        );
+    }
+
+    /// The rebind is a site-replication repair primitive, not a general capability: letting
+    /// any caller re-parent a service account would let it inherit another user's policies.
+    #[tokio::test]
+    async fn test_parent_rebind_is_rejected_for_ordinary_service_accounts() {
+        ensure_test_global_credentials();
+
+        let store = StsTestMockStore::new(false);
+        let cache_manager = IamCache::new(store).await.unwrap();
+        let iam_sys = IamSys::new(cache_manager);
+
+        let (cred, _) = iam_sys
+            .new_service_account(
+                "ordinary-parent",
+                None,
+                NewServiceAccountOpts {
+                    access_key: "ordinary-service-account".to_string(),
+                    secret_key: "ordinaryServiceAccountSecret1234567890".to_string(),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("ordinary service account should be created");
+
+        for allow in [false, true] {
+            assert!(
+                iam_sys
+                    .update_service_account(
+                        &cred.access_key,
+                        UpdateServiceAccountOpts {
+                            session_policy: None,
+                            secret_key: None,
+                            name: None,
+                            description: None,
+                            expiration: None,
+                            status: None,
+                            parent_user: Some("victim-user".to_string()),
+                            allow_site_replicator_account: allow,
+                        },
+                    )
+                    .await
+                    .is_err(),
+                "re-parenting an ordinary service account must be rejected (allow={allow})"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn test_created_access_token_authorizes_with_parent_policy() {
         ensure_test_global_credentials();

crates/madmin/src/site_replication.rs

@@ -509,6 +509,13 @@ pub struct SRBucketInfo {
     pub cors_config_updated_at: Option<OffsetDateTime>,
     #[serde(default, skip_serializing_if = "String::is_empty")]
     pub location: String,
+    /// Whether every `site-repl-*` rule on the reporting site resolves to a usable bucket
+    /// target. A rule without one silently drops every object, and the rule set alone cannot
+    /// reveal it — the config is well-formed, the endpoint behind it is not reachable.
+    ///
+    /// `None` means the peer predates this field: treat it as "unknown", never as a fault.
+    #[serde(rename = "replicationTargetsOnline", default, skip_serializing_if = "Option::is_none")]
+    pub replication_targets_online: Option<bool>,
     #[serde(rename = "apiVersion", skip_serializing_if = "Option::is_none")]
     pub api_version: Option<String>,
 }