From c3aac2279fcdc1e07972b28a4ff995529dc45b0e Mon Sep 17 00:00:00 2001 From: abdullahnah92 <157595835+abdullahnah92@users.noreply.github.com> Date: Mon, 27 Jul 2026 17:39:23 +0300 Subject: [PATCH] fix(site-replication): keep reverse direction after config broadcast (#5292) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 * 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 * 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 * 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 --------- Co-authored-by: Claude Opus 5 Co-authored-by: houseme --- .config/nextest.toml | 4 +- .../src/replication_extension_test.rs | 147 +++ .../ecstore/src/bucket/bucket_target_sys.rs | 32 +- .../replication/replication_resyncer.rs | 8 +- crates/iam/src/manager.rs | 15 + crates/iam/src/sys.rs | 123 ++ crates/madmin/src/site_replication.rs | 7 + docs/testing/e2e-suite-inventory.md | 4 +- rustfs/src/admin/handlers/service_account.rs | 1 + rustfs/src/admin/handlers/site_replication.rs | 1065 +++++++++++++++-- rustfs/src/admin/handlers/user.rs | 1 + rustfs/src/lib.rs | 1 + rustfs/src/site_replication_reconcile.rs | 81 ++ rustfs/src/startup_services.rs | 6 + 14 files changed, 1390 insertions(+), 105 deletions(-) create mode 100644 rustfs/src/site_replication_reconcile.rs diff --git a/.config/nextest.toml b/.config/nextest.toml index f14ba3abd..f264f7559 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -173,7 +173,7 @@ test-group = 'ecstore-serial-flaky' # the nightly profile derives its set as "the replication module MINUS this # allowlist", so any new replication test lands in nightly by default (never # silently unrun) until it is explicitly blessed as fast here. Keep the two -# regexes byte-identical. Count invariant: 20 here + 27 nightly = 47 total +# regexes byte-identical. Count invariant: 20 here + 28 nightly = 48 total # (authority: `cargo nextest list`; docs/testing/e2e-suite-inventory.md). # HISTORY (2026-07-11): the 20 fast tests were briefly pulled out of this lane # (#4724) because they set a loopback (127.0.0.1) replication target that the @@ -228,7 +228,7 @@ fail-fast = false # and poll until source and target converge; two replicate over HTTPS, two # pin active SSE failure contracts, and one guards event/history observers. # The SSE-S3 contract remains ignored under backlog#1291. -# * 11 `_real_dual_node` site-replication tests — each spawns TWO full rustfs +# * 12 `_real_dual_node` site-replication tests — each spawns TWO full rustfs # servers and drives the cross-process site-replication control plane. # * 1 `_real_three_node` site-replication test. # * 1 `_real_single_node` service-account round-trip test. diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index bf61f2c3c..8c6ff2a04 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/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> { + 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 { diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 9c16ae413..cf2a3e1ec 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/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()); diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index fb2f85ce5..e8a2f3472 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -2098,14 +2098,20 @@ pub async fn replicate_object(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(), diff --git a/crates/iam/src/manager.rs b/crates/iam/src/manager.rs index aa83a1b80..feb53fbe3 100644 --- a/crates/iam/src/manager.rs +++ b/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, }; diff --git a/crates/iam/src/sys.rs b/crates/iam/src/sys.rs index 47b58add2..608632b20 100644 --- a/crates/iam/src/sys.rs +++ b/crates/iam/src/sys.rs @@ -1634,6 +1634,13 @@ pub struct UpdateServiceAccountOpts { pub description: Option, pub expiration: Option, pub status: Option, + /// 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, 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(); diff --git a/crates/madmin/src/site_replication.rs b/crates/madmin/src/site_replication.rs index e131d2ed2..a8555c904 100644 --- a/crates/madmin/src/site_replication.rs +++ b/crates/madmin/src/site_replication.rs @@ -509,6 +509,13 @@ pub struct SRBucketInfo { pub cors_config_updated_at: Option, #[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, #[serde(rename = "apiVersion", skip_serializing_if = "Option::is_none")] pub api_version: Option, } diff --git a/docs/testing/e2e-suite-inventory.md b/docs/testing/e2e-suite-inventory.md index e02d48774..ec71b2729 100644 --- a/docs/testing/e2e-suite-inventory.md +++ b/docs/testing/e2e-suite-inventory.md @@ -73,7 +73,7 @@ | quota_test | 14 | | | reliability_disk_fault_test | 3 | | | reliant | 10 | 5 ✅ | -| replication_extension_test | 47 | 20 ✅ +27 🌙 | +| replication_extension_test | 48 | 20 ✅ +28 🌙 | | security_boundary_test | 4 | | | ssec_copy_test | 2 | ✅ | | server_startup_failfast_test | 1 | | @@ -87,4 +87,4 @@ `notification_webhook_test` also has 1 ignored store-and-forward regression tracked by rustfs#4852; ignored tests are excluded from the active counts above. -**Total listed: 485 tests across 67 modules · PR smoke subset: 129 tests / 32 modules** (30 full modules + 5 `reliant` tests + 20 of `replication_extension_test`) **· nightly `e2e-repl-nightly`: 27 tests** · generated 2026-07-24. +**Total listed: 486 tests across 67 modules · PR smoke subset: 129 tests / 32 modules** (30 full modules + 5 `reliant` tests + 20 of `replication_extension_test`) **· nightly `e2e-repl-nightly`: 28 tests** · generated 2026-07-26. diff --git a/rustfs/src/admin/handlers/service_account.rs b/rustfs/src/admin/handlers/service_account.rs index 7919c5c66..63a11dc61 100644 --- a/rustfs/src/admin/handlers/service_account.rs +++ b/rustfs/src/admin/handlers/service_account.rs @@ -652,6 +652,7 @@ impl Operation for UpdateServiceAccount { description: new_description.clone(), expiration: new_expiration, session_policy: sp, + parent_user: None, allow_site_replicator_account: false, }; diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 0c7e7489b..c94ca6f75 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -149,6 +149,7 @@ const SITE_REPLICATION_PEER_TLS_CAPABILITY_PATH: &str = const SITE_REPLICATION_PEER_EDIT_REFRESH_PATH: &str = "/rustfs/admin/v3/site-replication/peer/edit?refresh-targets=true"; const SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH: &str = "internal:endpoint-target-refresh"; const SITE_REPLICATION_PEER_REMOVE_PATH: &str = "/rustfs/admin/v3/site-replication/peer/remove"; +const SITE_REPLICATION_DEVNULL_PATH: &str = "/rustfs/admin/v3/site-replication/devnull"; const RUSTFS_ADMIN_V3_PREFIX: &str = "/rustfs/admin/v3"; const MINIO_ADMIN_V3_PREFIX: &str = "/minio/admin/v3"; const MINIO_SITE_REPLICATION_JOIN_PATH: &str = "/minio/admin/v3/site-replication/join"; @@ -369,6 +370,17 @@ impl SiteReplicationLifecycleGuard { _guard: SITE_REPLICATION_LIFECYCLE_LOCK.lock().await, } } + + /// Non-blocking variant for background work that must never interleave with an + /// add/remove/endpoint-refresh: those run in phases, and rebuilding rules between two of + /// them would resurrect exactly what the operation just tore down. Skipping a round is + /// free — the next tick picks it up. + fn try_acquire() -> Option { + SITE_REPLICATION_LIFECYCLE_LOCK + .try_lock() + .ok() + .map(|guard| Self { _guard: guard }) + } } impl SiteReplicationAddInProgressGuard { @@ -779,6 +791,11 @@ impl SRStatusOptions { } pub fn register_site_replication_route(r: &mut S3Router) -> std::io::Result<()> { + // Hand the reconciler to the infra-layer scheduler here rather than letting startup call + // into this module: startup sits below this layer and must not depend upwards. The admin + // router is built before startup reconciles, so the hook is always installed in time. + crate::site_replication_reconcile::register_site_replication_reconciler(reconcile_site_replication_wiring); + for (method, path, operation) in [ (Method::PUT, "/v3/site-replication/add", AdminOperation(&SiteReplicationAddHandler {})), ( @@ -2468,6 +2485,7 @@ async fn set_site_replicator_service_account_secret(parent_user: &str, secret_ke description: None, expiration: None, status: None, + parent_user: None, allow_site_replicator_account: true, }, ) @@ -2518,6 +2536,347 @@ async fn ensure_site_replicator_service_account(parent_user: &str, rotate_secret Ok((access_key, secret_key)) } +/// Whether a bucket target is one this site's own peer topology produced. +/// +/// Bucket targets are writable by anyone holding `admin:SetBucketTarget`, so a target that +/// merely carries the `site-replicator-0` access key proves nothing: an attacker with only +/// that permission could plant a secret of their choosing and have reconciliation recreate +/// the broadly privileged replication account with it. Require the target to name a peer in +/// the persisted state *and* to point at that peer's recorded endpoint, which is state only +/// `replicate add`/`edit` can write. +fn bucket_target_matches_configured_peer(target: &BucketTarget, state: &SiteReplicationState) -> bool { + let Some(deployment_id) = bucket_target_deployment_id(target) else { + return false; + }; + state + .peers + .get(&deployment_id) + .is_some_and(|peer| bucket_target_endpoint(target) == canonical_endpoint(&peer.endpoint)) +} + +/// Recover the shared site-replication secret from a bucket target belonging to a configured +/// peer. +/// +/// Every site-replication bucket target stores the `site-replicator-0` credentials in +/// `BUCKET_TARGETS_FILE`, which is not encrypted with the root credentials. That makes it the +/// one local copy that survives a root-credential change, so it can reseed the IAM account +/// when the IAM record itself became unreadable. +/// +/// Returns `None` unless every matching target agrees on the secret: disagreement means at +/// least one was written by something other than this site's own reconciliation, and picking +/// either would be a guess. +async fn site_replicator_secret_from_bucket_targets(access_key: &str, state: &SiteReplicationState) -> Option { + let store = current_object_store_handle()?; + let buckets = store.list_bucket(&BucketOptions::default()).await.ok()?; + let mut recovered: Option = None; + + for bucket in buckets { + let Ok(targets) = metadata_sys::list_bucket_targets(&bucket.name).await else { + continue; + }; + for target in targets.targets { + if target.target_type != BucketTargetType::ReplicationService + || !bucket_target_matches_configured_peer(&target, state) + { + continue; + } + let Some(credentials) = target.credentials.as_ref() else { + continue; + }; + if credentials.access_key != access_key || credentials.secret_key.is_empty() { + continue; + } + match &recovered { + Some(seen) if seen != &credentials.secret_key => return None, + Some(_) => {} + None => recovered = Some(credentials.secret_key.clone()), + } + } + } + + recovered +} + +/// Whether the parent recorded in site-replication state can actually back a service account. +/// +/// Repairing against a parent that does not exist would produce an account no policy path can +/// resolve, so a missing parent means "leave the current binding alone and report it". +async fn site_replicator_parent_is_usable(parent: &str) -> bool { + if rustfs_iam::is_root_access_key(parent) { + return true; + } + match current_iam_handle() { + Some(iam_sys) => iam_sys.get_user_info(parent).await.is_ok(), + None => false, + } +} + +/// Whether an IAM lookup failure means the account is absent or unreadable, as opposed to a +/// transient store failure. Only the former may trigger a reseed from bucket targets. +fn is_missing_service_account_error(err: &rustfs_iam::error::Error) -> bool { + matches!( + err, + rustfs_iam::error::Error::NoSuchAccount(_) + | rustfs_iam::error::Error::NoSuchServiceAccount(_) + | rustfs_iam::error::Error::NoSuchUser(_) + | rustfs_iam::error::Error::ConfigNotFound + ) +} + +/// Reconcile the local `site-replicator-0` account against the persisted site-replication +/// state, repairing the two drifts that no other code path can undo. +/// +/// `update_service_account` cannot rewrite `parent_user`, so once the account is bound to a +/// parent that a root-credential change invalidated, every later `replicate add` takes the +/// update branch and preserves the stale binding forever. Worse, IAM records encrypted with +/// the previous root secret fail to decrypt and surface as "no such account", which silently +/// disables every control-plane push while `replicate info` still reports the site enabled. +/// Both used to require deleting and recreating the account by hand. +async fn reconcile_site_replicator_service_account() -> S3Result<()> { + let _state_guard = SITE_REPLICATION_STATE_LOCK.lock().await; + let state = load_site_replication_state().await?; + if !state.enabled() || state.service_account_access_key != SITE_REPLICATOR_SERVICE_ACCOUNT { + return Ok(()); + } + + let Some(iam_sys) = current_iam_handle() else { + return Err(s3_error!(InvalidRequest, "iam not init")); + }; + + let parent_user = state.service_account_parent.clone(); + if parent_user.is_empty() { + warn!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + result = "service_account_parent_unknown", + "admin site replication state" + ); + return Ok(()); + } + + let access_key = SITE_REPLICATOR_SERVICE_ACCOUNT; + let session_policy = site_replicator_service_account_policy()?; + let reason = match iam_sys.get_site_replicator_service_account_secret(access_key).await { + Ok(_) => { + // Never rebuild on an unread parent: `unwrap_or_default` here would compare an + // empty string against a real parent and repair a healthy account on every boot. + let Ok((credentials, _)) = iam_sys.get_service_account(access_key).await else { + return Ok(()); + }; + if credentials.parent_user == parent_user { + return Ok(()); + } + + if !site_replicator_parent_is_usable(&parent_user).await { + warn!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + result = "service_account_parent_missing", + reason = "stale_parent", + parent = %parent_user, + "admin site replication state" + ); + return Ok(()); + } + + // The record is intact and may be authenticating replication traffic right now, + // so rebind it in place. Deleting first would open a window — however brief — + // where a crash or a storage error leaves the site with no replication account + // at all, which is worse than the stale binding being repaired. + iam_sys + .update_service_account( + access_key, + UpdateServiceAccountOpts { + session_policy: Some(session_policy), + secret_key: None, + name: None, + description: None, + expiration: None, + status: None, + parent_user: Some(parent_user.clone()), + allow_site_replicator_account: true, + }, + ) + .await + .map_err(ApiError::from)?; + "stale_parent" + } + // Only a genuinely absent or unreadable account may be reseeded from a bucket + // target. A transient store error must not trigger a rewrite of a live account. + Err(err) if is_missing_service_account_error(&err) => { + let Some(secret) = site_replicator_secret_from_bucket_targets(access_key, &state).await else { + warn!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + result = "service_account_unrecoverable", + error = ?err, + "admin site replication state" + ); + return Ok(()); + }; + + if !site_replicator_parent_is_usable(&parent_user).await { + warn!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + result = "service_account_parent_missing", + reason = "account_unreadable", + parent = %parent_user, + "admin site replication state" + ); + return Ok(()); + } + + // Nothing readable to preserve, so creation is the whole repair — there is no + // delete to leave a gap behind. + iam_sys + .new_service_account( + &parent_user, + None, + NewServiceAccountOpts { + session_policy: Some(session_policy), + access_key: access_key.to_string(), + secret_key: secret, + name: None, + description: None, + expiration: None, + allow_site_replicator_account: true, + claims: None, + }, + ) + .await + .map_err(ApiError::from)?; + "account_unreadable" + } + Err(err) => return Err(ApiError::from(err).into()), + }; + + warn!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + result = "service_account_repaired", + reason, + parent = %parent_user, + "admin site replication state" + ); + + Ok(()) +} + +/// Rebuild every replicated bucket's outbound rules and targets from the current peer set. +/// +/// A bucket whose `site-repl-*` rule was overwritten by a peer's config points at this very +/// deployment and replicates nothing. Nothing else revisits an existing bucket — the rule +/// builders only run on bucket creation, peer bucket-ops and metadata pushes — so without a +/// pass here an upgraded site keeps the broken rules until someone recreates the bucket. +/// Reconciliation is a no-op write-wise when the rules already match. +async fn reconcile_site_replication_buckets() -> S3Result<()> { + let Some(runtime) = runtime_site_replication_targets().await? else { + return Ok(()); + }; + let Some(store) = current_object_store_handle() else { + return Ok(()); + }; + let buckets = store.list_bucket(&BucketOptions::default()).await.map_err(ApiError::from)?; + + for bucket in buckets { + if let Err(err) = ensure_site_replication_bucket_setup_with_runtime(&bucket.name, &runtime).await { + warn!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + bucket = %bucket.name, + result = "bucket_reconcile_failed", + error = ?err, + "admin site replication state" + ); + continue; + } + + // Once per bucket per pass, so an operator sees a rule that resolves to nothing + // without the replication hot path logging it for every object. Reconciliation + // cannot fix this case: the rules are right and the peer endpoint is not reachable. + if let Ok(metadata) = metadata_sys::get(&bucket.name).await + && !site_replication_targets_online(&bucket.name, &metadata.replication_config_xml).await + { + warn!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + bucket = %bucket.name, + result = "replication_target_offline", + "a site replication rule has no usable remote target; objects for it are not replicating" + ); + } + } + + Ok(()) +} + +/// Repair drifted site-replication wiring: the service account first, since the bucket pass +/// signs its targets with that account's secret. +/// +/// Registered into the infra-layer scheduler (`site_replication_reconcile`) rather than +/// called from it, so startup never has to reach up into this layer. +/// +/// Gives up the whole round rather than racing a multi-phase operation. Two mechanisms are +/// needed: the lifecycle lock covers add and remove, while an endpoint refresh commits +/// bucket targets and peer state in separate steps *without* holding that lock +/// (`SiteReplicationEditHandler`), so a tick landing between them would rewrite the targets +/// from the stale endpoint. The pending marker in the persisted state closes that window. +/// Skipping costs nothing — the timer comes back. +fn reconcile_site_replication_wiring() -> std::pin::Pin + Send>> { + Box::pin(async { + // The scheduler starts before IAM and the object store are guaranteed ready (IAM + // bootstrap may still be recovering), so an early tick returns quietly instead of + // logging a failure for every reconciler. + if current_iam_handle().is_none() || current_object_store_handle().is_none() { + return; + } + + let Some(_lifecycle) = SiteReplicationLifecycleGuard::try_acquire() else { + return; + }; + + match load_site_replication_state().await { + Ok(state) => { + if state.pending_endpoint_refresh.is_some() || state.pending_remove.is_some() || state.pending_rotation.is_some() + { + return; + } + } + // Unreadable state is reported by the reconcilers below; do not double-log here. + Err(_) => return, + } + + if let Err(err) = reconcile_site_replicator_service_account().await { + warn!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + result = "service_account_reconcile_failed", + error = ?err, + "admin site replication state" + ); + } + if let Err(err) = reconcile_site_replication_buckets().await { + warn!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + result = "bucket_reconcile_failed", + error = ?err, + "admin site replication state" + ); + } + }) +} + fn site_replication_peer_wire_path(path: &str) -> String { let (path_only, query) = path .split_once('?') @@ -3839,6 +4198,8 @@ async fn build_sr_info(state: &SiteReplicationState, local_peer: &PeerInfo) -> S entry.quota_config_updated_at = maybe_time(metadata.quota_config_updated_at); entry.expiry_lc_config_updated_at = maybe_time(metadata.lifecycle_config_updated_at); entry.cors_config_updated_at = maybe_time(metadata.cors_config_updated_at); + entry.replication_targets_online = + Some(site_replication_targets_online(&bucket.name, &metadata.replication_config_xml).await); } info.buckets.insert(bucket.name, entry); @@ -4134,7 +4495,7 @@ fn canonical_status_json(value: &Value) -> Value { } } -fn site_replication_rule_complete(rule: &ReplicationRule) -> bool { +fn site_replication_rule_complete(rule: &ReplicationRule, owner_deployment_id: &str) -> bool { let delete_marker_enabled = rule .delete_marker_replication .as_ref() @@ -4155,16 +4516,29 @@ fn site_replication_rule_complete(rule: &ReplicationRule) -> bool { replica_modifications.status == ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED) }); + // A rule whose destination ARN names the site that holds it can never replicate: + // `reconcile_site_replication_bucket_targets` skips the local peer, so no bucket + // target backs that ARN and every object is dropped. Two sites holding byte-identical + // configs used to satisfy this check while exactly one of them could push. + let points_at_remote_site = replication_target_arn_deployment_id(&rule.destination.bucket) + .is_some_and(|deployment_id| deployment_id != owner_deployment_id); + rule.id.as_deref().is_some_and(|id| id.starts_with("site-repl-")) && rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED) + && points_at_remote_site && delete_marker_enabled && delete_enabled && existing_object_enabled && replica_modifications_enabled } -fn site_replication_config_mismatch<'a>(values: impl Iterator>, total_sites: usize) -> (usize, bool) { - let values = values.flatten().collect::>(); +fn site_replication_config_mismatch<'a>( + values: impl Iterator)>, + total_sites: usize, +) -> (usize, bool) { + let values = values + .filter_map(|(deployment_id, value)| value.map(|value| (deployment_id, value))) + .collect::>(); let present = values.len(); if present == 0 { return (0, false); @@ -4174,15 +4548,20 @@ fn site_replication_config_mismatch<'a>(values: impl Iterator(&xml) - .is_ok_and(|config| config.rules.len() == expected_rules && config.rules.iter().all(site_replication_rule_complete)) + deserialize::(&xml).is_ok_and(|config| { + config.rules.len() == expected_rules + && config + .rules + .iter() + .all(|rule| site_replication_rule_complete(rule, deployment_id)) + }) }); (present, !replicated) @@ -4233,14 +4612,27 @@ fn merge_bucket_status_info(status: &mut SRStatusInfo, site_infos: &BTreeMap bool { + rule.id.as_deref().is_some_and(|id| id.starts_with("site-repl-")) +} + +/// Whether every `site-repl-*` rule on this bucket resolves to a live remote target. +/// +/// The rule set alone cannot answer this: a rule can be perfectly formed while the endpoint +/// recorded for its peer is one this site cannot reach, so `update_all_targets` never built +/// a client for it and `replicate_object` drops every object against that ARN. Reads the +/// already-resolved client map rather than rebuilding clients, so it stays cheap enough for +/// the status path. +async fn site_replication_targets_online(bucket: &str, replication_config_xml: &[u8]) -> bool { + let Ok(config) = deserialize::(replication_config_xml) else { + return true; + }; + + for rule in config.rules.iter().filter(|rule| is_site_replication_rule(rule)) { + if BucketTargetSys::get() + .get_remote_target_client_by_arn(bucket, &rule.destination.bucket) + .await + .is_none() + { + return false; + } + } + + true +} + +/// Merge a peer's replication config into the local one. +/// +/// `site-repl-*` rules encode the *sender's* outbound direction — their destination ARN +/// names the receiver — so applying a peer's rule set verbatim replaces the receiver's +/// reverse rule with one pointing at itself. No bucket target can satisfy that ARN +/// (`reconcile_site_replication_bucket_targets` skips the local peer), so the receiver +/// silently stops replicating back: the one-directional symptom. Only operator-authored +/// rules travel between sites; each site owns its own `site-repl-*` rules. +fn merge_incoming_replication_config( + incoming: Option, + local: Option, +) -> Option { + let incoming_role = incoming.as_ref().map(|config| config.role.clone()).unwrap_or_default(); + // Operator rules first, then the local site rules — the same order + // `ensure_site_replication_bucket_replication_config_with_runtime` produces, so its + // no-op check matches and the bucket metadata is written once per broadcast, not twice. + let mut rules: Vec = incoming + .into_iter() + .flat_map(|config| config.rules) + .filter(|rule| !is_site_replication_rule(rule)) + .collect(); + rules.extend( + local + .into_iter() + .flat_map(|config| config.rules) + .filter(is_site_replication_rule), + ); + + if rules.is_empty() { + return None; + } + + for (index, rule) in rules.iter_mut().enumerate() { + rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX)); + } + + // A site-replication ARN in `role` is the sender's, and `site_replication_target_arns_by_peer` + // reads it — carrying it over would pin the receiver's targets to the sender's identity. + let role = match replication_target_arn_deployment_id(&incoming_role) { + Some(_) => String::new(), + None => incoming_role, + }; + + Some(ReplicationConfiguration { role, rules }) +} + fn replication_rule_deployment_id(rule: &ReplicationRule) -> Option { if let Some(rule_id) = rule.id.as_deref() { if let Some(deployment_id) = rule_id.strip_prefix("site-repl-") @@ -6155,14 +6622,26 @@ fn build_site_replication_config( state: &SiteReplicationState, local_peer: &PeerInfo, service_account_secret_key: &str, + existing: Option<&ReplicationConfiguration>, ) -> S3Result> { + // Reuse the ARN already recorded for a peer so the rule keeps pointing at the same + // bucket target `reconcile_site_replication_bucket_targets` keys off (a MinIO-era + // `arn:minio:...` target would otherwise be orphaned by a freshly minted ARN). + let configured_arns = site_replication_target_arns_by_peer(existing); let mut rules = Vec::new(); for peer in state.peers.values() { if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) { continue; } - let Some(target) = site_replication_bucket_target_for_peer(bucket, state, peer, service_account_secret_key, None)? else { + let Some(target) = site_replication_bucket_target_for_peer( + bucket, + state, + peer, + service_account_secret_key, + configured_arns.get(&peer.deployment_id).cloned(), + )? + else { continue; }; rules.push(build_site_replication_rule( @@ -6194,6 +6673,8 @@ async fn ensure_site_replication_bucket_targets_with_runtime( Err(StorageError::ConfigNotFound) => BucketTargets::default(), Err(err) => return Err(ApiError::from(err).into()), }; + let existing_json = serde_json::to_vec(&existing) + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize bucket targets failed: {e}")))?; let updated = reconcile_site_replication_bucket_targets(existing, bucket, state, local_peer, config, service_account_secret_key)?; @@ -6203,6 +6684,12 @@ async fn ensure_site_replication_bucket_targets_with_runtime( let json_targets = serde_json::to_vec(&updated) .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize bucket targets failed: {e}")))?; + // Rewriting identical targets would churn bucket metadata and rebuild every remote S3 + // client — noticeable now that startup reconciles all buckets, not just the one bucket + // an operation touched. + if json_targets == existing_json { + return Ok(()); + } metadata_sys::update(bucket, BUCKET_TARGETS_FILE, json_targets) .await .map_err(ApiError::from)?; @@ -6241,54 +6728,48 @@ async fn ensure_site_replication_bucket_replication_config_with_runtime( local_peer: &PeerInfo, service_account_secret_key: &str, ) -> S3Result<()> { - // Fix 6: reconcile rather than early-returning when any config already exists. - // The old code bailed on Ok(_), so the second site joined a replicated bucket and - // ended up with NO rule pointing back to the first site. Objects on that site could - // never travel back, producing the "one-directional" replication symptom. - let Some(desired) = build_site_replication_config(bucket, state, local_peer, service_account_secret_key)? else { - return Ok(()); - }; - - // Load the existing rules (may be empty if never configured). - let mut existing_rules = match metadata_sys::get_replication_config(bucket).await { - Ok((existing, _)) => existing.rules, - Err(StorageError::ConfigNotFound) => Vec::new(), + let existing = match metadata_sys::get_replication_config(bucket).await { + Ok((existing, _)) => Some(existing), + Err(StorageError::ConfigNotFound) => None, Err(err) => return Err(ApiError::from(err).into()), }; - // Collect the IDs of existing site-repl-* rules so we don't duplicate them. - let existing_rule_ids: HashSet = existing_rules - .iter() - .filter_map(|r| r.id.as_deref()) - .filter(|id| id.starts_with("site-repl-")) - .map(String::from) - .collect(); + let Some(desired) = build_site_replication_config(bucket, state, local_peer, service_account_secret_key, existing.as_ref())? + else { + return Ok(()); + }; - let mut added = false; - for rule in desired.rules { - let rule_id = rule.id.as_deref().unwrap_or(""); - if !existing_rule_ids.contains(rule_id) { - existing_rules.push(rule); - added = true; - } + // `site-repl-*` rules are derived state owned by this site: rebuild them from the + // current peer set on every pass instead of preserving whatever is on disk. A rule + // left over from a removed peer — or one whose destination ARN names this very + // deployment, which no bucket target can ever satisfy — must not survive, otherwise + // objects are queued against an ARN that resolves to nothing. + let (existing_role, existing_rules) = existing + .map(|config| (config.role, config.rules)) + .unwrap_or_else(|| (String::new(), Vec::new())); + let mut rules: Vec = existing_rules + .iter() + .filter(|rule| !is_site_replication_rule(rule)) + .cloned() + .collect(); + rules.extend(desired.rules); + for (index, rule) in rules.iter_mut().enumerate() { + rule.priority = Some(i32::try_from(index + 1).unwrap_or(i32::MAX)); } - if !added { - // All desired rules are already present — nothing to write. + // Only a site-replication ARN in `role` is ours to drop — an operator-authored role is + // part of the bucket's S3-visible configuration, and repairing a reverse rule must not + // quietly rewrite it. Same rule as `merge_incoming_replication_config`. + let role = match replication_target_arn_deployment_id(&existing_role) { + Some(_) => String::new(), + None => existing_role.clone(), + }; + + if rules == existing_rules && role == existing_role { return Ok(()); } - // Re-assign contiguous priorities to avoid conflicts with any preserved rules. - for (i, rule) in existing_rules.iter_mut().enumerate() { - rule.priority = Some((i + 1) as i32); - } - - let config = ReplicationConfiguration { - role: String::new(), - rules: existing_rules, - }; - - let data = serialize(&config) + let data = serialize(&ReplicationConfiguration { role, rules }) .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize replication failed: {e}")))?; metadata_sys::update(bucket, BUCKET_REPLICATION_CONFIG, data) .await @@ -6421,6 +6902,52 @@ pub async fn site_replication_peer_deployment_id_for_endpoint(endpoint: &str) -> /// kick a resync toward every remote peer so pre-existing objects back-fill. Returns a list of /// human-readable per-bucket failure messages (empty on full success) so the caller can surface /// them to the operator instead of silently reporting success; a failure never aborts the caller. +/// Probe every remote peer from the joining site before reporting the join a success. +/// +/// A peer's endpoint is whatever that peer derived from the `Host` header of the admin +/// request that created the topology, so the initiator can record an address only it can +/// reach — a console-port rewrite, a NAT address, a LAN-only host. The initiator's own +/// probes all succeed in that case, and the reverse direction then fails silently forever +/// because nothing else pushes from here until an object is written. Report it in the add +/// response instead of rejecting the join: an operator may legitimately be opening the +/// return path afterwards. +async fn probe_reverse_peer_reachability(state: &SiteReplicationState, local_peer: &PeerInfo) -> SiteReplicationErrorSummary { + let mut errors = SiteReplicationErrorSummary::default(); + let secret_key = match site_replicator_service_account_secret(&state.service_account_access_key).await { + Ok(secret) => secret, + Err(err) => { + errors.push(format!("reverse reachability probe skipped: {err}")); + return errors; + } + }; + + for peer in state.peers.values() { + if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) { + continue; + } + let connection = match runtime_peer_connection(peer) { + Ok(connection) => connection, + Err(err) => { + errors.push(format!("{} is not reachable from this site: {err}", peer.endpoint)); + continue; + } + }; + if let Err(err) = send_peer_admin_request( + &connection, + SITE_REPLICATION_DEVNULL_PATH, + &state.service_account_access_key, + &secret_key, + &serde_json::json!({}), + ) + .await + { + errors.push(format!("{} is not reachable from this site: {err}", peer.endpoint)); + } + } + + errors +} + async fn backfill_existing_buckets_after_add( state: &SiteReplicationState, local_peer: &PeerInfo, @@ -7045,16 +7572,40 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> { } } - if item.r#type == "replication-config" { - item.replication_config - .as_ref() - .map(|raw| { - let data = decode_bucket_meta_wire_value(raw); - deserialize::(&data) - }) - .transpose() - .map_err(|e| s3_error!(InvalidRequest, "invalid replication config: {e}"))?; - } + // Nothing to write and nothing on disk to clear: the common case for a site joining a + // replicated bucket, where every incoming rule is the sender's own. Skipping the write + // avoids stamping an empty config over a bucket that never had one; + // `ensure_site_replication_bucket_setup` below still installs this site's own rules. + let mut skip_config_write = false; + let merged_replication_config = + if item.r#type == "replication-config" { + let incoming = item + .replication_config + .as_ref() + .map(|raw| { + let data = decode_bucket_meta_wire_value(raw); + deserialize::(&data) + }) + .transpose() + .map_err(|e| s3_error!(InvalidRequest, "invalid replication config: {e}"))?; + let local = match metadata_sys::get_replication_config(&item.bucket).await { + Ok((config, _)) => Some(config), + Err(StorageError::ConfigNotFound) => None, + Err(err) => return Err(ApiError::from(err).into()), + }; + let local_absent = local.is_none(); + match merge_incoming_replication_config(incoming, local) { + Some(config) => Some(serialize(&config).map_err(|e| { + S3Error::with_message(S3ErrorCode::InternalError, format!("serialize replication failed: {e}")) + })?), + None => { + skip_config_write = local_absent; + None + } + } + } else { + None + }; let data = match item.r#type.as_str() { "policy" => item @@ -7071,25 +7622,29 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> { "version-config" => decode_bucket_meta_wire_option(item.versioning), "object-lock-config" => decode_bucket_meta_wire_option(item.object_lock_config), "sse-config" => decode_bucket_meta_wire_option(item.sse_config), - "replication-config" => decode_bucket_meta_wire_option(item.replication_config), + "replication-config" => merged_replication_config, "lc-config" => decode_bucket_meta_wire_option(item.expiry_lc_config), "cors-config" => decode_bucket_meta_wire_option(item.cors), _ => unreachable!(), }; - if let Some(data) = data { - metadata_sys::update(&item.bucket, config_file, data) - .await - .map_err(ApiError::from)?; - } else { - metadata_sys::delete(&item.bucket, config_file) - .await - .map_err(ApiError::from)?; + if !skip_config_write { + if let Some(data) = data { + metadata_sys::update(&item.bucket, config_file, data) + .await + .map_err(ApiError::from)?; + } else { + metadata_sys::delete(&item.bucket, config_file) + .await + .map_err(ApiError::from)?; + } } drop(targets_guard); if item.r#type == "replication-config" { - ensure_site_replication_bucket_targets(&item.bucket).await?; + // Rebuild the local outbound rules too: a site that joined an already-replicated + // bucket receives this item before it has any `site-repl-*` rule of its own. + ensure_site_replication_bucket_setup(&item.bucket).await?; } if item.r#type == "version-config" @@ -7369,6 +7924,7 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> { secret_key: Some(create.secret_key), expiration: create.expiration, status: (!create.status.is_empty()).then_some(create.status), + parent_user: None, allow_site_replicator_account: create.access_key == SITE_REPLICATOR_SERVICE_ACCOUNT, }, ) @@ -7421,6 +7977,9 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> { description: (!update.description.is_empty()).then_some(update.description), expiration: update.expiration, status: (!update.status.is_empty()).then_some(update.status), + // Peers replicate credentials, never the local parent binding: + // each site resolves its own parent from its own IAM. + parent_user: None, allow_site_replicator_account, }, ) @@ -7890,6 +8449,7 @@ impl Operation for SRPeerJoinHandler { description: None, expiration: None, status: None, + parent_user: None, allow_site_replicator_account: join_req.svc_acct_access_key == SITE_REPLICATOR_SERVICE_ACCOUNT, }, ) @@ -7936,7 +8496,8 @@ impl Operation for SRPeerJoinHandler { // Fix 1 (receiving side): ensure the joining peer also sets up replication for any // buckets it already owns so the reverse direction works from the start. Per-bucket // failures are logged (BUG2) so a reverse-direction back-fill gap is observable. - let backfill_errors = backfill_existing_buckets_after_add(&state, &local_peer, bootstrap_token.as_deref()).await; + let mut backfill_errors = probe_reverse_peer_reachability(&state, &local_peer).await; + backfill_errors.extend(backfill_existing_buckets_after_add(&state, &local_peer, bootstrap_token.as_deref()).await); if !backfill_errors.is_empty() { warn!( event = EVENT_ADMIN_SITE_REPLICATION_STATE, @@ -10627,13 +11188,118 @@ mod tests { let site_b_xml = String::from_utf8(serialize(&site_b_config).expect("site replication XML should serialize")) .expect("site replication XML should be UTF-8"); - assert!(site_replication_rule_complete(&site_a_config.rules[0])); + assert!(site_replication_rule_complete(&site_a_config.rules[0], "site-a")); assert_eq!( - site_replication_config_mismatch(vec![Some(&site_a_xml), Some(&site_b_xml)].into_iter(), 2), + site_replication_config_mismatch(vec![("site-a", Some(&site_a_xml)), ("site-b", Some(&site_b_xml))].into_iter(), 2), (2, false) ); } + // A site whose rules are well-formed but whose peer endpoint it cannot reach builds no + // target client, so it replicates nothing while its rule set still reads as correct. + // Rule-shape checking alone cannot see that, so the reporting site says so directly. + #[test] + fn test_merge_bucket_status_reports_offline_targets_as_mismatch() { + // Both sites carry a correct, peer-specific rule set: rule-shape checking alone + // sees a healthy pair. Only the reported target health distinguishes them. + let site_info = |peer: &str, targets_online: Option| { + let xml = String::from_utf8(serialize(&site_repl_config(peer)).unwrap()).unwrap(); + let mut info = SRInfo::default(); + info.buckets.insert( + "photos".to_string(), + SRBucketInfo { + bucket: "photos".to_string(), + replication_config: Some(xml), + replication_targets_online: targets_online, + ..Default::default() + }, + ); + info + }; + + let mut status = SRStatusInfo::default(); + let site_infos = BTreeMap::from([ + ("site-a".to_string(), site_info("site-b", Some(true))), + ("site-b".to_string(), site_info("site-a", Some(false))), + ]); + merge_bucket_status_info(&mut status, &site_infos, &SRStatusOptions::default()); + + let summary = status + .bucket_stats + .get("photos") + .and_then(|per_site| per_site.get("site-a")) + .expect("bucket stats should carry a per-site summary"); + assert!( + summary.replication_cfg_mismatch, + "a peer reporting an offline replication target must not read as in sync" + ); + } + + // A peer that predates the field reports nothing; that is unknown, not a fault, and must + // not flip every bucket to out-of-sync during a mixed-version upgrade. + #[test] + fn test_merge_bucket_status_treats_absent_target_health_as_unknown() { + let site_info = |peer: &str| { + let xml = String::from_utf8(serialize(&site_repl_config(peer)).unwrap()).unwrap(); + let mut info = SRInfo::default(); + info.buckets.insert( + "photos".to_string(), + SRBucketInfo { + bucket: "photos".to_string(), + replication_config: Some(xml), + replication_targets_online: None, + ..Default::default() + }, + ); + info + }; + + let mut status = SRStatusInfo::default(); + let site_infos = BTreeMap::from([ + ("site-a".to_string(), site_info("site-b")), + ("site-b".to_string(), site_info("site-a")), + ]); + merge_bucket_status_info(&mut status, &site_infos, &SRStatusOptions::default()); + + let summary = status + .bucket_stats + .get("photos") + .and_then(|per_site| per_site.get("site-a")) + .expect("bucket stats should carry a per-site summary"); + assert!( + !summary.replication_cfg_mismatch, + "peers that do not report target health must not be treated as broken" + ); + } + + // The one-directional regression: a `replication-config` broadcast overwrote the receiver's + // rules with the sender's, leaving both sites holding byte-identical XML whose destination + // ARN names the receiver. Only one site could push, yet the status check counted rules and + // reported "in sync" — the operator's single health signal agreed with the broken state. + #[test] + fn test_site_replication_config_mismatch_rejects_rule_pointing_at_owning_site() { + let shared_config = ReplicationConfiguration { + role: String::new(), + rules: vec![build_site_replication_rule( + "arn:rustfs:replication::site-b:test-replication", + 1, + "site-repl-site-b", + )], + }; + let shared_xml = String::from_utf8(serialize(&shared_config).expect("site replication XML should serialize")) + .expect("site replication XML should be UTF-8"); + + assert!( + !site_replication_rule_complete(&shared_config.rules[0], "site-b"), + "a rule whose destination ARN names its own site can never replicate" + ); + assert_eq!( + site_replication_config_mismatch(vec![("site-a", Some(&shared_xml)), ("site-b", Some(&shared_xml))].into_iter(), 2), + (2, true), + "identical configs mean site-b points at itself and cannot push" + ); + } + #[test] fn test_status_policy_compare_ignores_string_array_order() { let site_a_policy = serde_json::json!({ @@ -12330,6 +12996,219 @@ mod tests { assert_eq!(deployment_id.as_deref(), Some("remote-dep")); } + fn site_repl_config(peer: &str) -> ReplicationConfiguration { + ReplicationConfiguration { + role: String::new(), + rules: vec![build_site_replication_rule( + &format!("arn:rustfs:replication::{peer}:photos"), + 1, + &format!("site-repl-{peer}"), + )], + } + } + + fn replication_target(deployment_id: &str, endpoint: &str, secret: &str) -> BucketTarget { + BucketTarget { + source_bucket: "photos".to_string(), + target_bucket: "photos".to_string(), + endpoint: endpoint.to_string(), + deployment_id: deployment_id.to_string(), + arn: format!("arn:rustfs:replication::{deployment_id}:photos"), + target_type: BucketTargetType::ReplicationService, + credentials: Some(crate::admin::storage_api::bucket::target::Credentials { + access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(), + secret_key: secret.to_string(), + session_token: None, + expiration: None, + }), + ..Default::default() + } + } + + fn state_with_peer(deployment_id: &str, endpoint: &str) -> SiteReplicationState { + let mut state = SiteReplicationState::default(); + state.peers.insert( + deployment_id.to_string(), + PeerInfo { + deployment_id: deployment_id.to_string(), + ..peer(deployment_id, endpoint) + }, + ); + state + } + + // Bucket targets are writable by anyone holding `admin:SetBucketTarget`. Recovering a + // secret from a target that merely carries the site-replicator access key would let such + // a principal choose the secret for the broadly privileged replication account. + #[test] + fn test_secret_recovery_rejects_target_outside_the_peer_topology() { + let state = state_with_peer("remote", "http://remote.example.com:9000"); + + assert!( + bucket_target_matches_configured_peer( + &replication_target("remote", "remote.example.com:9000", "shared-secret"), + &state + ), + "a target naming a configured peer at its recorded endpoint is ours" + ); + assert!( + !bucket_target_matches_configured_peer( + &replication_target("attacker", "attacker.example.com:9000", "planted-secret"), + &state + ), + "a target naming an unknown deployment must never seed the replication account" + ); + assert!( + !bucket_target_matches_configured_peer( + &replication_target("remote", "attacker.example.com:9000", "planted-secret"), + &state + ), + "a target reusing a peer id but pointing elsewhere must not seed the account" + ); + } + + // A transient store failure must not be read as "the account is gone" and trigger a + // reseed that overwrites a live account. + #[test] + fn test_only_missing_account_errors_allow_reseeding() { + use rustfs_iam::error::Error as IamError; + + assert!(is_missing_service_account_error(&IamError::NoSuchAccount("x".into()))); + assert!(is_missing_service_account_error(&IamError::NoSuchServiceAccount("x".into()))); + assert!(is_missing_service_account_error(&IamError::ConfigNotFound)); + assert!( + !is_missing_service_account_error(&IamError::IAMActionNotAllowed), + "a permission failure is not evidence that the account is absent" + ); + } + + fn operator_rule(id: &str) -> ReplicationRule { + ReplicationRule { + id: Some(id.to_string()), + ..build_site_replication_rule("arn:aws:s3:::backup", 1, id) + } + } + + // The one-directional bug: the joined site applied the initiator's replication config + // verbatim, so its own `site-repl-` rule was replaced by a rule pointing at + // itself. No bucket target backs that ARN, so every object was dropped without a log. + #[test] + fn test_merge_incoming_replication_config_keeps_local_reverse_rule() { + let merged = merge_incoming_replication_config(Some(site_repl_config("home")), Some(site_repl_config("office"))) + .expect("merge should keep the local rule"); + + assert_eq!(merged.rules.len(), 1); + assert_eq!(merged.rules[0].id.as_deref(), Some("site-repl-office")); + assert_eq!(merged.rules[0].destination.bucket, "arn:rustfs:replication::office:photos"); + } + + // A peer deleting its replication config must not delete the receiver's reverse rule + // either — the delete travels as `replication-config` with no payload. + #[test] + fn test_merge_incoming_replication_config_survives_peer_delete() { + let merged = merge_incoming_replication_config(None, Some(site_repl_config("office"))) + .expect("local site rules must survive a peer delete"); + + assert_eq!(merged.rules.len(), 1); + assert_eq!(merged.rules[0].id.as_deref(), Some("site-repl-office")); + } + + #[test] + fn test_merge_incoming_replication_config_replicates_operator_rules() { + let mut incoming = site_repl_config("home"); + incoming.rules.push(operator_rule("nightly-backup")); + incoming.role = "arn:rustfs:replication::home:photos".to_string(); + + let merged = merge_incoming_replication_config(Some(incoming), Some(site_repl_config("office"))) + .expect("merge should produce rules"); + + let ids: Vec<_> = merged.rules.iter().filter_map(|rule| rule.id.as_deref()).collect(); + assert_eq!(ids, vec!["nightly-backup", "site-repl-office"]); + assert_eq!(merged.rules[0].priority, Some(1)); + assert_eq!(merged.rules[1].priority, Some(2)); + assert!( + merged.role.is_empty(), + "a site-replication ARN in `role` belongs to the sender and must not be adopted" + ); + } + + #[test] + fn test_merge_incoming_replication_config_returns_none_when_nothing_remains() { + assert!(merge_incoming_replication_config(Some(site_repl_config("home")), None).is_none()); + } + + // `role` is part of the bucket's S3-visible configuration. Repairing a reverse rule must + // drop only a sender-owned site-replication ARN, never an operator's own role — the same + // rule the merge path applies, so both paths agree on what is ours to rewrite. + #[test] + fn test_replication_role_is_only_cleared_when_it_is_a_site_replication_arn() { + let operator_role = "arn:aws:iam::123456789012:role/replication"; + assert!( + replication_target_arn_deployment_id(operator_role).is_none(), + "an operator IAM role is not a site-replication ARN and must be preserved" + ); + assert_eq!( + replication_target_arn_deployment_id("arn:rustfs:replication::home:photos").as_deref(), + Some("home"), + "a site-replication ARN is sender-owned and gets cleared" + ); + + let mut incoming = site_repl_config("home"); + incoming.role = operator_role.to_string(); + let merged = merge_incoming_replication_config(Some(incoming), Some(site_repl_config("office"))) + .expect("merge should produce rules"); + assert_eq!(merged.role, operator_role, "operator role must survive the merge"); + } + + // Rules and targets are keyed off the same ARN. Minting a fresh one while + // `reconcile_site_replication_bucket_targets` preserves a MinIO-era `arn:minio:...` + // target would leave the rule pointing at an ARN no target satisfies. + #[test] + fn test_build_site_replication_config_reuses_configured_arn() { + let mut state = SiteReplicationState { + service_account_access_key: "site-replicator-0".to_string(), + ..Default::default() + }; + state.peers.insert( + "local".to_string(), + PeerInfo { + deployment_id: "local".to_string(), + ..peer("local", "https://local.example.com") + }, + ); + state.peers.insert( + "remote".to_string(), + PeerInfo { + deployment_id: "remote".to_string(), + ..peer("remote", "http://remote.example.com:9000") + }, + ); + let existing = ReplicationConfiguration { + role: String::new(), + rules: vec![build_site_replication_rule( + "arn:minio:replication::remote:photos", + 1, + "site-repl-remote", + )], + }; + + let config = build_site_replication_config( + "photos", + &state, + &PeerInfo { + deployment_id: "local".to_string(), + ..peer("local", "https://local.example.com") + }, + "runtime-iam-secret", + Some(&existing), + ) + .expect("build site replication config") + .expect("a remote peer yields one rule"); + + assert_eq!(config.rules.len(), 1); + assert_eq!(config.rules[0].destination.bucket, "arn:minio:replication::remote:photos"); + } + #[test] fn test_reconcile_site_replication_bucket_targets_upserts_remote_peer_targets() { let mut state = SiteReplicationState { @@ -13303,31 +14182,32 @@ mod tests { fn test_derive_sync_state_from_replication_completeness() { // A peer that is reachable and has complete replication rules for all other peers // should be Enable; one that is reachable but has an incomplete config should be Disable. - let repl_complete_xml = { + let site_config_xml = |peer: &str| { let config = ReplicationConfiguration { role: String::new(), rules: vec![build_site_replication_rule( - "arn:rustfs:replication::dep-b:bucket", + &format!("arn:rustfs:replication::{peer}:bucket"), 1, - "site-repl-dep-b", + &format!("site-repl-{peer}"), )], }; String::from_utf8(serialize(&config).unwrap()).unwrap() }; + let dep_a_xml = site_config_xml("dep-b"); + let dep_b_xml = site_config_xml("dep-a"); // Peer that has complete config for 2-site setup - assert!(site_replication_rule_complete(&build_site_replication_rule( - "arn:rustfs:replication::dep-b:bucket", - 1, - "site-repl-dep-b" - ))); + assert!(site_replication_rule_complete( + &build_site_replication_rule("arn:rustfs:replication::dep-b:bucket", 1, "site-repl-dep-b"), + "dep-a" + )); assert_eq!( - site_replication_config_mismatch(vec![Some(&repl_complete_xml), Some(&repl_complete_xml)].into_iter(), 2), + site_replication_config_mismatch(vec![("dep-a", Some(&dep_a_xml)), ("dep-b", Some(&dep_b_xml))].into_iter(), 2), (2, false), "complete rules on both sites → no mismatch" ); assert_eq!( - site_replication_config_mismatch(vec![Some(&repl_complete_xml)].into_iter(), 2), + site_replication_config_mismatch(vec![("dep-a", Some(&dep_a_xml)), ("dep-b", None)].into_iter(), 2), (1, true), "config only on one of two sites → mismatch" ); @@ -13339,34 +14219,37 @@ mod tests { // bucket as out-of-sync ("0/N Buckets in sync"). This test feeds the real base64 wire form. #[test] fn test_site_replication_config_mismatch_accepts_base64_wire_form() { - let xml = { + let site_config_xml = |peer: &str| { let config = ReplicationConfiguration { role: String::new(), rules: vec![build_site_replication_rule( - "arn:rustfs:replication::dep-b:bucket", + &format!("arn:rustfs:replication::{peer}:bucket"), 1, - "site-repl-dep-b", + &format!("site-repl-{peer}"), )], }; String::from_utf8(serialize(&config).unwrap()).unwrap() }; - let b64 = BASE64_STANDARD.encode(xml.as_bytes()); + let dep_a_xml = site_config_xml("dep-b"); + let dep_b_xml = site_config_xml("dep-a"); + let dep_a_b64 = BASE64_STANDARD.encode(dep_a_xml.as_bytes()); + let dep_b_b64 = BASE64_STANDARD.encode(dep_b_xml.as_bytes()); // Both sites present the complete config in base64 wire form → NOT a mismatch. assert_eq!( - site_replication_config_mismatch(vec![Some(&b64), Some(&b64)].into_iter(), 2), + site_replication_config_mismatch(vec![("dep-a", Some(&dep_a_b64)), ("dep-b", Some(&dep_b_b64))].into_iter(), 2), (2, false), "base64-encoded complete configs on both sites must not be reported as a mismatch" ); // The tolerant decode keeps plain-XML callers working too. assert_eq!( - site_replication_config_mismatch(vec![Some(&xml), Some(&xml)].into_iter(), 2), + site_replication_config_mismatch(vec![("dep-a", Some(&dep_a_xml)), ("dep-b", Some(&dep_b_xml))].into_iter(), 2), (2, false), "raw-XML wire form still parses via the base64 fallback" ); // A base64 config present on only one of two sites is still a mismatch. assert_eq!( - site_replication_config_mismatch(vec![Some(&b64)].into_iter(), 2), + site_replication_config_mismatch(vec![("dep-a", Some(&dep_a_b64)), ("dep-b", None)].into_iter(), 2), (1, true), "config present on only one site is a mismatch regardless of encoding" ); diff --git a/rustfs/src/admin/handlers/user.rs b/rustfs/src/admin/handlers/user.rs index 803e9e915..463cd02b8 100644 --- a/rustfs/src/admin/handlers/user.rs +++ b/rustfs/src/admin/handlers/user.rs @@ -1198,6 +1198,7 @@ impl Operation for ImportIam { description: None, expiration: None, status: Some(status), + parent_user: None, allow_site_replicator_account: false, }, ) diff --git a/rustfs/src/lib.rs b/rustfs/src/lib.rs index 59ebed379..3a9034113 100644 --- a/rustfs/src/lib.rs +++ b/rustfs/src/lib.rs @@ -90,6 +90,7 @@ pub mod protocols; pub mod runtime_capabilities; pub(crate) mod runtime_sources; pub mod server; +pub(crate) mod site_replication_reconcile; pub(crate) mod startup_audit; pub(crate) mod startup_auth; pub(crate) mod startup_background; diff --git a/rustfs/src/site_replication_reconcile.rs b/rustfs/src/site_replication_reconcile.rs new file mode 100644 index 000000000..e327e3669 --- /dev/null +++ b/rustfs/src/site_replication_reconcile.rs @@ -0,0 +1,81 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Startup and periodic reconciliation of site-replication wiring. +//! +//! The reconcilers themselves live in `admin::handlers::site_replication`, next to the +//! site-replication state they repair. Startup runs in this (infra) layer and must not +//! reach up into the admin (interface) layer, so the direction is inverted here: this +//! module owns the schedule and the contract, and the admin layer registers its +//! implementations into it from `register_site_replication_route`. That registration +//! happens while the admin router is built, which `init_startup_http_servers` does before +//! `init_startup_runtime_services` calls in — so the hooks are always installed by the +//! time startup reconciles. + +use std::future::Future; +use std::pin::Pin; +use std::sync::OnceLock; +use std::time::Duration; + +use tokio_util::sync::CancellationToken; +use tracing::warn; + +const RECONCILE_INTERVAL: Duration = Duration::from_secs(600); + +/// A reconciler reports its own failures; the outcome carries no value because neither +/// caller can act on one — a site that cannot repair its replication wiring still serves S3. +type ReconcileHook = fn() -> Pin + Send>>; + +static RECONCILER: OnceLock = OnceLock::new(); + +/// Install the admin layer's reconciler. Idempotent: a second call is ignored, which keeps +/// repeated router construction (tests, the embedded server) from panicking. +pub(crate) fn register_site_replication_reconciler(reconcile: ReconcileHook) { + let _ = RECONCILER.set(reconcile); +} + +/// Repair drifted site-replication wiring, immediately and then on a timer. +/// +/// The first pass runs inside the spawned task rather than on the caller's path: it walks +/// every bucket, so awaiting it would delay the rest of startup in proportion to bucket +/// count. It is also spawned unconditionally, including when IAM bootstrap was deferred — +/// a node that recovers IAM in the background would otherwise keep self-pointing rules +/// until someone restarted it. The reconciler itself returns early while IAM or the object +/// store are still unavailable, so an early tick is silent rather than noisy. +/// +/// Later passes are no-ops once the wiring matches — the bucket pass compares the serialized +/// targets and rule set before writing — so a healthy deployment reads and writes nothing. +pub(crate) fn spawn_site_replication_reconcile_task(ctx: CancellationToken) { + if RECONCILER.get().is_none() { + warn!("site replication reconciler is not registered; periodic reconcile disabled"); + return; + } + + tokio::spawn(async move { + let mut ticker = tokio::time::interval(RECONCILE_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + + loop { + tokio::select! { + _ = ctx.cancelled() => break, + // The first tick fires immediately, which is the startup repair pass. + _ = ticker.tick() => { + if let Some(reconcile) = RECONCILER.get() { + reconcile().await; + } + } + } + } + }); +} diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index 73988839b..7a587ddec 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::site_replication_reconcile::spawn_site_replication_reconcile_task; use crate::storage_api::startup::services::{ECStore, EndpointServerPools, ServerContextSlot}; use crate::{ config::Config, @@ -79,6 +80,11 @@ pub(crate) async fn init_startup_runtime_services( let buckets = init_bucket_metadata_runtime(store.clone(), ctx.clone()).await?; let iam_bootstrap = init_iam_runtime(store.clone(), ctx.clone(), readiness, state_manager, server_ctx).await?; + // Unconditional: deferred IAM recovers in the background and has no callback into this + // scheduler, so gating on the inline disposition would leave a recovered node with + // self-pointing replication rules until the next restart. The task waits for IAM and + // bucket metadata itself, and its first pass runs off this path. + spawn_site_replication_reconcile_task(ctx.clone()); init_auth_integrations().await?; init_notification_runtime(endpoint_pools, buckets).await?; let enable_scanner = init_background_service_runtime(store.clone()).await?;