diff --git a/.gitignore b/.gitignore index 60325653d..8a26b9e94 100644 --- a/.gitignore +++ b/.gitignore @@ -52,6 +52,7 @@ docs __pycache__/ !docs/ docs/* +!docs/README.md !docs/architecture/ !docs/architecture/** !docs/operations/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 40ec98e4e..9bac39307 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Replication + +- Object Lock replication PUTs now carry a required integrity header, fixing target rejection introduced by the plain-payload default ([#7097](https://github.com/rustfs/rustfs/pull/7097)). This changes the default outbound request for locked objects but adds no persisted format. +- Multipart source objects stay on the multipart transport even when their checksum record is a whole-object checksum, so objects above the single-PUT limit remain replicable ([#7047](https://github.com/rustfs/rustfs/pull/7047)). +- Targets that mint their own version IDs now use a per-target version ledger for tag, retention, legal-hold, and permanent-delete mutations; ambiguous pre-ledger matches fail with backoff instead of guessing ([#7368](https://github.com/rustfs/rustfs/pull/7368)). This adds dual-prefixed internal metadata keys that older readers ignore. +- Single-part source checksums are forwarded as `x-amz-checksum-*` headers instead of user metadata, so the replica preserves checksum responses ([#7313](https://github.com/rustfs/rustfs/pull/7313)). This changes the default outbound headers for checksummed objects. +- Site-replication outage recovery now uses a bounded 30-second retry drain plus the 600-second full reconciliation pass, persists destructive liabilities before local deletion, and fences replay settlement and peer edits ([#7148](https://github.com/rustfs/rustfs/pull/7148)). Persisted additions are optional and ignored by older readers. +- IAM snapshot/deletion replay, target-assigned delete-marker purges, timestamp ordering, and best-effort peer broadcast now close the control-plane gaps found by the R6 review ([#7195](https://github.com/rustfs/rustfs/pull/7195)). +- Upgrade and rollback: upgrade every node in one site consecutively and verify reconciliation before moving to the next site; do not intentionally run a site mixed-version. Target-version ledger keys are harmless on rollback, although old code cannot use their routing. Before rolling back past [#7307](https://github.com/rustfs/rustfs/pull/7307), drain or repair every pending version purge: older code can free a retained version's data directory before its remote purge is acknowledged. See `docs/operations/site-replication-operations.md`. + ### Security - **Presigned URLs honour only signed headers** (GHSA-g8w9-qw9q-fghr): a SigV4 presigned request that carries an `x-amz-*` request header not listed in `X-Amz-SignedHeaders` is now rejected with `403 AccessDenied` ("There were headers present in the request which were not signed"), matching AWS S3. Previously the holder of a presigned `PutObject` URL could add unsigned `x-amz-tagging`, `x-amz-storage-class`, `x-amz-website-redirect-location`, ACL, metadata, Object Lock or SSE headers and have them applied. Presigners that intend a property must set it before signing so the SDK lists the header in `SignedHeaders`; `x-amz-cf-id` (CloudFront) remains tolerated unsigned. Header-signed SigV4 and SigV2 requests are unchanged. diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 10a106975..48dfd8cf5 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -4235,6 +4235,16 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR Enabled {target_b_arn} + + matrix-and-tags + 135 + Enabled + and-tags/envprodtiergold + Disabled + Enabled + Enabled + {target_b_arn} + matrix-disabled 140 @@ -4289,6 +4299,7 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR "matrix-prefix", "matrix-tag", "matrix-disabled", + "matrix-and-tags", "matrix-priority-high", "Priority>200", "Disabled", @@ -4409,6 +4420,30 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR put_single_tag_current(&source_client, source_bucket, "tagged/no-match.txt", "route", "tagged").await?; assert_replication_key_absent(&target_client_b, target_bucket_b, "tagged/no-match.txt", Duration::from_secs(3)).await?; + // S3 and MinIO both read `And.Tags` as AND: an object carrying only one of + // the required tags is not admitted. Matching any single tag would push + // data to a destination the rule never selected (backlog#2366 P1-1), and + // the two-tag rule is the shape `mc replicate add --tags "k1=v1&k2=v2"` + // writes, so a single-tag rule passing is not evidence for this. + source_client + .put_object() + .bucket(source_bucket) + .key("and-tags/partial.txt") + .tagging("env=prod") + .body(ByteStream::from_static(b"one of two tags")) + .send() + .await?; + assert_replication_key_absent(&target_client_b, target_bucket_b, "and-tags/partial.txt", Duration::from_secs(3)).await?; + source_client + .put_object() + .bucket(source_bucket) + .key("and-tags/full.txt") + .tagging("env=prod&tier=gold") + .body(ByteStream::from_static(b"both tags")) + .send() + .await?; + wait_for_user_get_object(&target_client_b, target_bucket_b, "and-tags/full.txt").await?; + source_client .put_object() .bucket(source_bucket) diff --git a/crates/replication/src/config.rs b/crates/replication/src/config.rs index c7dba3508..16a258ff7 100644 --- a/crates/replication/src/config.rs +++ b/crates/replication/src/config.rs @@ -166,6 +166,24 @@ fn rule_replicates(rule: &ReplicationRule, obj: &ObjectOpts) -> bool { } } +fn replication_filter_tags_match(filter: &s3s::dto::ReplicationRuleFilter, object_tags: &HashMap) -> bool { + let tag_matches = |tag: &s3s::dto::Tag| match (&tag.key, &tag.value) { + (None, None) => true, + (Some(key), _) if key.is_empty() => true, + (Some(key), Some(value)) => object_tags.get(key) == Some(value), + _ => false, + }; + + filter + .and + .as_ref() + .and_then(|and| and.tags.as_deref()) + .into_iter() + .flatten() + .chain(filter.tag.iter()) + .all(tag_matches) +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ReplicationTargetValidationError { RoleWithMultipleDestinations, @@ -704,7 +722,7 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { if let Some(filter) = &rule.filter { let object_tags = ReplicationTagFilter::decode_tags_to_map(&obj.user_tags); - if filter.test_tags(&object_tags) { + if replication_filter_tags_match(filter, &object_tags) { rules.push(rule.clone()); } } else { @@ -1139,6 +1157,47 @@ mod tests { assert_eq!(validate_replication_config_structure(&structure_config(vec![rule])), Ok(())); } + #[test] + fn actionable_rules_require_every_and_tag_to_match() { + let mut rule = replication_rule("rule-1", "arn:target:a"); + rule.filter = Some(s3s::dto::ReplicationRuleFilter { + and: Some(s3s::dto::ReplicationRuleAndOperator { + prefix: None, + tags: Some(vec![ + s3s::dto::Tag { + key: Some("env".to_string()), + value: Some("prod".to_string()), + }, + s3s::dto::Tag { + key: Some("tier".to_string()), + value: Some("gold".to_string()), + }, + ]), + }), + ..Default::default() + }); + let config = structure_config(vec![rule]); + let object = |user_tags: &str| ObjectOpts { + name: "object".to_string(), + user_tags: user_tags.to_string(), + ..Default::default() + }; + + assert!(config.filter_target_arns(&object("env=prod")).is_empty()); + assert_eq!(config.filter_target_arns(&object("env=prod&tier=gold")), vec!["arn:target:a"]); + assert!(config.filter_target_arns(&object("")).is_empty()); + + let mut malformed = config; + malformed.rules[0].filter.as_mut().unwrap().and.as_mut().unwrap().tags = Some(vec![s3s::dto::Tag { + key: Some("env".to_string()), + value: None, + }]); + assert!( + malformed.filter_target_arns(&object("env=prod")).is_empty(), + "a malformed tag filter must fail closed" + ); + } + #[test] fn structure_validation_allows_tag_filter_when_delete_marker_replication_disabled() { let mut rule = replication_rule("rule-1", "arn:target:a"); diff --git a/crates/replication/src/stats.rs b/crates/replication/src/stats.rs index 9f606bd23..70a4964d1 100644 --- a/crates/replication/src/stats.rs +++ b/crates/replication/src/stats.rs @@ -580,6 +580,30 @@ impl FailStats { FailedMetric { count, size } } + /// Both rolling windows from one walk of the samples. `short` must be the + /// narrower window; the walk stops at `long`. Callers that need both (the + /// per-node site snapshot) would otherwise scan the deque twice while + /// holding the bucket-stats read lock, and the deque is only bounded by + /// the one-hour window - an unreachable target under load fills it. + pub fn recent_windows(&self, short: Duration, long: Duration) -> (FailedMetric, FailedMetric) { + let now = Instant::now(); + let mut short_metric = FailedMetric::default(); + let mut long_metric = FailedMetric::default(); + for sample in self.recent.iter().rev() { + let age = now.duration_since(sample.observed_at); + if age > long { + break; + } + if age <= short { + short_metric.count += 1; + short_metric.size += sample.size; + } + long_metric.count += 1; + long_metric.size += sample.size; + } + (short_metric, long_metric) + } + pub fn merge(&self, other: &FailStats) -> Self { Self { count: self.count.saturating_add(other.count), @@ -912,6 +936,26 @@ mod tests { assert_eq!(last_hour.size, 96); } + #[test] + fn fail_stats_recent_windows_matches_two_separate_scans() { + let mut stats = FailStats::default(); + stats.add_size(64, None::<&()>); + stats.add_size(32, None::<&()>); + + let (minute, hour) = stats.recent_windows(Duration::from_secs(60), Duration::from_secs(60 * 60)); + let expected_minute = stats.recent_since(Duration::from_secs(60)); + let expected_hour = stats.recent_since(Duration::from_secs(60 * 60)); + + assert_eq!((minute.count, minute.size), (expected_minute.count, expected_minute.size)); + assert_eq!((hour.count, hour.size), (expected_hour.count, expected_hour.size)); + assert_eq!(minute.count, 2); + assert_eq!(hour.size, 96); + + let empty = FailStats::default(); + let (minute, hour) = empty.recent_windows(Duration::from_secs(60), Duration::from_secs(60 * 60)); + assert_eq!((minute.count, minute.size, hour.count, hour.size), (0, 0, 0, 0)); + } + #[test] fn fail_stats_saturate_instead_of_wrapping() { let mut stats = FailStats { diff --git a/docs/README.md b/docs/README.md new file mode 100644 index 000000000..872b5b9be --- /dev/null +++ b/docs/README.md @@ -0,0 +1,23 @@ +# Documentation + +Use the focused indexes rather than treating this directory as an unordered +collection: + +- [Architecture knowledge base](architecture/README.md) +- [Testing references](testing/README.md) + +## Operations + +Operational runbooks live under [`operations/`](operations/). Replication +operators should start with: + +| Runbook | Use it for | +|---|---| +| [Site replication operations](operations/site-replication-operations.md) | Health fields, pending operations, outage recovery, re-pair admission, IAM/SSE boundaries, and upgrades. | +| [Replication target check](operations/replication-check.md) | Validating an S3 destination and version fidelity before enabling replication. | +| [Replication object size limits](operations/replication-object-size-limits.md) | Multipart routing, large-object limits, and retry characteristics. | +| [Replication outbound transport](operations/replication-outbound-transport.md) | Integrity headers, generic target behavior, and transport knobs. | + +Other runbooks remain grouped by filename in [`operations/`](operations/); +architecture pages link to the relevant runbook where a cross-boundary +procedure is required. diff --git a/docs/architecture/README.md b/docs/architecture/README.md index 9a39edc7b..c151711cf 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -60,6 +60,8 @@ Required headings and strings in these files are asserted by `scripts/check_arch | [minio-rustfs-router-compatibility.md](minio-rustfs-router-compatibility.md) | a client or `mc` call that works against MinIO fails against RustFS and you need to know whether the endpoint is missing, stubbed, or deliberately different | | [minio-file-format-compat.md](minio-file-format-compat.md) | deciding whether a MinIO drive set, bucket-metadata blob, or SSE object can be read or imported by a given RustFS build, or before touching a listed version anchor | -Operations runbooks live in [../operations/](../operations/) and testing references in [../testing/README.md](../testing/README.md). +Operations runbooks are registered in the [documentation operations index](../README.md#operations), and testing references live in [../testing/README.md](../testing/README.md). + +For replication operations, start with [site replication operations](../operations/site-replication-operations.md), [replication target check](../operations/replication-check.md), [replication object size limits](../operations/replication-object-size-limits.md), and [replication outbound transport](../operations/replication-outbound-transport.md). For per-node HTTP failure ratios and cached storage probe provenance, see [S3 write failure diagnostics](../operations/s3-write-failure-diagnostics.md). diff --git a/docs/architecture/s3-compatibility-matrix.md b/docs/architecture/s3-compatibility-matrix.md index d4dfdf5c7..f296003dd 100644 --- a/docs/architecture/s3-compatibility-matrix.md +++ b/docs/architecture/s3-compatibility-matrix.md @@ -38,6 +38,42 @@ Counts ignore blank lines and comments; compute them from the files. The lifecyc "Supported" for the SSE row means RustFS encrypts and decrypts its own objects. MinIO SSE objects (SSE-S3, SSE-KMS, SSE-C) are not readable in default builds; see [minio-file-format-compat.md Part C](minio-file-format-compat.md#part-c--server-side-encryption-sse) for the `rio-v2` migration build. +## Replication Support Boundary + +Site replication and bucket replication are not the same compatibility claim. +Site replication requires RustFS-compatible peer admin APIs and coordinates +IAM, topology, buckets, and metadata. A generic S3-compatible service can only +be a bucket-replication data target. + +For a generic S3 target, RustFS supports object PUT/HEAD/DELETE, multipart +uploads, tags, version deletes, and Object Lock mutations when the target +implements the corresponding S3 APIs and has versioning enabled. Targets that +mint their own version IDs are supported through a per-target version ledger; +pre-ledger replicas are adopted only when exact key and ETag identify one +unambiguous target version. `NoSuchVersion` for an already absent addressed +replica is treated as converged. + +The following are capability boundaries, not universal S3 claims: + +- `GET /BUCKET?replication-check` must pass the phases required by the intended + workload. `VersionFidelity` may report a minting target as mismatched even + though ledger-addressed delete and Object Lock phases succeed. +- A target that rejects standard multipart constraints, required Object Lock + integrity headers, or the configured checksum framing is unsupported until + its transport settings are made compatible. +- SSE-S3 and SSE-KMS are decrypted at the source and re-encrypted by the + destination's KMS. SSE-C uses ciphertext passthrough and requires target + evidence. Unsupported or ambiguous encryption metadata fails closed. +- ACL authorization is intentionally unsupported, and generic targets never + receive RustFS IAM/site-control-plane state. +- RustFS does not guess between multiple target versions with the same key and + ETag. The mutation remains failed and retryable until repair establishes an + unambiguous mapping. + +See [site replication operations](../operations/site-replication-operations.md) +for health, recovery, and upgrade rules and [replication outbound transport](../operations/replication-outbound-transport.md) +for the tested target classes and knobs. + ## Not Yet Passing Standard S3 areas that must not be described as complete: diff --git a/docs/operations/site-replication-operations.md b/docs/operations/site-replication-operations.md new file mode 100644 index 000000000..556ef1a51 --- /dev/null +++ b/docs/operations/site-replication-operations.md @@ -0,0 +1,258 @@ +# Site Replication Operations + +**Use this when:** operating a site-replication deployment, diagnosing a peer +outage or incomplete topology change, pairing sites that already contain data, +or planning an upgrade. + +**Source of truth:** `rustfs/src/admin/handlers/site_replication.rs`, +`rustfs/src/site_replication/`, and the bucket-replication worker under +`crates/ecstore/src/bucket/replication/`. + +Site replication combines two different convergence paths: + +- the control plane replicates buckets, bucket metadata, IAM, and topology; +- ordinary bucket replication moves object versions and delete operations. + +An `enabled: true` response only says that a site has more than one configured +peer. It does not prove that every peer is reachable or caught up. Always read +`pendingOperation`, `retryStats`, `PeerErrors`, and `Metrics` as well. + +## Routine checks + +Run these commands from an admin workstation with one alias per site: + +```console +mc admin replicate info site-a +mc admin replicate status site-a +``` + +Check more than one site. A partition can leave each side with a different but +locally valid view. + +`replicate info` is the compact control-plane view: + +| Field | Interpretation | +|---|---| +| `enabled` | More than one site is configured; this is not a health verdict. | +| `sites` | The locally persisted topology. Compare deployment IDs and endpoints on every site. | +| `retryStats.pending` | Collapsed peer deliveries waiting to be retried. | +| `retryStats.failed` | Deliveries that crossed the escalation threshold and require attention. | +| `retryStats.lastError` | A redacted summary of the most recent delivery failure. | +| `pendingOperation` | A durable multi-step topology operation described below. Absence is the healthy steady state. | + +`replicate status` adds detailed convergence state: + +| Field | Interpretation | +|---|---| +| `Sites` / `PeerStates` | Configured peers and derived reachability/configuration state. | +| `PeerErrors` | A peer could not be queried. Its detailed counters may be absent; do not read zeros as success. | +| `BucketStats` | Per-bucket presence and versioning, replication, lifecycle, Object Lock, and metadata mismatches. | +| `PolicyStats`, `UserStats`, `GroupStats` | IAM inventory mismatches. | +| `RetryStats` | Durable control-plane retry backlog and escalation count. | +| `Metrics.replMetrics` | Per-destination online state, downtime, replicated counts/bytes, and `failed` totals/windows. | +| `Metrics.queued` / `Metrics.inProgress` | Object work waiting or active on the responding node. | +| `Metrics.errors` | Node-level object-replication failures. When only queue statistics are available, RustFS synthesizes a node entry and preserves this counter rather than reporting zero. | +| `Metrics.retries` | Redeliveries. Always zero today: a failed object is not retried by an event, it waits for the scanner pass described below. Read `errors` instead. | + +Healthy means: the same topology is visible on all sites, no pending operation, +no peer error, no failed retry escalation, required bucket/IAM state is in sync, +and queue/error counters are stable or falling. Counters are cumulative; alert on +their rate and on a backlog that does not drain, not merely on a non-zero total. + +## Pending operations and recovery + +`pendingOperation` contains `operation`, an opaque `id`, `pendingPeers`, and +`ackedPeers`. Do not edit the site-replication state object by hand. The marker +is the crash-recovery journal and removing it can make a partially applied +operation look complete. + +The heavyweight reconciler runs once at startup and every 600 seconds. The +lightweight retry drain runs every 30 seconds. A restart is therefore a valid +way to cause an immediate heavyweight pass after the underlying fault has been +fixed, but it is not a substitute for fixing connectivity, credentials, TLS, +or the remote endpoint. + +### `remove` + +The original topology and each peer acknowledgement are persisted before the +operation finalizes. While peers remain in `pendingPeers`, restore access to +them and wait for reconciliation. If a peer is permanently gone, a new remove +request may remove all currently active unacknowledged peers; RustFS permits +that request and then finalizes against the remaining topology. Removing the +local site or all sites is also an explicit completion path. + +Do not re-add a site merely to hide this marker. First compare the topology on +all reachable peers. If the same operation ID makes no progress for more than +one heavyweight interval, collect `PeerErrors`, `RetryStats`, and the +site-replication logs before retrying the remove. + +### `rotate-svc-acct` + +Service-account rotation keeps the candidate secrets and peer acknowledgements +until every current remote peer accepts the rotation. Restore the failing peer +and allow the reconciler to resume it. Do not manually delete either candidate +credential during this window: doing so can remove the only credential that a +not-yet-acknowledged peer accepts. + +After the marker clears, verify `replicate status` from every site, then retire +any separately retained old credential material according to local policy. + +### `endpoint-refresh` + +An endpoint, CA, or TLS-verification edit first refreshes the replication +target on every active peer and records acknowledgements. On startup and every +heavyweight pass, RustFS probes peer capability, uses the endpoint-refresh API +when supported (or the legacy peer-edit fallback), refreshes local bucket +targets, and commits the edit only after every still-active peer acknowledges. + +If this marker is stuck: + +1. Confirm that the proposed endpoint and CA are correct and reachable from + every site, not only from the admin workstation. +2. Restore the site-replication service account and TLS trust path. +3. Wait for one 600-second pass or restart one healthy node to trigger the + startup pass. +4. Re-run the identical edit only if the operation remains visible; a different + endpoint edit is rejected while the existing refresh is pending. The journal + pins the edit's payload, so a re-run without `--replicate-ilm-expiry` keeps + the value the first attempt recorded, and a re-run asking for a different + value is rejected. Finish or remove the pending refresh before changing it. + +A peer removed from the topology no longer blocks completion. A remove request +is accepted when it removes every active unacknowledged peer. + +While this marker is present, control-plane retry replay to the other peers +keeps running, but bucket wiring reconciliation waits: it rewrites the same +targets the refresh is changing. Expect bucket-level drift on this site to +persist until the refresh settles. + +## Outage recovery and convergence time + +Control-plane retry begins on the 30-second drain, while heavyweight snapshots, +pending topology operations, and bucket wiring are revisited on the 600-second +pass. Object MRF entries are persisted every 10 seconds by default and target +health is probed every 5 seconds. These are scheduling bounds, not delivery +SLAs: network timeouts and the amount of queued work add to them. + +Objects that must be rediscovered by the scanner have this conservative upper +bound before discovery: + +```text +RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES + × max(RUSTFS_SCANNER_CYCLE, actual duration of one scanner cycle) +``` + +The defaults re-descend a compacted directory every 16 cycles. A practical +production starting point for a tighter recovery objective is +`RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES=4`; `1` forces re-descent every cycle. +Measure the additional disk and metadata load before lowering it further or +tuning the scanner cadence. For an immediate operator-driven recovery, start a +site resync with `mc admin replicate resync start` and monitor its status. +Transfer time after discovery remains proportional to backlog size, bandwidth, +worker capacity, and target latency. Use queue depth and the rate of +`Metrics.errors` rather than the formula alone to decide whether convergence is +progressing. + +## Pairing sites that already contain data + +When more than one requested site is non-empty, preflight considers each bucket +name held by more than one site: + +- versioning must be `Enabled` on every site holding the shared bucket; +- Object Lock enablement must be identical on every holder. + +A bucket present on only one site is safe: post-add backfill creates it on the +other peers. A shared unversioned bucket is rejected because merging can +overwrite the only copy of an object. An Object Lock mismatch is rejected +because lock enablement cannot be changed after bucket creation and convergence +could otherwise strip a WORM guarantee. + +If preflight rejects the pair, keep the authoritative copy, delete the +conflicting bucket (or its contents) from all other sites, run `replicate add` +again, and then start `replicate resync` from the surviving site. Back up and +validate the authoritative data before deleting anything. + +## IAM convergence and repair boundary + +Ordinary IAM changes are delivered to each peer. A successful bulk IAM import +also schedules one collapsed full-IAM snapshot per remote peer. A failed IAM +deletion is replayed before that snapshot so the snapshot cannot re-create a +principal or grant that was already revoked. + +The safety state has two bounds: + +- deletion high-water marks are retained for 30 days; +- deletion replay bodies are capped at 256 distinct entities per peer. + +Repeated deletion of the same entity replaces its saved body. When the per-peer +cap is exceeded or the body cannot be serialized, the retry entry remains +escalated rather than pretending the deletion is replayable. An item from an +older sender without a source timestamp cannot install the 30-day high-water +mark, so verify it explicitly after a prolonged split. A successful drain +clears replay bodies; removing the peer prunes its bodies. For an escalated IAM +retry, use the site-replication repair workflow for the affected peer and IAM +family, then verify users, service accounts, groups, policies, and mappings on +both sides. Repair is the operator's explicit accountability transfer and +clears the saved deletion bodies only after the IAM repair succeeds. + +A group's status converges in one direction. An explicit disable is applied +everywhere, including through a snapshot, but a membership change never +carries an enable - it would otherwise re-enable a group frozen on the +receiving site. If a group ended up disabled on one site only, re-enable it +there explicitly with `mc admin group enable`; a snapshot or repair will not +do it. + +Treat IAM divergence as a security incident: a user deleted on one site can +remain usable on an unreachable peer until replay or repair completes. A peer +whose IAM entry is escalated does not receive scheduled snapshots either - +including the one a bulk import schedules - until the repair settles it. + +## Encrypted objects + +| Source form | Replication behavior | Fail-closed condition | +|---|---|---| +| SSE-S3 | The source decrypts the object; the request sends only `AES256` intent; the destination encrypts with its own KMS. Source envelope material never leaves the site. | The destination cannot satisfy the encryption request, or the source metadata is incomplete/unsupported. The replica is `FAILED`; plaintext is not silently stored. | +| SSE-KMS | The source decrypts the object; the request sends `aws:kms` intent without the source-local key ID; the destination selects its own configured KMS key. | Either side cannot decrypt/encrypt, or the metadata mixes incompatible encryption evidence. | +| SSE-C | Stored ciphertext and the required SSE-C replication transport metadata pass through. RustFS verifies target evidence before accepting the replica. | The target does not echo the customer-algorithm evidence, required material/layout is absent, or the metadata is ambiguous. | + +Unknown MinIO/RustFS encryption markers are never forwarded as ordinary user +metadata. They fail replication so an operator must migrate or repair the +object with a supported format. + +## Rolling upgrades and rollback + +Keep every node in one site on the same version whenever possible. Upgrade all +nodes of one site consecutively, verify its startup reconciliation and status, +then move to the next site. Do not intentionally leave a site mixed-version: +admin requests can land on different nodes, and an older node may not resume a +new pending-operation shape or expose its health fields. + +Current state additions are optional and defaulted, so older readers ignore +them. The target-version ledger is stored as dual-prefixed internal object +metadata and is also ignored by older readers; rollback does not corrupt the +object format, but older code loses the assigned-version routing improvement. + +Before rolling back across the fix that retains the data directory of a version +awaiting purge replication (rustfs/rustfs#7307), ensure no version purge is +pending. Older code can free that retained version's data directory before the +remote purge is acknowledged, leaving unreadable metadata and blocking bucket +deletion. Drain or repair replication and take a metadata/data backup first. + +## Runtime knobs + +These values are read when the owning background task starts. Restart the +server after changing them. The millisecond intervals have a 10 ms floor; +invalid values fall back to the default with a warning. + +| Variable | Default | Effect | +|---|---:|---| +| `RUSTFS_REPL_HEALTH_CHECK_INTERVAL_MS` | `5000` | Remote-target health probe interval. Lowering it increases outbound probes. | +| `RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS` | `10000` | Maximum periodic interval between MRF persistence flushes; 1,000 new entries also trigger a flush. | +| `RUSTFS_REPL_RESYNC_POLL_MAX_MS` | `60000` | Upper bound for randomized resync retry-poll sleep. | +| `RUSTFS_REPL_RESYNC_MAX_JOBS` | `2` | Concurrent resync jobs; values are bounded to `1..=32`. | + +Transport-specific controls and target behavior are documented in +[Replication outbound transport](replication-outbound-transport.md). Validate a +new destination with [Replication target check](replication-check.md), and read +[Replication object size limits](replication-object-size-limits.md) before +moving large objects. diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index e4ca6ec9d..0179074c4 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -627,6 +627,32 @@ pub(crate) async fn cluster_replication_stats(bucket: &str, context: Option>, action: &'static str) { + let Some(notification_system) = current_notification_system_for_context(context.as_deref()) else { + return; + }; + if let Err(err) = notification_system.load_bucket_metadata(bucket).await { + warn!( + event = EVENT_ADMIN_REMOTE_TARGET_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_REPLICATION, + action = action, + result = "peer_metadata_reload_failed", + bucket = %bucket, + error = ?err, + "admin remote target state" + ); + } +} + fn unique_replication_peers(peer_clients: &[Option]) -> (Vec<&PeerRestClient>, u32) { let mut seen_grid_hosts = HashSet::new(); let peers: Vec<_> = peer_clients @@ -699,6 +725,7 @@ pub struct SetRemoteTargetHandler {} impl Operation for SetRemoteTargetHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { let cred = validate_replication_admin_request(&req, AdminAction::SetBucketTargetAction).await?; + let app_context = app_context_from_req(&req); let queries = extract_query_params(&req.uri); @@ -926,6 +953,8 @@ impl Operation for SetRemoteTargetHandler { .map_err(map_bucket_target_error)?; let _targets_guard = lock_bucket_targets_metadata(bucket).await; let arn = persist_remote_target_write(bucket, remote_target, incarnation, mode).await?; + drop(_targets_guard); + notify_remote_target_metadata_reload(bucket, app_context, "set_remote_target").await; let arn_str = serde_json::to_string(&arn) .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "Failed to serialize target ARN"))?; @@ -1006,6 +1035,7 @@ pub struct RemoveRemoteTargetHandler {} impl Operation for RemoveRemoteTargetHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { validate_replication_admin_request(&req, AdminAction::SetBucketTargetAction).await?; + let app_context = app_context_from_req(&req); debug!("remove remote target called"); let queries = extract_query_params(&req.uri); @@ -1081,6 +1111,7 @@ impl Operation for RemoveRemoteTargetHandler { } let json_targets = serde_json::to_vec(&targets) .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "Failed to serialize targets"))?; + let notification_bucket = bucket.clone(); let bucket = bucket.clone(); let arn = arn_str.clone(); // The pool cancellation owns a detached task. Both outer guards must @@ -1101,6 +1132,8 @@ impl Operation for RemoveRemoteTargetHandler { S3Error::with_message(S3ErrorCode::InternalError, format!("remote target removal task failed: {error}")) })??; + notify_remote_target_metadata_reload(¬ification_bucket, app_context, "remove_remote_target").await; + Ok(S3Response::new((StatusCode::NO_CONTENT, Body::from("".to_string())))) } } @@ -1787,6 +1820,25 @@ mod tests { pairs.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect() } + #[test] + fn remote_target_writes_notify_peer_metadata_caches() { + let source = include_str!("replication.rs"); + for (start, end) in [ + ("impl Operation for SetRemoteTargetHandler", "pub struct ListRemoteTargetHandler"), + ("impl Operation for RemoveRemoteTargetHandler", "async fn cancel_active_resync_intent"), + ] { + let body = source + .split(start) + .nth(1) + .and_then(|rest| rest.split(end).next()) + .expect(start); + assert!( + body.contains("notify_remote_target_metadata_reload"), + "{start} must notify every node before returning success" + ); + } + } + #[test] fn update_ops_parse_minio_query_contract() { let ops = parse_remote_target_update_ops(&query_map(&[ diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 318c8a093..ad810384f 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -62,12 +62,13 @@ use rustfs_iam::sys::{ }; use rustfs_madmin::{ BucketBandwidth, GroupStatus, IDPSettings, InProgressMetric, InQueueMetric, LDAPConfigSettings, LDAPSettings, - OpenIDProviderSettings, OpenIDSettings, PeerInfo, PeerSite, QStat, ReplProxyMetric, ReplicateAddStatus, ReplicateEditStatus, - 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, SRSiteSummary, SRStateEditReq, SRStateInfo, - SRStatusInfo, SRSvcAccChange, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat, + OpenIDProviderSettings, OpenIDSettings, PeerInfo, PeerSite, QStat, RStat, ReplProxyMetric, ReplicateAddStatus, + ReplicateEditStatus, 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, SRSiteSummary, SRStateEditReq, + SRStateInfo, SRStatusInfo, SRSvcAccChange, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, + TimedErrStats, WorkerStat, }; use rustfs_policy::policy::{ Policy, @@ -87,7 +88,7 @@ use std::sync::{LazyLock, Mutex as StdMutex}; use std::time::Duration; use time::OffsetDateTime; use tokio::sync::Mutex; -use tracing::{info, warn}; +use tracing::{debug, info, warn}; use url::Url; use url::form_urlencoded; use uuid::Uuid; @@ -1866,7 +1867,17 @@ fn reconcile_site_replication_wiring() -> std::pin::Pin { - if state.pending_endpoint_refresh.is_some() { + if let Some(pending_endpoint_refresh) = state.pending_endpoint_refresh.clone() { + resume_pending_endpoint_refresh(&state, &pending_endpoint_refresh).await; + // The bucket reconciler below rewrites targets from the + // topology, which is exactly what this refresh is in the + // middle of changing, so it still waits for the next pass. + // The retry queue does not: it replays per-peer deliveries + // against the endpoints currently committed in state, and a + // refresh that cannot finish - a peer that never comes back + // - must not also stall replay to the healthy peers. + drop(lifecycle); + drain_site_replication_retry_queue().await; return; } // A wedged rotation is worse than a wedged removal: the local @@ -2173,6 +2184,7 @@ fn peer_metric_entry( reachable: bool, (total_downtime_ns, last_online): (i64, Option), local_counters: (i64, i64), + local_failures: TimedErrStats, ) -> SRMetric { let (replica_size, replica_count) = local_counters; @@ -2187,6 +2199,25 @@ fn peer_metric_entry( // remote entries would double-count them cluster-wide. replicated_size: if is_local { replica_size } else { 0 }, replicated_count: if is_local { replica_count } else { 0 }, + failed: if is_local { local_failures } else { TimedErrStats::default() }, + ..Default::default() + } +} + +fn site_failure_stats(node: &crate::storage::storage_api::ReplicationSiteMetricsSnapshot) -> TimedErrStats { + TimedErrStats { + last_minute: RStat { + count: node.failed_last_minute_count as f64, + bytes: node.failed_last_minute_bytes, + }, + last_hour: RStat { + count: node.failed_last_hour_count as f64, + bytes: node.failed_last_hour_bytes, + }, + totals: RStat { + count: node.failed_count as f64, + bytes: node.failed_bytes, + }, ..Default::default() } } @@ -2201,6 +2232,7 @@ async fn build_metrics_summary( }; let node = stats.site_metrics_snapshot().await; + let failures = site_failure_stats(&node); let mut metrics = BTreeMap::new(); // Emit an entry for every peer, not just the local one. An operator reading @@ -2221,6 +2253,7 @@ async fn build_metrics_summary( reachable, health, (node.replica_size, node.replica_count), + failures.clone(), ), ); } @@ -2234,6 +2267,7 @@ async fn build_metrics_summary( last_online: Some(OffsetDateTime::now_utc()), replicated_size: node.replica_size, replicated_count: node.replica_count, + failed: failures, ..Default::default() }); @@ -3015,6 +3049,22 @@ fn peer_endpoint_refresh_requested(state: &SiteReplicationState, incoming: &Peer .is_some_and(|peer| !peer_connection_settings_match(peer, incoming)) } +/// A persisted refresh journal pins the edit's payload: the commit reads the +/// peer and the ilm-expiry override back out of it, so a re-run cannot change +/// them. Re-running without the flag keeps the pinned value - that is the +/// documented way to redrive a stuck refresh - but a re-run that asks for a +/// different value must be rejected rather than accepted and ignored. +/// The one construction of the concurrent-change error: every writer of a +/// pending refresh re-checks the journal inside its own transaction and +/// reports the same condition when it no longer owns it. +fn endpoint_refresh_state_changed() -> S3Error { + s3_error!(InvalidRequest, "endpoint target refresh state changed during update") +} + +fn endpoint_refresh_ilm_override_conflicts(persisted: &PendingEndpointRefresh, requested: Option) -> bool { + requested.is_some() && requested != persisted.ilm_expiry_override +} + fn merge_pending_endpoint_refresh( state: &SiteReplicationState, candidate: &PendingEndpointRefresh, @@ -3025,7 +3075,7 @@ fn merge_pending_endpoint_refresh( || latest.peer.deployment_id != candidate.peer.deployment_id || !peer_connection_settings_match(&latest.peer, &candidate.peer) { - return Err(s3_error!(InvalidRequest, "endpoint target refresh state changed during update")); + return Err(endpoint_refresh_state_changed()); } latest } else { @@ -3762,6 +3812,21 @@ fn pending_operation_for_state(state: &SiteReplicationState, local_peer: &PeerIn }); } + if let Some(pending) = pending_endpoint_refresh(state) { + let pending_peers = pending_endpoint_refresh_required_peer_ids(state, &pending, local_peer) + .into_iter() + .filter(|deployment_id| !pending.acked_deployment_ids.contains(deployment_id)) + .collect(); + return Some(SRPendingOperation { + operation: "endpoint-refresh".to_string(), + id: pending.id, + pending_peers, + acked_peers: pending.acked_deployment_ids.into_iter().collect(), + updated_at: None, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }); + } + state.pending_rotation.as_ref().map(|pending| { let pending_peers = pending_remote_peer_ids(&pending.peers, local_peer) .into_iter() @@ -3798,6 +3863,78 @@ fn pending_all_remote_peers_acked( .all(|deployment_id| acked_deployment_ids.contains(deployment_id)) } +fn pending_endpoint_refresh_required_peer_ids( + state: &SiteReplicationState, + pending: &PendingEndpointRefresh, + local_peer: &PeerInfo, +) -> BTreeSet { + pending + .remote_peers + .values() + .filter(|peer| { + state.peers.contains_key(&peer.deployment_id) + && peer.deployment_id != local_peer.deployment_id + && !same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) + }) + .map(|peer| peer.deployment_id.clone()) + .collect() +} + +fn pending_endpoint_refresh_is_complete( + state: &SiteReplicationState, + pending: &PendingEndpointRefresh, + local_peer: &PeerInfo, +) -> bool { + pending_endpoint_refresh_required_peer_ids(state, pending, local_peer) + .iter() + .all(|deployment_id| pending.acked_deployment_ids.contains(deployment_id)) +} + +fn pending_endpoint_refresh_allows_remove( + state: &SiteReplicationState, + pending: &PendingEndpointRefresh, + local_peer: &PeerInfo, + remove_req: &SRRemoveReq, +) -> bool { + if remove_req.remove_all || remove_req.site_names.iter().any(|name| name == &local_peer.name) { + return true; + } + let removed = removed_deployment_ids_for_remove_req(state, remove_req); + pending_endpoint_refresh_required_peer_ids(state, pending, local_peer) + .difference(&pending.acked_deployment_ids) + .all(|deployment_id| removed.contains(deployment_id)) +} + +fn discard_endpoint_refresh_if_target_was_removed(state: &mut SiteReplicationState) { + if pending_endpoint_refresh(state) + .as_ref() + .is_some_and(|pending| !state.peers.contains_key(&pending.peer.deployment_id)) + { + clear_pending_endpoint_refresh(state); + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum EndpointRefreshPostwriteState { + Current, + Superseded, + TargetRemoved, +} + +fn endpoint_refresh_postwrite_state( + state: &SiteReplicationState, + pending_id: &str, + target_deployment_id: &str, +) -> EndpointRefreshPostwriteState { + if !state.peers.contains_key(target_deployment_id) { + EndpointRefreshPostwriteState::TargetRemoved + } else if pending_endpoint_refresh(state).is_some_and(|pending| pending.id == pending_id) { + EndpointRefreshPostwriteState::Current + } else { + EndpointRefreshPostwriteState::Superseded + } +} + fn push_unique_secret_candidate(candidates: &mut Vec, secret: String) { if !secret.is_empty() && !candidates.iter().any(|candidate| candidate == &secret) { candidates.push(secret); @@ -4019,6 +4156,171 @@ async fn resume_pending_rotation(state: &SiteReplicationState, pending: &Pending } } +async fn mark_pending_endpoint_refresh_peer_acked(refresh_id: &str, deployment_id: &str) -> S3Result<()> { + let refresh_id = refresh_id.to_string(); + let deployment_id = deployment_id.to_string(); + update_site_replication_state_when_changed(move |state| { + let Some(pending) = state + .pending_endpoint_refresh + .as_mut() + .filter(|pending| pending.id == refresh_id) + else { + return Ok(StateCommit::Unchanged(())); + }; + pending.acked_deployment_ids.insert(deployment_id); + Ok(StateCommit::Changed(())) + }) + .await +} + +async fn finalize_pending_endpoint_refresh_if_complete(refresh_id: &str, service_account_secret_key: &str) -> S3Result { + let state = load_site_replication_state().await?; + let Some(pending) = pending_endpoint_refresh(&state).filter(|pending| pending.id == refresh_id) else { + return Ok(true); + }; + let local_peer = current_local_runtime_peer(&state); + if !pending_endpoint_refresh_is_complete(&state, &pending, &local_peer) { + return Ok(false); + } + + refresh_bucket_targets_after_endpoint_edit(refresh_id, service_account_secret_key).await?; + + let refresh_id = refresh_id.to_string(); + update_site_replication_state_when_changed(move |state| { + let Some(pending) = pending_endpoint_refresh(state).filter(|pending| pending.id == refresh_id) else { + return Ok(StateCommit::Unchanged(true)); + }; + let local_peer = current_local_runtime_peer(state); + if !pending_endpoint_refresh_is_complete(state, &pending, &local_peer) { + return Ok(StateCommit::Unchanged(false)); + } + *state = edit_state(std::mem::take(state), pending.peer, pending.ilm_expiry_override); + clear_pending_endpoint_refresh(state); + Ok(StateCommit::Changed(true)) + }) + .await +} + +async fn drive_pending_endpoint_refresh( + state: &SiteReplicationState, + pending: &PendingEndpointRefresh, +) -> S3Result<(Vec, bool)> { + if state.service_account_access_key.is_empty() { + return Err(s3_error!(InvalidRequest, "site replication service account is not configured")); + } + let service_account_secret_key = site_replicator_service_account_secret(&state.service_account_access_key).await?; + let local_peer = current_local_runtime_peer(state); + let required = pending_endpoint_refresh_required_peer_ids(state, pending, &local_peer); + let mut peer_errors = Vec::new(); + + for target in pending.remote_peers.values().filter(|target| { + required.contains(&target.deployment_id) && !pending.acked_deployment_ids.contains(&target.deployment_id) + }) { + let refreshed = async { + let (status, body) = send_endpoint_refresh_admin_request_raw( + target, + pending, + SITE_REPLICATION_PEER_EDIT_CAPABILITY_PATH, + &state.service_account_access_key, + &service_account_secret_key, + &(), + ) + .await?; + if endpoint_refresh_capability_supported(target, status, &body)? { + let request = EndpointRefreshRequest { + id: pending.id.clone(), + peer: pending.peer.clone(), + }; + let body = send_endpoint_refresh_admin_request( + target, + pending, + SITE_REPLICATION_PEER_EDIT_REFRESH_PATH, + &state.service_account_access_key, + &service_account_secret_key, + &request, + ) + .await?; + parse_endpoint_refresh_status(target, &body) + } else { + refresh_legacy_peer_bucket_targets( + target, + pending, + &state.service_account_access_key, + &service_account_secret_key, + ) + .await + } + } + .await; + + match refreshed { + Ok(()) => mark_pending_endpoint_refresh_peer_acked(&pending.id, &target.deployment_id).await?, + Err(err) => peer_errors.push(summarize_peer_error_detail(&format!("{}: {err}", target.endpoint))), + } + } + + let complete = finalize_pending_endpoint_refresh_if_complete(&pending.id, &service_account_secret_key).await?; + Ok((peer_errors, complete)) +} + +/// The coordinator of an endpoint edit snapshots the topology it must fan out +/// to into `remote_peers`; the peer-side handler persists its journal with an +/// empty map ([`SRPeerEditHandler`]) because it fans out to nobody. Only the +/// coordinator's journal may be resumed here. A receiver's journal has no +/// required peers, so it would read as complete on the first tick and commit +/// through [`edit_state`] instead of [`apply_internal_peer_edit`] — dropping +/// the local-name sync and racing the in-flight request that owns it, whose +/// own commit would then report the refresh as changed and leave the +/// coordinator waiting for an acknowledgement it will never get. A receiver's +/// journal is redriven by the coordinator resending the same refresh id. +fn pending_endpoint_refresh_is_locally_driven(pending: &PendingEndpointRefresh) -> bool { + !pending.remote_peers.is_empty() +} + +async fn resume_pending_endpoint_refresh(state: &SiteReplicationState, pending: &PendingEndpointRefresh) { + if !pending_endpoint_refresh_is_locally_driven(pending) { + debug!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + result = "pending_endpoint_refresh_owned_by_peer_request", + "admin site replication state" + ); + return; + } + match drive_pending_endpoint_refresh(state, pending).await { + Ok((peer_errors, true)) if peer_errors.is_empty() => { + info!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + result = "pending_endpoint_refresh_resumed", + "admin site replication state" + ); + } + Ok((peer_errors, _)) => { + warn!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + result = "pending_endpoint_refresh_still_pending", + error_count = peer_errors.len(), + "admin site replication state" + ); + } + Err(err) => { + warn!( + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + result = "pending_endpoint_refresh_resume_failed", + error = ?err, + "admin site replication state" + ); + } + } +} + async fn pending_remove_ready_to_finalize(remove_id: &str, local_peer: &PeerInfo) -> S3Result> { let state = load_site_replication_state().await?; let Some(pending) = state.pending_remove.as_ref() else { @@ -5004,6 +5306,11 @@ async fn refresh_bucket_targets_after_endpoint_edit(pending_id: &str, service_ac .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "object store is not initialized".to_string()))?; let buckets = store.list_bucket(&BucketOptions::default()).await.map_err(ApiError::from)?; + // Every bucket already rewritten in this pass carries the edited peer's + // target. A remove accepted mid-pass must undo all of them, not just the + // one in flight: the remove's own cleanup may have already walked past a + // bucket this loop wrote afterwards. + let mut rewritten = Vec::new(); for bucket in buckets { let expected_incarnation_id = metadata_sys::capture_bucket_metadata_incarnation(&bucket.name) .await @@ -5012,7 +5319,7 @@ async fn refresh_bucket_targets_after_endpoint_edit(pending_id: &str, service_ac // every round, and the writes below are bucket metadata, not state. let state = load_site_replication_state().await?; let Some(pending) = pending_endpoint_refresh(&state).filter(|pending| pending.id == pending_id) else { - return Err(s3_error!(InvalidRequest, "endpoint target refresh state changed during update")); + return Err(endpoint_refresh_state_changed()); }; let target_state = endpoint_refresh_target_state(&state, &pending); let local_peer = current_local_runtime_peer(&target_state); @@ -5027,6 +5334,35 @@ async fn refresh_bucket_targets_after_endpoint_edit(pending_id: &str, service_ac expected_incarnation_id, ) .await?; + + rewritten.push(bucket.name.clone()); + + // A remove accepted on another node can clear this journal while the + // bucket rewrite is in flight. Re-check after the write: if it removed + // the edited peer, undo the stale targets this pass restored, after + // releasing the process-local target lock. If it committed the same + // edit, the write is equivalent but this driver no longer owns + // finalization. + let latest = load_site_replication_state().await?; + let postwrite = endpoint_refresh_postwrite_state(&latest, pending_id, &pending.peer.deployment_id); + drop(_targets_guard); + if postwrite == EndpointRefreshPostwriteState::TargetRemoved { + let removed = HashSet::from([pending.peer.deployment_id.clone()]); + let mut cleanup_error = None; + for name in &rewritten { + // Attempt every bucket: one failure must not leave the rest + // of this pass's stale targets behind. + if let Err(err) = cleanup_removed_site_replication_bucket(name, &removed).await { + cleanup_error.get_or_insert(err); + } + } + if let Some(err) = cleanup_error { + return Err(err); + } + } + if postwrite != EndpointRefreshPostwriteState::Current { + return Err(endpoint_refresh_state_changed()); + } } Ok(()) @@ -6023,14 +6359,26 @@ async fn apply_iam_group_info_item( return Ok(IamItemVerdict::Apply); } + // A membership change carries the sender's client payload verbatim, and + // the madmin wire maps an unset `groupStatus` to Enabled + // (`crates/madmin/src/group.rs`), so writing the status alongside members + // would re-enable a group this site has disabled. Disabled can only come + // from an explicit "disabled", so it is never that default: honouring it + // keeps the full-IAM snapshot (`site_replication::hooks`), which always + // carries members and the real status, able to propagate a disabled group + // to a peer that does not have it yet — `GroupInfo::new` would otherwise + // create it enabled and hand its members live access. + let status_is_explicit = update.members.is_empty() || matches!(update.status, GroupStatus::Disabled); iam_sys .add_users_to_group_at(&update.group, update.members, stamp) .await .map_err(ApiError::from)?; - iam_sys - .set_group_status_at(&update.group, matches!(update.status, GroupStatus::Enabled), stamp) - .await - .map_err(ApiError::from)?; + if status_is_explicit { + iam_sys + .set_group_status_at(&update.group, matches!(update.status, GroupStatus::Enabled), stamp) + .await + .map_err(ApiError::from)?; + } Ok(IamItemVerdict::Apply) } @@ -6611,13 +6959,15 @@ impl Operation for SiteReplicationRemoveHandler { let (pending_remove, local_peer) = { let _bucket_op_guard = SITE_REPLICATION_BUCKET_OP_LOCK.write().await; update_site_replication_state_when_changed(move |state| { - if pending_endpoint_refresh(state).is_some() { + let local_peer = local_peer_at_endpoint(local_endpoint, state); + if let Some(pending) = pending_endpoint_refresh(state) + && !pending_endpoint_refresh_allows_remove(state, &pending, &local_peer, &remove_req) + { return Err(s3_error!(InvalidRequest, "endpoint target refresh is pending")); } if state.pending_rotation.is_some() { return Err(s3_error!(InvalidRequest, "service account rotation is pending")); } - let local_peer = local_peer_at_endpoint(local_endpoint, state); // Resuming: the peers were already told about this pending // removal, so re-persisting the same record buys nothing. @@ -6632,6 +6982,11 @@ impl Operation for SiteReplicationRemoveHandler { let mut peer_remove_req = remove_req.clone(); peer_remove_req.requesting_dep_id = local_peer.deployment_id.clone(); *state = remove_sites(std::mem::take(state), remove_req); + // A permitted remove can include the endpoint being edited. + // Drop that obsolete journal before recording the removal; + // otherwise its resume path would commit the edited peer back + // into the topology on the next heavyweight tick. + discard_endpoint_refresh_if_target_was_removed(state); let pending = PendingRemove { id: Uuid::new_v4().to_string(), req: peer_remove_req, @@ -7194,10 +7549,20 @@ impl Operation for SiteReplicationEditHandler { if persisted_pending.is_some() && !endpoint_refresh_requested { return Err(s3_error!(InvalidRequest, "an endpoint target refresh is already pending")); } + if endpoint_refresh_requested + && let Some(persisted) = persisted_pending.as_ref() + && endpoint_refresh_ilm_override_conflicts(persisted, ilm_expiry_override) + { + return Err(s3_error!( + InvalidRequest, + "endpoint target refresh is pending with a different ilm expiry setting" + )); + } let pending = endpoint_refresh_requested.then(|| { persisted_pending.clone().unwrap_or_else(|| PendingEndpointRefresh { id: Uuid::new_v4().to_string(), peer: normalize_peer_info(incoming.clone()), + ilm_expiry_override, remote_peers: current_state.peers.clone(), acked_deployment_ids: BTreeSet::new(), }) @@ -7346,7 +7711,7 @@ impl Operation for SiteReplicationEditHandler { let acked_pending_id = pending_id.clone(); let service_account_access_key = update_site_replication_state(move |state| { let Some(pending) = pending_endpoint_refresh(state).filter(|pending| pending.id == acked_pending_id) else { - return Err(s3_error!(InvalidRequest, "endpoint target refresh state changed during update")); + return Err(endpoint_refresh_state_changed()); }; let pending = merge_pending_endpoint_refresh(state, &pending, acked_deployment_ids)?; set_pending_endpoint_refresh(state, pending)?; @@ -7360,9 +7725,9 @@ impl Operation for SiteReplicationEditHandler { refresh_bucket_targets_after_endpoint_edit(&pending_id, &service_account_secret_key).await?; update_site_replication_state(move |state| { let Some(pending) = pending_endpoint_refresh(state).filter(|pending| pending.id == pending_id) else { - return Err(s3_error!(InvalidRequest, "endpoint target refresh state changed during update")); + return Err(endpoint_refresh_state_changed()); }; - *state = edit_state(std::mem::take(state), pending.peer, ilm_expiry_override); + *state = edit_state(std::mem::take(state), pending.peer, pending.ilm_expiry_override); clear_pending_endpoint_refresh(state); Ok(()) }) @@ -7567,6 +7932,7 @@ impl Operation for SRPeerEditHandler { PendingEndpointRefresh { id: commit_refresh_id.unwrap_or_default(), peer: incoming, + ilm_expiry_override, remote_peers: BTreeMap::new(), acked_deployment_ids: BTreeSet::new(), }, @@ -7631,7 +7997,7 @@ impl Operation for SRPeerEditHandler { let Some(pending) = pending_endpoint_refresh(state).filter(|pending| pending.id == pending_id) else { return Ok(StateCommit::Unchanged(false)); }; - *state = apply_internal_peer_edit(std::mem::take(state), &local_peer, pending.peer, ilm_expiry_override)?; + *state = apply_internal_peer_edit(std::mem::take(state), &local_peer, pending.peer, pending.ilm_expiry_override)?; clear_pending_endpoint_refresh(state); Ok(StateCommit::Changed(true)) }) @@ -8108,6 +8474,7 @@ mod tests { false, (5_000_000_000, Some(heartbeat_last_online)), (4096, 8), + TimedErrStats::default(), ); assert!(!entry.online, "an unreachable peer must not be reported online"); @@ -8123,8 +8490,24 @@ mod tests { /// or a two-site cluster would double-count its own traffic. #[test] fn peer_metric_entry_keeps_replication_counters_on_the_local_entry() { - let local = peer_metric_entry("local", "http://local.example:9000", true, true, (0, None), (4096, 8)); - let remote = peer_metric_entry("remote", "http://remote.example:9000", false, true, (0, None), (4096, 8)); + let local = peer_metric_entry( + "local", + "http://local.example:9000", + true, + true, + (0, None), + (4096, 8), + TimedErrStats::default(), + ); + let remote = peer_metric_entry( + "remote", + "http://remote.example:9000", + false, + true, + (0, None), + (4096, 8), + TimedErrStats::default(), + ); assert_eq!(local.replicated_size, 4096); assert_eq!(local.replicated_count, 8); @@ -8133,6 +8516,28 @@ mod tests { assert!(remote.online, "a reachable remote peer is still online"); } + #[test] + fn peer_metric_entry_wires_local_replication_failures() { + let source = include_str!("site_replication.rs"); + let body = source + .split("fn peer_metric_entry(") + .nth(1) + .and_then(|rest| rest.split("async fn build_metrics_summary").next()) + .expect("peer metric builder"); + + assert!(body.contains("failed:"), "the local site metric must expose failed replication totals"); + + let failures = TimedErrStats { + totals: RStat { count: 3.0, bytes: 900 }, + ..Default::default() + }; + let local = peer_metric_entry("local", "http://local", true, true, (0, None), (0, 0), failures.clone()); + let remote = peer_metric_entry("remote", "http://remote", false, true, (0, None), (0, 0), failures); + assert_eq!(local.failed.totals.count, 3.0); + assert_eq!(local.failed.totals.bytes, 900); + assert_eq!(remote.failed.totals.count, 0.0, "node-local failures must not be copied to peers"); + } + #[test] fn test_rotation_secret_candidates_try_the_installed_secret_before_the_new_one() { let mut pending = PendingRotation { @@ -8316,12 +8721,22 @@ mod tests { } fn sr_group_item(group: &str, members: &[&str], is_remove: bool, updated_at: OffsetDateTime) -> SRIAMItem { + sr_group_item_with_status(group, members, is_remove, rustfs_madmin::GroupStatus::Enabled, updated_at) + } + + fn sr_group_item_with_status( + group: &str, + members: &[&str], + is_remove: bool, + status: rustfs_madmin::GroupStatus, + updated_at: OffsetDateTime, + ) -> SRIAMItem { let mut item = sr_item("group-info", updated_at); item.group_info = Some(SRGroupInfo { update_req: rustfs_madmin::GroupAddRemove { group: group.to_string(), members: members.iter().map(|member| member.to_string()).collect(), - status: rustfs_madmin::GroupStatus::Enabled, + status, is_remove, }, api_version: Some(SITE_REPL_API_VERSION.to_string()), @@ -8465,6 +8880,72 @@ mod tests { clear_seeded_state().await; } + #[tokio::test] + #[serial] + async fn apply_group_member_add_preserves_a_disabled_group_status() { + publish_ready_iam_context().await; + seed_two_peer_state_for_iam_apply().await; + let iam = current_iam_handle().expect("test IAM"); + let member = "sr-disabled-group-member"; + let group = "sr-disabled-group"; + iam.create_user(member, &user_req("member-secret-key-123", rustfs_madmin::AccountStatus::Enabled)) + .await + .expect("member"); + iam.add_users_to_group(group, Vec::new()).await.expect("group"); + iam.set_group_status(group, false).await.expect("disable group"); + + apply_iam_item(sr_group_item( + group, + &[member], + false, + OffsetDateTime::now_utc() + time::Duration::seconds(1), + )) + .await + .expect("replicated member add"); + + let info = iam.get_group_info(group).await.expect("group after replicated member add"); + assert_eq!(info.status, "disabled", "a membership-only update must not enable the group"); + assert!(info.members.iter().any(|current| current == member)); + clear_seeded_state().await; + } + + /// A full-IAM snapshot always carries members and the sender's real group + /// status (`site_replication::hooks`). A peer that does not have the group + /// yet creates it through `GroupInfo::new`, which is enabled — so dropping + /// an explicit Disabled here would hand every member of a frozen group + /// live access on the peer, which is the same escalation the + /// membership-only guard above exists to prevent. + #[tokio::test] + #[serial] + async fn apply_group_snapshot_propagates_a_disabled_status_with_members() { + publish_ready_iam_context().await; + seed_two_peer_state_for_iam_apply().await; + let iam = current_iam_handle().expect("test IAM"); + let member = "sr-snapshot-group-member"; + let group = "sr-snapshot-disabled-group"; + iam.create_user(member, &user_req("member-secret-key-123", rustfs_madmin::AccountStatus::Enabled)) + .await + .expect("member"); + + apply_iam_item(sr_group_item_with_status( + group, + &[member], + false, + rustfs_madmin::GroupStatus::Disabled, + OffsetDateTime::now_utc() + time::Duration::seconds(1), + )) + .await + .expect("replicated group snapshot"); + + let info = iam.get_group_info(group).await.expect("group after replicated snapshot"); + assert_eq!( + info.status, "disabled", + "a snapshot carrying an explicit disabled status must not create an enabled group" + ); + assert!(info.members.iter().any(|current| current == member)); + clear_seeded_state().await; + } + /// Review finding on rustfs#7195: an older grant and a newer revoke for /// the same record delivered concurrently must always leave the revoke, /// whichever request reaches the transaction first — the verdict and @@ -9458,6 +9939,50 @@ mod tests { assert_eq!(merged.acked_deployment_ids, BTreeSet::from(["peer-a".to_string(), "peer-b".to_string()])); } + /// A refresh that cannot finish used to take the whole heavyweight pass + /// with it, including the retry drain, so an unreachable peer froze IAM + /// and bucket replay to every healthy peer as well. The bucket reconciler + /// still waits - it rewrites the topology the refresh is changing - but + /// the drain must run. + #[test] + fn a_pending_endpoint_refresh_still_drains_the_retry_queue() { + let source = include_str!("site_replication.rs"); + let body = source + .split("fn reconcile_site_replication_wiring()") + .nth(1) + .and_then(|rest| rest.split("async fn send_site_replication_bootstrap_plan").next()) + .expect("reconcile function body"); + let refresh_arm = body + .split("resume_pending_endpoint_refresh(") + .nth(1) + .and_then(|rest| rest.split("A wedged rotation").next()) + .expect("endpoint refresh arm"); + + assert!( + refresh_arm.contains("drain_site_replication_retry_queue().await;"), + "a pending endpoint refresh must not stall replay to the healthy peers" + ); + assert!( + !refresh_arm.contains("reconcile_site_replication_buckets"), + "bucket wiring still waits for the refresh to settle" + ); + } + + #[test] + fn reconcile_redrives_a_pending_endpoint_refresh() { + let source = include_str!("site_replication.rs"); + let body = source + .split("fn reconcile_site_replication_wiring()") + .nth(1) + .and_then(|rest| rest.split("async fn send_site_replication_bootstrap_plan").next()) + .expect("reconcile function body"); + + assert!( + body.contains("resume_pending_endpoint_refresh"), + "the heavyweight reconcile tick must resume a persisted endpoint refresh" + ); + } + #[test] fn test_internal_endpoint_refresh_retry_is_strictly_idempotent() { let committed = PeerInfo { @@ -11391,6 +11916,188 @@ mod tests { ); } + fn endpoint_refresh_remove_fixture() -> (SiteReplicationState, PendingEndpointRefresh, PeerInfo) { + let local = PeerInfo { + deployment_id: "site-a-dep".to_string(), + name: "site-a".to_string(), + ..peer("site-a", "https://site-a.example.com") + }; + let edited = PeerInfo { + deployment_id: "site-b-dep".to_string(), + name: "site-b".to_string(), + ..peer("site-b", "https://new-site-b.example.com") + }; + let unavailable = PeerInfo { + deployment_id: "site-c-dep".to_string(), + name: "site-c".to_string(), + ..peer("site-c", "https://site-c.example.com") + }; + let peers = BTreeMap::from([ + (local.deployment_id.clone(), local.clone()), + (edited.deployment_id.clone(), edited.clone()), + (unavailable.deployment_id.clone(), unavailable), + ]); + let pending = PendingEndpointRefresh { + id: "refresh-1".to_string(), + peer: edited, + remote_peers: peers.clone(), + acked_deployment_ids: BTreeSet::from(["site-b-dep".to_string()]), + ..Default::default() + }; + let state = SiteReplicationState { + name: local.name.clone(), + peers, + pending_endpoint_refresh: Some(pending.clone()), + ..Default::default() + }; + (state, pending, local) + } + + #[test] + fn pending_endpoint_refresh_allows_removing_every_unacked_peer() { + let (state, pending, local) = endpoint_refresh_remove_fixture(); + let remove_unacked = SRRemoveReq { + site_names: vec!["site-c".to_string()], + ..Default::default() + }; + let remove_acked_only = SRRemoveReq { + site_names: vec!["site-b".to_string()], + ..Default::default() + }; + + assert!(pending_endpoint_refresh_allows_remove(&state, &pending, &local, &remove_unacked)); + assert!(!pending_endpoint_refresh_allows_remove(&state, &pending, &local, &remove_acked_only)); + assert!(pending_endpoint_refresh_allows_remove( + &state, + &pending, + &local, + &SRRemoveReq { + remove_all: true, + ..Default::default() + } + )); + } + + /// The peer-side edit handler journals a refresh with no `remote_peers` + /// and commits it inside the same request. That journal reads as complete + /// on sight, so the reconcile tick must not adopt it: committing it here + /// would use `edit_state` (no local-name sync) and would clear the journal + /// under the request that owns it, whose own commit then reports the + /// refresh as changed and denies the coordinator its acknowledgement. + /// Needs a live store to exercise, so pin the shape instead: the undo + /// must cover every bucket the pass already rewrote, not only the one + /// holding the lock when the removal was noticed. + #[test] + fn an_interrupted_endpoint_refresh_undoes_every_bucket_it_rewrote() { + let body = include_str!("site_replication.rs") + .split("async fn refresh_bucket_targets_after_endpoint_edit") + .nth(1) + .and_then(|rest| rest.split("async fn site_bucket_resync_manifest_entry").next()) + .expect("refresh body"); + + assert!(body.contains("rewritten.push(bucket.name.clone());")); + assert!( + body.contains("for name in "), + "the removal undo must walk every bucket rewritten in this pass" + ); + } + + #[test] + fn a_pending_endpoint_refresh_pins_its_ilm_expiry_override() { + let (_, pending, _) = endpoint_refresh_remove_fixture(); + let with_override = PendingEndpointRefresh { + ilm_expiry_override: Some(true), + ..pending.clone() + }; + + assert!( + !endpoint_refresh_ilm_override_conflicts(&with_override, None), + "re-running the edit without the flag redrives the pinned refresh" + ); + assert!(!endpoint_refresh_ilm_override_conflicts(&with_override, Some(true))); + assert!( + endpoint_refresh_ilm_override_conflicts(&with_override, Some(false)), + "a different requested value must be rejected, not silently ignored" + ); + assert!( + endpoint_refresh_ilm_override_conflicts(&pending, Some(false)), + "the journal pins an unset override too" + ); + } + + #[test] + fn a_peer_side_endpoint_refresh_journal_is_not_resumed_locally() { + let (state, coordinator_pending, local) = endpoint_refresh_remove_fixture(); + let peer_side = PendingEndpointRefresh { + remote_peers: BTreeMap::new(), + acked_deployment_ids: BTreeSet::new(), + ..coordinator_pending.clone() + }; + + assert!(pending_endpoint_refresh_is_locally_driven(&coordinator_pending)); + assert!(!pending_endpoint_refresh_is_locally_driven(&peer_side)); + assert!( + pending_endpoint_refresh_is_complete(&state, &peer_side, &local), + "a peer-side journal has no required peers, so only the driver guard keeps the tick off it" + ); + assert!( + include_str!("site_replication.rs") + .split("async fn resume_pending_endpoint_refresh") + .nth(1) + .and_then(|rest| rest.split("async fn pending_remove_ready_to_finalize").next()) + .expect("resume body") + .contains("pending_endpoint_refresh_is_locally_driven"), + "the resume path must refuse a journal it does not own" + ); + } + + #[test] + fn pending_endpoint_refresh_completion_ignores_a_peer_removed_from_the_topology() { + let (mut state, pending, local) = endpoint_refresh_remove_fixture(); + assert!(!pending_endpoint_refresh_is_complete(&state, &pending, &local)); + + state.peers.remove("site-c-dep"); + assert!(pending_endpoint_refresh_is_complete(&state, &pending, &local)); + } + + #[test] + fn removing_the_edited_peer_discards_its_pending_endpoint_refresh() { + let (state, pending, _) = endpoint_refresh_remove_fixture(); + assert_eq!( + endpoint_refresh_postwrite_state(&state, &pending.id, &pending.peer.deployment_id), + EndpointRefreshPostwriteState::Current + ); + let mut state = remove_sites( + state, + SRRemoveReq { + site_names: vec!["site-b".to_string(), "site-c".to_string()], + ..Default::default() + }, + ); + + discard_endpoint_refresh_if_target_was_removed(&mut state); + + assert!(pending_endpoint_refresh(&state).is_none()); + assert!( + state + .retry_queue + .iter() + .all(|event| event.path != SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH) + ); + assert_eq!( + endpoint_refresh_postwrite_state(&state, &pending.id, &pending.peer.deployment_id), + EndpointRefreshPostwriteState::TargetRemoved, + "a refresh write racing this removal must clean the stale target it may have restored" + ); + + let mut completed = endpoint_refresh_remove_fixture().0; + completed.pending_endpoint_refresh = None; + assert_eq!( + endpoint_refresh_postwrite_state(&completed, &pending.id, &pending.peer.deployment_id), + EndpointRefreshPostwriteState::Superseded + ); + } + #[test] fn test_normalize_join_peers_rewrites_local_endpoint_to_real_deployment_id() { let local_peer = PeerInfo { @@ -12394,6 +13101,7 @@ mod tests { PendingEndpointRefresh { id: "refresh-1".to_string(), peer: retry_peer.clone(), + ilm_expiry_override: None, remote_peers, acked_deployment_ids: BTreeSet::new(), }, @@ -12651,6 +13359,51 @@ mod tests { assert_eq!(operation.pending_peers, vec!["remote-b".to_string()]); } + #[test] + fn test_pending_operation_for_state_reports_endpoint_refresh_progress() { + let local = PeerInfo { + deployment_id: "local".to_string(), + ..peer("local", "https://local.example.com") + }; + let remote_a = PeerInfo { + deployment_id: "remote-a".to_string(), + ..peer("remote-a", "https://remote-a.example.com") + }; + let remote_b = PeerInfo { + deployment_id: "remote-b".to_string(), + ..peer("remote-b", "https://remote-b.example.com") + }; + let state = SiteReplicationState { + peers: BTreeMap::from([ + (local.deployment_id.clone(), local.clone()), + (remote_a.deployment_id.clone(), remote_a.clone()), + (remote_b.deployment_id.clone(), remote_b.clone()), + ]), + pending_endpoint_refresh: Some(PendingEndpointRefresh { + id: "refresh-id".to_string(), + peer: PeerInfo { + endpoint: "https://remote-a-new.example.com".to_string(), + ..remote_a.clone() + }, + remote_peers: BTreeMap::from([ + (remote_a.deployment_id.clone(), remote_a), + (remote_b.deployment_id.clone(), remote_b), + (local.deployment_id.clone(), local.clone()), + ]), + acked_deployment_ids: BTreeSet::from(["remote-b".to_string()]), + ..Default::default() + }), + ..Default::default() + }; + + let operation = pending_operation_for_state(&state, &local).expect("pending endpoint refresh operation"); + + assert_eq!(operation.operation, "endpoint-refresh"); + assert_eq!(operation.id, "refresh-id"); + assert_eq!(operation.acked_peers, vec!["remote-b".to_string()]); + assert_eq!(operation.pending_peers, vec!["remote-a".to_string()]); + } + #[test] fn test_status_peer_error_summarizes_details() { let remote = PeerInfo { diff --git a/rustfs/src/admin/handlers/user.rs b/rustfs/src/admin/handlers/user.rs index 04d28290f..fb715cb1c 100644 --- a/rustfs/src/admin/handlers/user.rs +++ b/rustfs/src/admin/handlers/user.rs @@ -1318,6 +1318,24 @@ impl Operation for ImportIam { failed, }; + // The entities are already imported locally. A snapshot that cannot be + // scheduled is a convergence delay the reconcile pass still closes, so + // it must not turn a completed import into a failed request - the same + // best-effort contract every other site-replication hook here follows. + if let Err(err) = + crate::site_replication::enqueue_site_replication_iam_snapshot("iam import scheduled a full snapshot").await + { + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_USER, + event = EVENT_ADMIN_USER_STATE, + action = "import_iam", + result = "site_replication_snapshot_not_scheduled", + error = ?err, + "admin user state" + ); + } + let body = serde_json::to_vec(&ret).map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?; let mut header = HeaderMap::new(); @@ -1424,6 +1442,16 @@ mod tests { assert!(include_str!("user.rs").contains(mapper_call)); } + #[test] + fn import_iam_enqueues_a_site_replication_snapshot() { + let body = source_block(include_str!("user.rs"), "impl Operation for ImportIam"); + + assert!( + body.contains("enqueue_site_replication_iam_snapshot"), + "a successful IAM import must schedule a full IAM snapshot for every remote site" + ); + } + #[test] fn test_should_check_deny_only_for_regular_self_request() { let cred = Credentials { diff --git a/rustfs/src/admin/replication_metrics_wire.rs b/rustfs/src/admin/replication_metrics_wire.rs index 15c9cad81..5cdc5851a 100644 --- a/rustfs/src/admin/replication_metrics_wire.rs +++ b/rustfs/src/admin/replication_metrics_wire.rs @@ -409,6 +409,26 @@ fn transfer_summaries(stats: &InternalReplicationStats) -> (XferSummaryWire, Tar (summary, per_target) } +/// Node-level failure counters for `errors`. The sibling `retries` field +/// stays zero on purpose: it means redeliveries in the minio-go shape, and a +/// failed object is not retried by an event today (it waits for the scanner's +/// heal pass), so reporting failures there would claim a redelivery that +/// never happened. +fn failure_counters(stats: &InternalReplicationStats) -> CounterSummaryWire { + let (total, last1m, last1hr) = stats.stats.values().fold((0i64, 0i64, 0i64), |acc, stat| { + ( + acc.0.saturating_add(stat.fail_stats.count), + acc.1.saturating_add(stat.fail_stats.last_minute.count), + acc.2.saturating_add(stat.fail_stats.last_hour.count), + ) + }); + CounterSummaryWire { + total: u64::try_from(total.max(0)).unwrap_or_default(), + last1m: u64::try_from(last1m.max(0)).unwrap_or_default(), + last1hr: u64::try_from(last1hr.max(0)).unwrap_or_default(), + } +} + impl MetricsV2Wire { /// Project the aggregated internal stats onto the `MetricsV2` shape. /// @@ -418,6 +438,7 @@ impl MetricsV2Wire { /// `queueStats.nodes` and treats an empty list as "no data". pub(crate) fn from_stats(bucket_stats: &BucketStats, node_name: &str) -> Self { let (xfer_stats, tgt_xfer_stats) = transfer_summaries(&bucket_stats.replication_stats); + let failed = failure_counters(&bucket_stats.replication_stats); let mut nodes: Vec = bucket_stats .queue_stats .nodes @@ -436,6 +457,7 @@ impl MetricsV2Wire { q_stats: InQueueMetricWire::from(&bucket_stats.replication_stats.q_stat), xfer_stats: xfer_stats.clone(), tgt_xfer_stats: tgt_xfer_stats.clone(), + errors: failed, ..Default::default() }); } else { @@ -444,6 +466,7 @@ impl MetricsV2Wire { if let Some(first) = nodes.first_mut() { first.xfer_stats = xfer_stats.clone(); first.tgt_xfer_stats = tgt_xfer_stats.clone(); + first.errors = failed; } } @@ -478,6 +501,12 @@ mod tests { target.replicated_size = 4096; target.failed.count = 3; target.failed.size = 900; + target.fail_stats.count = 3; + target.fail_stats.size = 900; + target.fail_stats.last_minute.count = 2; + target.fail_stats.last_minute.size = 600; + target.fail_stats.last_hour.count = 3; + target.fail_stats.last_hour.size = 900; target.bandwidth_limit_bytes_per_sec = 1024; target.current_bandwidth_bytes_per_sec = 512.5; stats @@ -537,6 +566,10 @@ mod tests { assert_eq!(node["queueStats"]["peak"], node["queueStats"]["max"]); assert!(node["activeWorkers"].get("curr").is_some()); assert!(node["transferSummary"].get("Total").is_some()); + assert_eq!(node["errors"]["total"], 3); + assert_eq!(node["errors"]["last1m"], 2); + assert_eq!(node["errors"]["last1hr"], 3); + assert_eq!(node["retries"]["total"], 0, "failures are not redeliveries; retries must not claim one"); assert_eq!(json["downtimeInfo"], serde_json::json!({})); } diff --git a/rustfs/src/site_replication/retry.rs b/rustfs/src/site_replication/retry.rs index c9b9e73a0..738a645d6 100644 --- a/rustfs/src/site_replication/retry.rs +++ b/rustfs/src/site_replication/retry.rs @@ -217,6 +217,26 @@ pub(crate) fn settle_observed_site_replication_retry_event( before.saturating_sub(queue.len()) } +/// Make sure `peer` has a collapsed entry for `path` without counting the +/// call as a delivery failure. A bulk local mutation (`import-iam`) needs the +/// entry to exist so the next drain sends the snapshot; routing it through +/// [`upsert_site_replication_retry_event`] would raise `retry_count` on every +/// import and escalate a healthy peer to `failed` after +/// [`SITE_REPLICATION_RETRY_FAILED_AFTER`] of them, with the scheduling note +/// shown to operators as `lastError`. +pub(crate) fn ensure_site_replication_retry_event( + queue: &mut Vec, + peer: &PeerInfo, + path: &str, + reason: &str, +) -> S3Result> { + let path = collapsed_retry_queue_path(path).unwrap_or(path); + if queue.iter().any(|event| retry_event_matches(event, peer, path)) { + return Ok(Vec::new()); + } + push_site_replication_retry_event(queue, peer, path, summarize_peer_error_detail(reason), false, None) +} + pub(crate) fn upsert_site_replication_retry_event( queue: &mut Vec, peer: &PeerInfo, @@ -244,6 +264,17 @@ pub(crate) fn upsert_site_replication_retry_event( return Ok(Vec::new()); } + push_site_replication_retry_event(queue, peer, path, detail, peer_unreachable, generation) +} + +fn push_site_replication_retry_event( + queue: &mut Vec, + peer: &PeerInfo, + path: &str, + detail: String, + peer_unreachable: bool, + generation: Option, +) -> S3Result> { let slots_needed = queue .len() .saturating_add(1) @@ -274,7 +305,7 @@ pub(crate) fn upsert_site_replication_retry_event( retry_count: 1, failed: false, last_error: detail, - updated_at: Some(now), + updated_at: Some(OffsetDateTime::now_utc()), edit_generation: generation, peer_unreachable, deletions_recorded: false, @@ -365,6 +396,60 @@ pub(crate) async fn enqueue_site_replication_retry_event_for_generation( } } +/// Returns the number of peers whose snapshot entry is escalated and therefore +/// will not carry this scheduling: the marker records a deletion that a +/// snapshot cannot replay, and only a repair settles it, so clearing it to make +/// the entry drainable again would drop that liability. +pub(crate) fn record_iam_snapshot_retries( + state: &mut SiteReplicationState, + local_peer: &PeerInfo, + reason: &str, +) -> S3Result { + let peers = state + .peers + .values() + .filter(|peer| { + peer.deployment_id != local_peer.deployment_id && !same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) + }) + .cloned() + .collect::>(); + let mut escalated = 0usize; + for peer in peers { + if state.retry_queue.iter().any(|event| { + retry_event_matches(event, &peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH) + && event.last_error == SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER + }) { + escalated += 1; + continue; + } + ensure_site_replication_retry_event(&mut state.retry_queue, &peer, SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH, reason)?; + } + Ok(escalated) +} + +/// Schedule one collapsed full-IAM snapshot per remote peer after a bulk +/// local mutation such as `import-iam`. +pub(crate) async fn enqueue_site_replication_iam_snapshot(reason: &str) -> S3Result<()> { + let state = load_site_replication_state().await?; + if !state.enabled() { + return Ok(()); + } + let local_peer = current_local_runtime_peer(&state); + let reason = reason.to_string(); + let escalated = update_site_replication_state(move |state| record_iam_snapshot_retries(state, &local_peer, &reason)).await?; + if escalated > 0 { + warn!( + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_SITE_REPLICATION, + event = EVENT_ADMIN_SITE_REPLICATION_STATE, + escalated, + result = "iam_snapshot_not_scheduled_for_escalated_peer", + "site replication peers hold an escalated IAM entry; the snapshot waits for a repair" + ); + } + Ok(()) +} + pub(crate) const SITE_REPLICATION_PEER_IAM_ITEM_WIRE_PATH: &str = "/rustfs/admin/v3/site-replication/peer/iam-item"; /// Per-peer cap on recorded deletion bodies. Beyond it the peer's collapsed diff --git a/rustfs/src/site_replication/state.rs b/rustfs/src/site_replication/state.rs index efc8b16e5..731dd5496 100644 --- a/rustfs/src/site_replication/state.rs +++ b/rustfs/src/site_replication/state.rs @@ -168,6 +168,8 @@ mod rfc3339_map { pub(crate) struct PendingEndpointRefresh { pub(crate) id: String, pub(crate) peer: PeerInfo, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) ilm_expiry_override: Option, #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] pub(crate) remote_peers: BTreeMap, #[serde(default, skip_serializing_if = "BTreeSet::is_empty")] diff --git a/rustfs/src/site_replication/tests.rs b/rustfs/src/site_replication/tests.rs index f1c06131f..b69b1d3d7 100644 --- a/rustfs/src/site_replication/tests.rs +++ b/rustfs/src/site_replication/tests.rs @@ -693,6 +693,122 @@ fn test_record_iam_deletion_marks_newest_wins_and_expires_by_age_only() { ); } +/// Scheduling a snapshot is not a delivery failure. Repeated imports - the +/// normal way a bulk IAM migration is done, one archive at a time - must not +/// walk the peer's entry up to the escalation threshold and report a healthy +/// site as `retryStats.failed` with the scheduling note as its `lastError`. +#[test] +fn repeated_iam_import_snapshots_do_not_escalate_a_healthy_peer() { + let local = PeerInfo { + deployment_id: "local-dep".to_string(), + ..peer("local", "https://local.example.com") + }; + let remote = PeerInfo { + deployment_id: "remote-a".to_string(), + ..peer("remote-a", "https://a.example.com") + }; + let mut state = SiteReplicationState { + peers: BTreeMap::from([ + (local.deployment_id.clone(), local.clone()), + (remote.deployment_id.clone(), remote), + ]), + ..Default::default() + }; + + for _ in 0..(SITE_REPLICATION_RETRY_FAILED_AFTER + 2) { + record_iam_snapshot_retries(&mut state, &local, "iam import scheduled a full snapshot").expect("record snapshot"); + } + + assert_eq!(state.retry_queue.len(), 1); + let event = &state.retry_queue[0]; + assert_eq!(event.retry_count, 1, "a schedule must not count as a delivery attempt"); + assert!(!event.failed, "a scheduled snapshot must not report as an escalated failure"); +} + +/// An escalated entry records a deletion a snapshot cannot replay: only a +/// repair settles it. Scheduling an import snapshot must not clear that +/// marker to make the entry drainable again, and the peer it skips has to be +/// reported rather than silently left behind. +#[test] +fn an_escalated_peer_keeps_its_marker_and_is_reported() { + let local = PeerInfo { + deployment_id: "local-dep".to_string(), + ..peer("local", "https://local.example.com") + }; + let remote = PeerInfo { + deployment_id: "remote-a".to_string(), + ..peer("remote-a", "https://a.example.com") + }; + let mut state = SiteReplicationState { + peers: BTreeMap::from([ + (local.deployment_id.clone(), local.clone()), + (remote.deployment_id.clone(), remote.clone()), + ]), + retry_queue: vec![SiteReplicationRetryEvent { + id: "escalated".to_string(), + peer_deployment_id: remote.deployment_id.clone(), + peer_endpoint: remote.endpoint, + path: SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH.to_string(), + retry_count: SITE_REPLICATION_RETRY_FAILED_AFTER, + failed: true, + last_error: SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER.to_string(), + deletions_recorded: true, + ..Default::default() + }], + ..Default::default() + }; + + let escalated = + record_iam_snapshot_retries(&mut state, &local, "iam import scheduled a full snapshot").expect("record snapshot retries"); + + assert_eq!(escalated, 1); + assert_eq!(state.retry_queue.len(), 1); + assert_eq!( + state.retry_queue[0].last_error, SITE_REPLICATION_RETRY_SNAPSHOT_REPLAYED_MARKER, + "the unreplayable-deletion marker must survive a snapshot schedule" + ); +} + +#[test] +fn iam_import_snapshot_retry_is_recorded_once_per_remote_peer() { + let local = PeerInfo { + deployment_id: "local-dep".to_string(), + ..peer("local", "https://local.example.com") + }; + let remote_a = PeerInfo { + deployment_id: "remote-a".to_string(), + ..peer("remote-a", "https://a.example.com") + }; + let remote_b = PeerInfo { + deployment_id: "remote-b".to_string(), + ..peer("remote-b", "https://b.example.com") + }; + let mut state = SiteReplicationState { + peers: BTreeMap::from([ + (local.deployment_id.clone(), local.clone()), + (remote_a.deployment_id.clone(), remote_a), + (remote_b.deployment_id.clone(), remote_b), + ]), + ..Default::default() + }; + + record_iam_snapshot_retries(&mut state, &local, "IAM import snapshot pending").expect("record snapshot retries"); + + assert_eq!(state.retry_queue.len(), 2); + assert!( + state + .retry_queue + .iter() + .all(|event| event.path == SITE_REPLICATION_RETRY_IAM_SNAPSHOT_PATH) + ); + assert!( + state + .retry_queue + .iter() + .all(|event| event.peer_deployment_id != local.deployment_id) + ); +} + /// A failed deletion delivery persists a replay record next to the collapsed /// retry entry; a fresh entry is stamped `deletions_recorded` so a later /// replay can settle it, and a repeated deletion of the same entity keeps the diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 363be627c..4dbc1b341 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -17,6 +17,7 @@ use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher}; use std::sync::{Arc, LazyLock}; +use std::time::Duration; use rand::RngExt as _; use rustfs_storage_api as storage_contracts; @@ -836,6 +837,39 @@ impl StorageReplicationStatsHandle { pub(crate) async fn site_metrics_snapshot(&self) -> ReplicationSiteMetricsSnapshot { let metrics = self.inner.get_sr_metrics_for_node().await; + // Aggregate under the read lock rather than through `get_all`: that + // clones every bucket's stats, and `FailStats.recent` is bounded only + // by the one-hour window, so an unreachable target under load - the + // very case an operator polls this for - makes the copy large. The + // windows come from the live samples; the serialized `last_minute` / + // `last_hour` snapshots are stamped onto per-bucket clones elsewhere + // and stay zero in this node-local cache. + let ( + failed_count, + failed_bytes, + failed_last_minute_count, + failed_last_minute_bytes, + failed_last_hour_count, + failed_last_hour_bytes, + ) = { + let cache = self.inner.cache.read().await; + cache + .values() + .flat_map(|bucket| bucket.stats.values()) + .fold((0i64, 0i64, 0i64, 0i64, 0i64, 0i64), |totals, stat| { + let (minute, hour) = stat + .fail_stats + .recent_windows(Duration::from_secs(60), Duration::from_secs(3600)); + ( + totals.0.saturating_add(stat.fail_stats.count), + totals.1.saturating_add(stat.fail_stats.size), + totals.2.saturating_add(minute.count), + totals.3.saturating_add(minute.size), + totals.4.saturating_add(hour.count), + totals.5.saturating_add(hour.size), + ) + }) + }; ReplicationSiteMetricsSnapshot { uptime: metrics.uptime, queued_curr_count: metrics.queued.curr.count, @@ -859,6 +893,12 @@ impl StorageReplicationStatsHandle { proxy_delete_tag_failed: metrics.proxied.delete_tag_failed, replica_size: metrics.replica_size, replica_count: metrics.replica_count, + failed_count, + failed_bytes, + failed_last_minute_count, + failed_last_minute_bytes, + failed_last_hour_count, + failed_last_hour_bytes, } } @@ -899,6 +939,12 @@ pub(crate) struct ReplicationSiteMetricsSnapshot { pub(crate) proxy_delete_tag_failed: i64, pub(crate) replica_size: i64, pub(crate) replica_count: i64, + pub(crate) failed_count: i64, + pub(crate) failed_bytes: i64, + pub(crate) failed_last_minute_count: i64, + pub(crate) failed_last_minute_bytes: i64, + pub(crate) failed_last_hour_count: i64, + pub(crate) failed_last_hour_bytes: i64, } pub(crate) async fn get_local_server_property() -> rustfs_madmin::ServerProperties { @@ -2043,13 +2089,32 @@ pub(crate) async fn init_compression_total_memory_from_backend(store: Arc); + let mut bucket = ecstore_bucket::replication::BucketReplicationStats::new(); + bucket.stats.insert("arn:replication::remote:photos".to_string(), target); + stats.inner.cache.write().await.insert("photos".to_string(), bucket); + + let snapshot = stats.site_metrics_snapshot().await; + + assert_eq!(snapshot.failed_count, 1); + assert_eq!(snapshot.failed_bytes, 2048); + assert_eq!(snapshot.failed_last_minute_count, 1); + assert_eq!(snapshot.failed_last_minute_bytes, 2048); + assert_eq!(snapshot.failed_last_hour_count, 1); + assert_eq!(snapshot.failed_last_hour_bytes, 2048); + } + #[tokio::test] async fn bucket_target_metadata_locks_serialize_only_matching_shards() { let bucket = "bucket-target-lock"; diff --git a/scripts/check_s3s_footprint.sh b/scripts/check_s3s_footprint.sh index 3bbebafb3..d8d504314 100755 --- a/scripts/check_s3s_footprint.sh +++ b/scripts/check_s3s_footprint.sh @@ -55,8 +55,11 @@ cd "$(dirname "$0")/.." # now reports an unreadable configuration as a plain string instead of raising # an S3 error per arm (24 invocation lines removed from # rustfs/src/admin/handlers/bucket_meta.rs; measured after merging the two). +# 1589 -> 1588 on 2026-09-08: the GA blocker set (rustfs/backlog#2366) added +# three invocation lines to the endpoint-refresh paths and folded the five +# copies of the concurrent-change error into one constructor, netting -1. S3S_IMPORT_FILES_BASELINE=213 -S3_ERROR_LINES_BASELINE=1589 +S3_ERROR_LINES_BASELINE=1588 # ecstore-scoped ratchet (rustfs/backlog#1842): the storage engine must not # know S3 wire/DTO types (ARCHITECTURE.md invariant 4). The S3-*consuming* # client was extracted to crates/s3-client, where s3s usage is legitimate;