fix(site-replication): carry user credentials and service accounts in the IAM snapshot (backlog#2289)

The IAM snapshot used by the retry drain resend, repair and site-add
bootstrap was built from `list_users`, which strips secret keys and skips
service accounts. The plan builder dropped every user for lack of a
secret, so a user disable, secret rotation or service-account change
committed while a peer was unreachable never reached it — while the
collapsed retry entry was settled and repair reported success.

Read the credentials separately at plan time (`build_sr_iam_credentials`,
used only on peer-delivery paths) so `SRInfo`, which is served to admin
callers, stays secret-free. Users travel with secret, status and the user
record's own update time; service accounts (except the replicator's) travel
as the create item the live hook emits, after their parents. The receiver
applies a disabled status after creating a new service account, and the
retry snapshot tombstones removed service accounts like the other kinds.

`encode_service_account_replication_policy` moves into the infra layer so
the snapshot builder can share it with the live hook.

(cherry picked from commit bd8cd497c965ae331fafa20362763224cec3b5b2)

rustfs/src/admin/handlers/site_replication.rs

@@ -66,8 +66,8 @@ use rustfs_madmin::{
     ReplicateRemoveStatus, ResyncBucketStatus, SITE_REPL_API_VERSION, SR_IAM_ITEM_STS_ACC, SR_IAM_ITEM_STS_ACC_LEGACY,
     SRBucketInfo, SRBucketMeta, SRBucketStatsSummary, SRGroupInfo, SRGroupStatsSummary, SRIAMItem, SRIAMUser,
     SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq, SRPendingOperation, SRPolicyMapping,
-    SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRSTSCredential, SRSessionPolicy, SRSiteSummary, SRStateEditReq,
-    SRStateInfo, SRStatusInfo, SRSvcAccChange, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
+    SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRSTSCredential, SRSiteSummary, SRStateEditReq, SRStateInfo,
+    SRStatusInfo, SRSvcAccChange, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat,
 };
 use rustfs_policy::policy::{
     Policy,
@@ -98,7 +98,6 @@ use uuid::Uuid;
 // paths keep resolving while this file keeps only the HTTP handlers.
 pub(crate) use crate::site_replication::*;
 
-const SERVICE_ACCOUNT_ENVELOPE_VERSION: u64 = 2;
 // Serializes peer-join admission (staleness check -> IAM upsert -> state
 // commit) across every node of this site; see admit_peer_join. Never an
 // actual object — only a namespace-lock key, like the repair execution lock.
@@ -1980,7 +1979,7 @@ async fn bootstrap_existing_metadata_after_add(
             return errors;
         }
     };
-    let plan = match site_replication_bootstrap_plan(&info) {
+    let plan = match build_site_replication_bootstrap_plan(&info).await {
         Ok(plan) => plan,
         Err(err) => {
             let mut errors = SiteReplicationErrorSummary::default();
@@ -5660,41 +5659,6 @@ fn group_info_requires_upsert(update: &rustfs_madmin::GroupAddRemove) -> bool {
     !update.is_remove
 }
 
-pub(crate) fn encode_service_account_replication_policy(
-    claims: &HashMap<String, Value>,
-    session_policy: Option<&str>,
-) -> S3Result<(SRSessionPolicy, Option<rustfs_madmin::SRSvcAccReplicationEnvelope>)> {
-    if !claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM) {
-        return session_policy
-            .map(SRSessionPolicy::from_json)
-            .transpose()
-            .map(|policy| policy.unwrap_or_default())
-            .map(|policy| (policy, None))
-            .map_err(|err| s3_error!(InvalidArgument, "marshal policy failed: {:?}", err));
-    }
-
-    let policy = match session_policy {
-        Some(policy) => serde_json::from_str::<Policy>(policy)
-            .map_err(|err| s3_error!(InvalidArgument, "invalid service account replication policy: {:?}", err))?,
-        None => Policy::default(),
-    };
-    if policy.statements.is_empty() && (!policy.id.is_empty() || !policy.version.is_empty())
-        || policy.version.is_empty() && !policy.statements.is_empty()
-    {
-        return Err(s3_error!(InvalidArgument, "service account replication policy is not normalized"));
-    }
-    let policy = serde_json::to_string(&policy)
-        .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?;
-    let policy = SRSessionPolicy::from_json(&policy)
-        .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?;
-    Ok((
-        policy,
-        Some(rustfs_madmin::SRSvcAccReplicationEnvelope {
-            version: SERVICE_ACCOUNT_ENVELOPE_VERSION,
-        }),
-    ))
-}
-
 #[derive(Debug)]
 struct ReplicatedServiceAccountPolicy {
     policy: Option<Policy>,
@@ -5979,6 +5943,8 @@ async fn apply_iam_service_account_item(
                     .map_err(ApiError::from)?;
             }
             Err(err) if is_err_no_such_service_account(&err) => {
+                let access_key = create.access_key.clone();
+                let status = create.status.clone();
                 iam_sys
                     .new_service_account(
                         &create.parent,
@@ -5996,6 +5962,28 @@ async fn apply_iam_service_account_item(
                     )
                     .await
                     .map_err(ApiError::from)?;
+                // A snapshot (bootstrap / repair / retry resend) carries the
+                // account's current status; creation always enables, so a
+                // disabled account must be switched off in a second step or
+                // the peer keeps accepting credentials the source rejects.
+                if !status.is_empty() && status != "on" {
+                    iam_sys
+                        .update_service_account(
+                            &access_key,
+                            UpdateServiceAccountOpts {
+                                session_policy: None,
+                                secret_key: None,
+                                name: None,
+                                description: None,
+                                expiration: None,
+                                status: Some(status),
+                                parent_user: None,
+                                allow_site_replicator_account: false,
+                            },
+                        )
+                        .await
+                        .map_err(ApiError::from)?;
+                }
             }
             Err(err) => return Err(ApiError::from(err).into()),
         }
@@ -7692,7 +7680,7 @@ impl Operation for SiteReplicationRepairHandler {
         let local_peer = current_local_peer(&req, &state);
         let body: SiteReplicationRepairRequest = read_site_replication_json(req, "", false).await?;
         let info = build_sr_info(&state, &local_peer).await?;
-        let plan = site_replication_bootstrap_plan(&info)?;
+        let plan = build_site_replication_bootstrap_plan(&info).await?;
         let signing_key = current_token_signing_key().ok_or_else(|| {
             S3Error::with_message(S3ErrorCode::InternalError, "token signing key is not initialized".to_string())
         })?;
@@ -7885,6 +7873,7 @@ impl Operation for SRRotateServiceAccountHandler {
 mod tests {
     use super::*;
     use crate::site_replication::identity::deployment_id_for_endpoint;
+    use rustfs_madmin::SRSessionPolicy;
 
     /// A peer the status probe could not reach must render as offline.
     ///

rustfs/src/site_replication/hooks.rs

@@ -302,7 +302,164 @@ pub(crate) fn site_replication_state_replicates_ilm_expiry(state: &SiteReplicati
     state.peers.values().any(|peer| peer.replicate_ilm_expiry)
 }
 
-pub(crate) fn site_replication_bootstrap_plan(info: &SRInfo) -> S3Result<SiteReplicationBootstrapPlan> {
+/// Secret-bearing half of the IAM snapshot. `SRInfo` is served to admin
+/// callers (`site-replication/info`, status, add preflight) and must stay
+/// secret-free, so the bootstrap plan receives credentials through this
+/// separate value, built only on the paths that deliver to peers (site add
+/// bootstrap, repair, retry snapshot resend). Never persisted, never served.
+#[derive(Debug, Clone, Default)]
+pub(crate) struct SiteReplicationIamCredentials {
+    /// Built-in users (access key -> credential); temp and service accounts
+    /// are excluded, external/IdP users never appear here.
+    pub(crate) users: BTreeMap<String, SiteReplicationUserCredential>,
+    /// Every service account except the site replicator's own, already
+    /// shaped as the `service-account` create item the live hook emits.
+    pub(crate) service_accounts: Vec<SiteReplicationServiceAccountSnapshot>,
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct SiteReplicationUserCredential {
+    pub(crate) secret_key: String,
+    pub(crate) status: AccountStatus,
+    /// The user record's own update time (the axis the receiver's staleness
+    /// check compares against), unlike `UserInfo::updated_at` which
+    /// `list_users` overwrites with the policy mapping's time.
+    pub(crate) updated_at: Option<OffsetDateTime>,
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct SiteReplicationServiceAccountSnapshot {
+    pub(crate) create: SRSvcAccCreate,
+    pub(crate) envelope: Option<SRSvcAccReplicationEnvelope>,
+    pub(crate) updated_at: Option<OffsetDateTime>,
+}
+
+pub(crate) const SERVICE_ACCOUNT_ENVELOPE_VERSION: u64 = 2;
+
+pub(crate) fn encode_service_account_replication_policy(
+    claims: &HashMap<String, Value>,
+    session_policy: Option<&str>,
+) -> S3Result<(SRSessionPolicy, Option<SRSvcAccReplicationEnvelope>)> {
+    if !claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM) {
+        return session_policy
+            .map(SRSessionPolicy::from_json)
+            .transpose()
+            .map(|policy| policy.unwrap_or_default())
+            .map(|policy| (policy, None))
+            .map_err(|err| s3_error!(InvalidArgument, "marshal policy failed: {:?}", err));
+    }
+
+    let policy = match session_policy {
+        Some(policy) => serde_json::from_str::<Policy>(policy)
+            .map_err(|err| s3_error!(InvalidArgument, "invalid service account replication policy: {:?}", err))?,
+        None => Policy::default(),
+    };
+    if policy.statements.is_empty() && (!policy.id.is_empty() || !policy.version.is_empty())
+        || policy.version.is_empty() && !policy.statements.is_empty()
+    {
+        return Err(s3_error!(InvalidArgument, "service account replication policy is not normalized"));
+    }
+    let policy = serde_json::to_string(&policy)
+        .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?;
+    let policy = SRSessionPolicy::from_json(&policy)
+        .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?;
+    Ok((
+        policy,
+        Some(SRSvcAccReplicationEnvelope {
+            version: SERVICE_ACCOUNT_ENVELOPE_VERSION,
+        }),
+    ))
+}
+
+/// Read the credentials the IAM snapshot needs straight from the IAM store:
+/// `list_users` deliberately strips secret keys and skips service accounts,
+/// which is right for an admin listing and wrong for a peer snapshot (the
+/// plan builder used to drop every user for lack of a secret, so a status
+/// change or secret rotation committed while a peer was unreachable never
+/// reached it — backlog#2289).
+pub(crate) async fn build_sr_iam_credentials() -> S3Result<SiteReplicationIamCredentials> {
+    let mut credentials = SiteReplicationIamCredentials::default();
+    let Some(iam_sys) = current_iam_handle() else {
+        return Ok(credentials);
+    };
+
+    let mut users = HashMap::new();
+    iam_sys.load_users(UserType::Reg, &mut users).await.map_err(ApiError::from)?;
+    for (access_key, identity) in users {
+        if identity.credentials.is_temp() || identity.credentials.is_service_account() {
+            continue;
+        }
+        credentials.users.insert(
+            access_key,
+            SiteReplicationUserCredential {
+                secret_key: identity.credentials.secret_key,
+                status: if identity.credentials.status == "off" {
+                    AccountStatus::Disabled
+                } else {
+                    AccountStatus::Enabled
+                },
+                updated_at: identity.update_at,
+            },
+        );
+    }
+
+    let mut service_accounts = HashMap::new();
+    iam_sys
+        .load_users(UserType::Svc, &mut service_accounts)
+        .await
+        .map_err(ApiError::from)?;
+    let mut service_accounts: Vec<_> = service_accounts.into_iter().collect();
+    service_accounts.sort_by(|(a, _), (b, _)| a.cmp(b));
+    for (access_key, identity) in service_accounts {
+        // The replicator account is installed by join / rotate, never by a snapshot.
+        if access_key == SITE_REPLICATOR_SERVICE_ACCOUNT || !identity.credentials.is_service_account() {
+            continue;
+        }
+        let claims = iam_sys.get_claims_for_svc_acc(&access_key).await.map_err(ApiError::from)?;
+        let (account, session_policy) = iam_sys.get_service_account(&access_key).await.map_err(ApiError::from)?;
+        let session_policy = session_policy
+            .map(|policy| serde_json::to_string(&policy))
+            .transpose()
+            .map_err(|err| {
+                S3Error::with_message(
+                    S3ErrorCode::InternalError,
+                    format!("marshal service account session policy failed: {err:?}"),
+                )
+            })?;
+        let (session_policy, envelope) = encode_service_account_replication_policy(&claims, session_policy.as_deref())?;
+        credentials.service_accounts.push(SiteReplicationServiceAccountSnapshot {
+            create: SRSvcAccCreate {
+                parent: identity.credentials.parent_user,
+                access_key,
+                secret_key: identity.credentials.secret_key,
+                groups: identity.credentials.groups.unwrap_or_default(),
+                claims,
+                session_policy,
+                status: identity.credentials.status,
+                name: account.name.unwrap_or_default(),
+                description: account.description.unwrap_or_default(),
+                expiration: account.expiration,
+                api_version: Some(SITE_REPL_API_VERSION.to_string()),
+            },
+            envelope,
+            updated_at: identity.update_at,
+        });
+    }
+
+    Ok(credentials)
+}
+
+/// The bootstrap plan for peer delivery: `info` (secret-free) plus the IAM
+/// credentials read at this moment.
+pub(crate) async fn build_site_replication_bootstrap_plan(info: &SRInfo) -> S3Result<SiteReplicationBootstrapPlan> {
+    let credentials = build_sr_iam_credentials().await?;
+    site_replication_bootstrap_plan(info, &credentials)
+}
+
+pub(crate) fn site_replication_bootstrap_plan(
+    info: &SRInfo,
+    credentials: &SiteReplicationIamCredentials,
+) -> S3Result<SiteReplicationBootstrapPlan> {
     let mut plan = SiteReplicationBootstrapPlan::default();
     let replicate_ilm_expiry = site_replication_info_replicates_ilm_expiry(info);
 
@@ -318,24 +475,57 @@ pub(crate) fn site_replication_bootstrap_plan(info: &SRInfo) -> S3Result<SiteRep
     }
 
     for (access_key, user) in &info.user_info_map {
-        if let Some(secret_key) = &user.secret_key {
-            plan.iam_items.push(SRIAMItem {
-                r#type: "iam-user".to_string(),
-                iam_user: Some(rustfs_madmin::SRIAMUser {
-                    access_key: access_key.clone(),
-                    is_delete_req: false,
-                    user_req: Some(AddOrUpdateUserReq {
-                        secret_key: secret_key.clone(),
-                        policy: user.policy_name.clone(),
-                        status: user.status.clone(),
-                    }),
-                    api_version: Some(SITE_REPL_API_VERSION.to_string()),
+        // Credentials come from the store snapshot; an inline `secret_key` on
+        // the SRInfo entry (older callers, tests) is accepted as a fallback.
+        // Users with neither (external / IdP identities) have nothing a peer
+        // could install and are skipped.
+        let credential = credentials.users.get(access_key);
+        let Some(secret_key) = credential
+            .map(|credential| credential.secret_key.clone())
+            .or_else(|| user.secret_key.clone())
+            .filter(|secret_key| !secret_key.is_empty())
+        else {
+            continue;
+        };
+        let status = credential
+            .map(|credential| credential.status.clone())
+            .unwrap_or_else(|| user.status.clone());
+        let updated_at = credential.and_then(|credential| credential.updated_at).or(user.updated_at);
+        plan.iam_items.push(SRIAMItem {
+            r#type: "iam-user".to_string(),
+            iam_user: Some(rustfs_madmin::SRIAMUser {
+                access_key: access_key.clone(),
+                is_delete_req: false,
+                user_req: Some(AddOrUpdateUserReq {
+                    secret_key,
+                    policy: user.policy_name.clone(),
+                    status,
                 }),
-                updated_at: user.updated_at,
+                api_version: Some(SITE_REPL_API_VERSION.to_string()),
+            }),
+            updated_at,
+            api_version: Some(SITE_REPL_API_VERSION.to_string()),
+            ..Default::default()
+        });
+    }
+
+    // Service accounts follow their parents: the receiver creates a missing
+    // account under `parent` and updates an existing one (secret, status,
+    // session policy), so a rotation or disable committed during an outage
+    // converges through the same snapshot as users do.
+    for account in &credentials.service_accounts {
+        plan.iam_items.push(SRIAMItem {
+            r#type: "service-account".to_string(),
+            svc_acc_change: Some(SRSvcAccChange {
+                create: Some(account.create.clone()),
+                oidc_service_account_envelope: account.envelope.clone(),
                 api_version: Some(SITE_REPL_API_VERSION.to_string()),
                 ..Default::default()
-            });
-        }
+            }),
+            updated_at: account.updated_at,
+            api_version: Some(SITE_REPL_API_VERSION.to_string()),
+            ..Default::default()
+        });
     }
 
     for (name, desc) in &info.group_desc_map {

rustfs/src/site_replication/mod.rs

@@ -79,13 +79,16 @@ use http::header::{CONTENT_TYPE, HOST};
 use http::{HeaderMap, HeaderValue, Uri};
 use hyper::{Method, StatusCode};
 use rustfs_config::{DEFAULT_CONSOLE_ADDRESS, DEFAULT_RUSTFS_TLS_PATH, ENV_RUSTFS_CONSOLE_ADDRESS, ENV_RUSTFS_TLS_PATH};
+use rustfs_iam::federation::OIDC_VIRTUAL_PARENT_CLAIM;
 use rustfs_iam::store::{MappedPolicy, UserType, sr_wire_user_type};
 use rustfs_iam::sys::SITE_REPLICATOR_SERVICE_ACCOUNT;
 use rustfs_madmin::{
-    AddOrUpdateUserReq, GroupAddRemove, GroupStatus, PeerInfo, PeerSite, ReplicateEditStatus, SITE_REPL_API_VERSION,
-    SRBucketInfo, SRBucketMeta, SRGroupInfo, SRIAMItem, SRIAMPolicy, SRInfo, SRPolicyMapping, SRRemoveReq, SRResyncOpStatus,
-    SRRetryStats, SRStateInfo, SyncStatus,
+    AccountStatus, AddOrUpdateUserReq, GroupAddRemove, GroupStatus, PeerInfo, PeerSite, ReplicateEditStatus,
+    SITE_REPL_API_VERSION, SRBucketInfo, SRBucketMeta, SRGroupInfo, SRIAMItem, SRIAMPolicy, SRInfo, SRPolicyMapping, SRRemoveReq,
+    SRResyncOpStatus, SRRetryStats, SRSessionPolicy, SRStateInfo, SRSvcAccChange, SRSvcAccCreate, SRSvcAccDelete,
+    SRSvcAccReplicationEnvelope, SyncStatus,
 };
+use rustfs_policy::policy::Policy;
 use rustfs_signer::constants::UNSIGNED_PAYLOAD;
 use rustfs_signer::sign_v4;
 use rustfs_tls_runtime::{GlobalPublishedOutboundTlsState, TlsGeneration};

rustfs/src/site_replication/repair.rs

@@ -726,7 +726,7 @@ pub(crate) async fn execute_site_replication_repair_locked(
         return Err(s3_error!(InvalidRequest, "site replication is not configured"));
     }
     let info = build_sr_info(&state, &request.local_peer).await?;
-    let plan = site_replication_bootstrap_plan(&info)?;
+    let plan = build_site_replication_bootstrap_plan(&info).await?;
     let plan_token = site_replication_repair_plan_token(&state, &plan)?;
     let preflight_token = site_replication_repair_preflight_token(&state, &plan, request.signing_key.as_bytes())?;
     let sites = site_replication_repair_sites(&state, &request.local_peer, &plan, request.signing_key.as_bytes())?;

rustfs/src/site_replication/retry.rs

@@ -954,6 +954,7 @@ pub(crate) enum IamSnapshotKey {
     User(String),
     Group(String),
     PolicyMapping { target: String, user_type: i64, is_group: bool },
+    ServiceAccount(String),
 }
 
 pub(crate) fn iam_snapshot_key(item: &SRIAMItem) -> Option<IamSnapshotKey> {
@@ -972,6 +973,11 @@ pub(crate) fn iam_snapshot_key(item: &SRIAMItem) -> Option<IamSnapshotKey> {
             user_type: mapping.user_type,
             is_group: mapping.is_group,
         }),
+        "service-account" => item
+            .svc_acc_change
+            .as_ref()
+            .and_then(|change| change.create.as_ref())
+            .map(|create| IamSnapshotKey::ServiceAccount(create.access_key.clone())),
         _ => None,
     }
 }
@@ -1006,6 +1012,24 @@ pub(crate) fn iam_snapshot_tombstones(item: &SRIAMItem, observed_at: OffsetDateT
                 mapping.policy.clear();
             }
         }
+        "service-account" => {
+            let Some(access_key) = item
+                .svc_acc_change
+                .as_ref()
+                .and_then(|change| change.create.as_ref())
+                .map(|create| create.access_key.clone())
+            else {
+                return Vec::new();
+            };
+            tombstone.svc_acc_change = Some(SRSvcAccChange {
+                delete: Some(SRSvcAccDelete {
+                    access_key,
+                    api_version: Some(SITE_REPL_API_VERSION.to_string()),
+                }),
+                api_version: Some(SITE_REPL_API_VERSION.to_string()),
+                ..Default::default()
+            });
+        }
         _ => return Vec::new(),
     }
     vec![tombstone]
@@ -1701,7 +1725,7 @@ pub(crate) async fn drain_site_replication_retry_queue_locked(
     // tick and only when a snapshot resend is actually due.
     let plan = if needs_plan {
         let info = build_sr_info(&runtime.state, &runtime.local_peer).await?;
-        Some(site_replication_bootstrap_plan(&info)?)
+        Some(build_site_replication_bootstrap_plan(&info).await?)
     } else {
         None
     };
@@ -1841,7 +1865,7 @@ pub(crate) async fn drain_one_site_replication_retry_event(
                     }
                 }
                 let fresh_info = build_sr_info(&runtime.state, &runtime.local_peer).await?;
-                let fresh_plan = site_replication_bootstrap_plan(&fresh_info)?;
+                let fresh_plan = build_site_replication_bootstrap_plan(&fresh_info).await?;
                 let fresh_snapshot = RetrySnapshot::from_plan(&action, &fresh_plan).expect("snapshot action has a snapshot");
                 if fresh_snapshot.fingerprint()? == current_fingerprint {
                     if is_iam {

rustfs/src/site_replication/tests.rs

@@ -1679,7 +1679,8 @@ fn test_site_replication_bootstrap_plan_includes_replayable_snapshot_items() {
         },
     );
 
-    let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build");
+    let plan =
+        site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build");
 
     assert_eq!(plan.iam_items.iter().map(|item| item.r#type.as_str()).collect::<Vec<_>>(), {
         vec!["policy", "iam-user", "group-info", "policy-mapping"]
@@ -1717,7 +1718,8 @@ fn test_site_replication_bootstrap_plan_skips_lifecycle_by_default() {
         },
     );
 
-    let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build");
+    let plan =
+        site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build");
 
     assert!(!plan.bucket_items.iter().any(|item| item.r#type == "lc-config"));
 }
@@ -1748,7 +1750,8 @@ fn test_site_replication_bootstrap_plan_emits_timestamped_lifecycle_delete() {
         },
     );
 
-    let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build");
+    let plan =
+        site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build");
 
     let item = plan
         .bucket_items
@@ -1935,8 +1938,8 @@ fn test_site_replication_repair_preflight_token_is_deterministic_for_equal_state
         },
     );
 
-    let plan_a = site_replication_bootstrap_plan(&info).expect("first plan");
-    let plan_b = site_replication_bootstrap_plan(&info).expect("second plan");
+    let plan_a = site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("first plan");
+    let plan_b = site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("second plan");
     let token_a = site_replication_repair_preflight_token(&state, &plan_a, b"test-signing-key").expect("first token");
     let token_b = site_replication_repair_preflight_token(&state, &plan_b, b"test-signing-key").expect("second token");
 
@@ -3219,3 +3222,148 @@ fn test_reconcile_adds_missing_peer_rules_to_existing_config() {
     assert!(rule_ids.contains(&"site-repl-dep-b"));
     assert!(rule_ids.contains(&"site-repl-dep-c"));
 }
+
+/// backlog#2289: the IAM snapshot (retry resend, repair, site-add bootstrap)
+/// used to be built from `list_users`, whose `UserInfo` never carries a
+/// secret key, so the plan dropped every user and a status change or secret
+/// rotation committed while a peer was unreachable never reached it. The
+/// credentials now come from a separate store read; SRInfo stays secret-free.
+#[test]
+fn test_bootstrap_plan_carries_users_from_the_credential_snapshot() {
+    let mut info = SRInfo::default();
+    // Exactly what `list_users` builds: status, policy, updated_at — never secret_key.
+    info.user_info_map.insert(
+        "alice".to_string(),
+        rustfs_madmin::UserInfo {
+            status: rustfs_madmin::AccountStatus::Disabled,
+            policy_name: Some("readwrite".to_string()),
+            updated_at: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")),
+            ..Default::default()
+        },
+    );
+    info.user_info_map.insert(
+        "external-idp-user".to_string(),
+        rustfs_madmin::UserInfo {
+            status: rustfs_madmin::AccountStatus::Enabled,
+            ..Default::default()
+        },
+    );
+    let user_updated_at = OffsetDateTime::from_unix_timestamp(1_700_000_500).expect("timestamp");
+    let mut credentials = SiteReplicationIamCredentials::default();
+    credentials.users.insert(
+        "alice".to_string(),
+        SiteReplicationUserCredential {
+            secret_key: "alice-secret".to_string(),
+            status: rustfs_madmin::AccountStatus::Disabled,
+            updated_at: Some(user_updated_at),
+        },
+    );
+
+    let plan = site_replication_bootstrap_plan(&info, &credentials).expect("bootstrap plan should build");
+
+    let users: Vec<_> = plan.iam_items.iter().filter(|item| item.r#type == "iam-user").collect();
+    assert_eq!(users.len(), 1, "only the user with a credential travels: {:?}", plan.iam_items);
+    let alice = users[0].iam_user.as_ref().expect("iam user body");
+    assert_eq!(alice.access_key, "alice");
+    let req = alice.user_req.as_ref().expect("user request");
+    assert_eq!(req.secret_key, "alice-secret");
+    assert_eq!(req.status, rustfs_madmin::AccountStatus::Disabled);
+    assert_eq!(req.policy.as_deref(), Some("readwrite"));
+    // the user record's own axis, not the policy-mapping time list_users reports
+    assert_eq!(users[0].updated_at, Some(user_updated_at));
+}
+
+fn service_account_snapshot(access_key: &str, parent: &str, status: &str) -> SiteReplicationServiceAccountSnapshot {
+    SiteReplicationServiceAccountSnapshot {
+        create: rustfs_madmin::SRSvcAccCreate {
+            parent: parent.to_string(),
+            access_key: access_key.to_string(),
+            secret_key: format!("{access_key}-secret"),
+            groups: Vec::new(),
+            claims: HashMap::new(),
+            session_policy: SRSessionPolicy::default(),
+            status: status.to_string(),
+            name: String::new(),
+            description: String::new(),
+            expiration: None,
+            api_version: Some(SITE_REPL_API_VERSION.to_string()),
+        },
+        envelope: None,
+        updated_at: Some(OffsetDateTime::from_unix_timestamp(1_700_000_600).expect("timestamp")),
+    }
+}
+
+/// backlog#2289: service accounts were absent from every snapshot (the
+/// listing filters them). They now travel as the create item the live hook
+/// emits — after their parents — carrying secret and status.
+#[test]
+fn test_bootstrap_plan_emits_service_accounts_after_their_parents() {
+    let mut info = SRInfo::default();
+    info.user_info_map
+        .insert("alice".to_string(), rustfs_madmin::UserInfo::default());
+    let mut credentials = SiteReplicationIamCredentials::default();
+    credentials.users.insert(
+        "alice".to_string(),
+        SiteReplicationUserCredential {
+            secret_key: "alice-secret".to_string(),
+            status: rustfs_madmin::AccountStatus::Enabled,
+            updated_at: None,
+        },
+    );
+    credentials
+        .service_accounts
+        .push(service_account_snapshot("alice-svc", "alice", "off"));
+
+    let plan = site_replication_bootstrap_plan(&info, &credentials).expect("bootstrap plan should build");
+
+    let types: Vec<_> = plan.iam_items.iter().map(|item| item.r#type.as_str()).collect();
+    assert_eq!(types, vec!["iam-user", "service-account"]);
+    let change = plan.iam_items[1].svc_acc_change.as_ref().expect("service account change");
+    let create = change.create.as_ref().expect("create body");
+    assert_eq!((create.access_key.as_str(), create.parent.as_str()), ("alice-svc", "alice"));
+    assert_eq!(create.secret_key, "alice-svc-secret");
+    assert_eq!(create.status, "off", "a disabled account must arrive disabled");
+    assert!(change.delete.is_none() && change.update.is_none());
+}
+
+/// A service account present in the previous snapshot but gone from the
+/// fresh one is replayed as an explicit delete, like the other IAM kinds.
+#[test]
+fn test_retry_snapshot_tombstones_removed_service_accounts() {
+    let observed_at = OffsetDateTime::from_unix_timestamp(1_700_001_000).expect("timestamp");
+    let mut info = SRInfo::default();
+    info.user_info_map
+        .insert("alice".to_string(), rustfs_madmin::UserInfo::default());
+    let mut credentials = SiteReplicationIamCredentials::default();
+    credentials.users.insert(
+        "alice".to_string(),
+        SiteReplicationUserCredential {
+            secret_key: "alice-secret".to_string(),
+            status: rustfs_madmin::AccountStatus::Enabled,
+            updated_at: None,
+        },
+    );
+    let mut with_account = credentials.clone();
+    with_account
+        .service_accounts
+        .push(service_account_snapshot("alice-svc", "alice", "on"));
+    let previous = site_replication_bootstrap_plan(&info, &with_account).expect("previous plan");
+    let fresh = site_replication_bootstrap_plan(&info, &credentials).expect("fresh plan");
+
+    let replay = RetrySnapshot::replay_after_change(
+        &RetrySnapshot::Iam(previous.iam_items),
+        &RetrySnapshot::Iam(fresh.iam_items),
+        observed_at,
+    );
+    let RetrySnapshot::Iam(items) = replay else {
+        panic!("IAM snapshot expected");
+    };
+    let tombstone = items
+        .iter()
+        .find(|item| item.r#type == "service-account")
+        .expect("service account tombstone");
+    let change = tombstone.svc_acc_change.as_ref().expect("change");
+    assert_eq!(change.delete.as_ref().map(|delete| delete.access_key.as_str()), Some("alice-svc"));
+    assert!(change.create.is_none());
+    assert_eq!(tombstone.updated_at, Some(observed_at));
+}