diff --git a/.github/workflows/rustfs-performance-test.yml b/.github/workflows/rustfs-performance-test.yml
index d698b27e0..699b583cf 100644
--- a/.github/workflows/rustfs-performance-test.yml
+++ b/.github/workflows/rustfs-performance-test.yml
@@ -82,6 +82,12 @@ jobs:
performance-test:
runs-on: pf-testing
timeout-minutes: 900
+ env:
+ RUSTFS_BENCH_SCRIPT: ${{ github.workspace }}/auto-testing/rustfs_performance_testing.sh
+ RUSTFS_WARP_METHODS: ${{ inputs.test_method }}
+ RUSTFS_WARP_SIZES: ${{ inputs.object_size }}
+ RUSTFS_WARP_DURATION: ${{ inputs.warp_duration || '5m' }}
+ RUSTFS_WARP_CONCURRENCY: ${{ inputs.warp_concurrency || '64' }}
# Run on manual dispatch, or when the nightly build completed successfully.
# Skipped when nightly failed.
if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
@@ -158,19 +164,15 @@ jobs:
- name: Run benchmark (GET/PUT/MIXED)
id: benchmark
run: |
- # Empty on automatic (workflow_run) runs -> full 30 rounds.
- # Manual dispatch can restrict method(s)/size(s).
- export WARP_METHODS="${{ inputs.test_method }}"
- export WARP_SIZES="${{ inputs.object_size }}"
./auto-testing/rustfs_performance_test.sh \
--step 5 -y \
- --warp-duration "${{ inputs.warp_duration || '5m' }}" \
- --warp-concurrency "${{ inputs.warp_concurrency || '64' }}" \
--log-file "${LOG_FILE}"
- name: Analyze results
if: ${{ steps.benchmark.conclusion == 'success' }}
run: |
+ export WARP_METHODS="${RUSTFS_WARP_METHODS}" WARP_SIZES="${RUSTFS_WARP_SIZES}"
+ export WARP_DURATION="${RUSTFS_WARP_DURATION}" WARP_CONCURRENCY="${RUSTFS_WARP_CONCURRENCY}"
./auto-testing/rustfs_performance_test.sh --step 6 -y --log-file "${LOG_FILE:-/dev/null}"
- name: Collect RustFS version info
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 6963f7084..588fb7799 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/README.md b/scripts/README.md
index 50f8ea7a8..7ff3c9701 100644
--- a/scripts/README.md
+++ b/scripts/README.md
@@ -51,6 +51,7 @@ their issue closes.
| `e2e-run.sh` | ci-gate | Boots a rustfs server and runs the `s3s-e2e` black-box conformance tool against it | ci.yml `e2e-tests` jobs; `docs/testing/README.md` |
| `run_ecstore_validation_suite.sh` | dev-tool | ecstore black-box validation suite (`quick`/`full`/`destructive`/`fuzz` profiles) | `docs/testing/README.md`, `docs/testing/ecstore-validation-suite-design.md` |
| `run_e2e_tests.sh` | dev-tool | Local `e2e_test` crate runner (starts a server, applies filters, cleans up) | `crates/e2e_test/README.md` |
+| `run_scanner_heal_g09_upgrade_evidence.sh` | dev-tool | Runs the Scanner/Heal G09 mixed-version and rollback upgrade evidence lanes against the pinned previous Linux release binary | e2e-upgrade workflow constants; backlog#2269 |
| `run.sh` | dev-tool | Local rustfs startup wrapper | `make e2e-server`; Justfile |
| `run.ps1` | dev-tool | Windows counterpart of `run.sh` | — |
| `probe.sh` | dev-tool | Probe-style e2e run | `make probe-e2e` |
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;
diff --git a/scripts/run_scanner_heal_g09_upgrade_evidence.sh b/scripts/run_scanner_heal_g09_upgrade_evidence.sh
new file mode 100755
index 000000000..6927ca425
--- /dev/null
+++ b/scripts/run_scanner_heal_g09_upgrade_evidence.sh
@@ -0,0 +1,159 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
+UPGRADE_WORKFLOW="${PROJECT_ROOT}/.github/workflows/e2e-upgrade.yml"
+
+SOURCE_DIR=""
+OUT_DIR=""
+SKIP_BUILD="false"
+
+usage() {
+ cat <<'USAGE'
+Usage:
+ scripts/run_scanner_heal_g09_upgrade_evidence.sh [--source-dir ] [--out-dir ] [--skip-build]
+
+Runs the Scanner/Heal G09 upgrade evidence lanes against the pinned previous
+Linux x86_64 RustFS release used by the e2e-upgrade workflow:
+
+ - rolling mixed-version reader/writer evidence
+ - rollback payload replay evidence
+
+The script builds the current PR head by default, downloads and verifies the
+pinned previous release binary, runs the ignored e2e tests, and fails unless all
+G09 JSON evidence artifacts are present and non-empty.
+USAGE
+}
+
+die() {
+ echo "ERROR: $*" >&2
+ exit 1
+}
+
+parse_args() {
+ while [[ $# -gt 0 ]]; do
+ case "$1" in
+ --source-dir)
+ [[ $# -ge 2 ]] || die "--source-dir requires a value"
+ SOURCE_DIR="$2"
+ shift 2
+ ;;
+ --out-dir)
+ [[ $# -ge 2 ]] || die "--out-dir requires a value"
+ OUT_DIR="$2"
+ shift 2
+ ;;
+ --skip-build)
+ SKIP_BUILD="true"
+ shift
+ ;;
+ -h|--help)
+ usage
+ exit 0
+ ;;
+ *)
+ die "unknown arg: $1"
+ ;;
+ esac
+ done
+}
+
+workflow_env_value() {
+ local key="$1"
+ awk -v key="${key}:" '$1 == key { gsub(/["'\'']/, "", $2); print $2; exit }' "$UPGRADE_WORKFLOW"
+}
+
+require_linux_x86_64() {
+ local kernel machine
+ kernel="$(uname -s)"
+ machine="$(uname -m)"
+ [[ "$kernel" == "Linux" ]] || die "G09 upgrade evidence uses the pinned Linux release binary; run this on Linux"
+ [[ "$machine" == "x86_64" || "$machine" == "amd64" ]] || die "G09 upgrade evidence requires x86_64/amd64"
+}
+
+source_binary_path() {
+ printf '%s/rustfs\n' "$SOURCE_DIR"
+}
+
+ensure_previous_release_binary() {
+ local version asset sha archive url binary
+ version="${UPGRADE_SOURCE_VERSION:-$(workflow_env_value UPGRADE_SOURCE_VERSION)}"
+ asset="${UPGRADE_SOURCE_ASSET:-$(workflow_env_value UPGRADE_SOURCE_ASSET)}"
+ sha="${UPGRADE_SOURCE_SHA256:-$(workflow_env_value UPGRADE_SOURCE_SHA256)}"
+ [[ -n "$version" ]] || die "UPGRADE_SOURCE_VERSION is missing"
+ [[ -n "$asset" ]] || die "UPGRADE_SOURCE_ASSET is missing"
+ [[ -n "$sha" ]] || die "UPGRADE_SOURCE_SHA256 is missing"
+
+ SOURCE_DIR="${SOURCE_DIR:-${PROJECT_ROOT}/target/scanner-heal-g09-source/${version}}"
+ mkdir -p "$SOURCE_DIR"
+
+ binary="$(source_binary_path)"
+ if [[ -x "$binary" ]]; then
+ return
+ fi
+
+ archive="${SOURCE_DIR}/${asset}"
+ url="https://github.com/rustfs/rustfs/releases/download/${version}/${asset}"
+ curl --fail --location --retry 3 --output "$archive" "$url"
+ echo "${sha} ${archive}" | sha256sum --check --strict
+ unzip -q -o "$archive" -d "$SOURCE_DIR"
+ chmod +x "$binary"
+ [[ -x "$binary" ]] || die "downloaded archive did not provide an executable rustfs binary"
+}
+
+build_current_rustfs() {
+ if [[ "$SKIP_BUILD" == "true" ]]; then
+ return
+ fi
+ "${CARGO:-cargo}" build --locked -p rustfs --bin rustfs
+ : > "${PROJECT_ROOT}/target/debug/rustfs.features"
+}
+
+prepare_output_dir() {
+ local stamp
+ stamp="$(date -u +%Y%m%dT%H%M%SZ)"
+ OUT_DIR="${OUT_DIR:-${PROJECT_ROOT}/target/scanner-heal-g09-evidence/${stamp}}"
+ mkdir -p "$OUT_DIR"
+}
+
+run_case() {
+ local artifact="$1"
+ local test_name="$2"
+ shift 2
+ local case_dir="${OUT_DIR}/${artifact}"
+ mkdir -p "$case_dir"
+ RUSTFS_UPGRADE_SOURCE_BINARY="$(source_binary_path)" \
+ RUSTFS_E2E_LOG_DIR="${OUT_DIR}/server-logs/${artifact}" \
+ RUSTFS_SCANNER_HEAL_G09_EVIDENCE_DIR="$case_dir" \
+ "${CARGO:-cargo}" test --locked -p e2e_test \
+ "upgrade_compatibility_test::${test_name}" \
+ -- --ignored --exact --nocapture
+
+ local expected
+ for expected in "$@"; do
+ [[ -s "${case_dir}/${expected}" ]] || die "missing non-empty ${artifact}/${expected}"
+ done
+}
+
+main() {
+ parse_args "$@"
+ [[ -f "$UPGRADE_WORKFLOW" ]] || die "missing e2e-upgrade workflow"
+ require_linux_x86_64
+ cd "$PROJECT_ROOT"
+ ensure_previous_release_binary
+ build_current_rustfs
+ prepare_output_dir
+ run_case \
+ "mixed-version-upgrade" \
+ "rolling_upgrade_from_rc2_preserves_mixed_version_contracts" \
+ "G09-mixed_version_reader_evidence.json" \
+ "G09-mixed_version_writer_evidence.json"
+ run_case \
+ "bucket-config-rollback" \
+ "rollback_to_previous_release_reads_current_bucket_metadata" \
+ "G09-rollback_payload_evidence.json"
+ echo "PASS: Scanner/Heal G09 upgrade evidence written under ${OUT_DIR}"
+}
+
+main "$@"
diff --git a/scripts/test_security_workflow.py b/scripts/test_security_workflow.py
index ba483571f..e9c70f44c 100644
--- a/scripts/test_security_workflow.py
+++ b/scripts/test_security_workflow.py
@@ -837,6 +837,55 @@ emit_step_result() {
self.assertIn(value, contents)
self.assertNotIn("OLD RUN EVIDENCE", contents)
+ def test_performance_commands_bind_runner_selection_and_preserve_failures(self) -> None:
+ self.prepare("performance")
+ source = self.source.splitlines()
+ job = yaml_block(source, "performance-test", 2)
+ runner = WorkflowSteps()
+ runner.directory = self.directory / "workspace with spaces"
+ scripts = runner.directory / "auto-testing"
+ scripts.mkdir(parents=True)
+ wrapper = scripts / "rustfs_performance_test.sh"
+ wrapper.write_text(f"#!{sys.executable}\nimport json, os, sys\n" +
+ "print(json.dumps({'args': sys.argv[1:], 'env': {key: os.environ.get(key) for key in " +
+ "('RUSTFS_BENCH_SCRIPT', 'RUSTFS_WARP_METHODS', 'RUSTFS_WARP_SIZES', " +
+ "'RUSTFS_WARP_DURATION', 'RUSTFS_WARP_CONCURRENCY', 'WARP_METHODS', " +
+ "'WARP_SIZES', 'WARP_DURATION', 'WARP_CONCURRENCY')}}))\n" +
+ "sys.exit(int(os.environ['FAKE_BENCH_EXIT']))\n")
+ wrapper.chmod(0o755)
+ runner.steps = named_steps(job)
+ for methods, sizes, duration, concurrency in (
+ ("get", "1KiB", "1s", "7"), ("all", "all", "5m", "64"), ("", "", "5m", "64")
+ ):
+ runner.context = {"github.workspace": str(runner.directory), "inputs.test_method": methods,
+ "inputs.object_size": sizes, "inputs.warp_duration || '5m'": duration,
+ "inputs.warp_concurrency || '64'": concurrency}
+ runner.env = {**self.env, "RUSTFS_BENCH_SCRIPT": "/unverified/home-script.sh",
+ "RUSTFS_WARP_METHODS": "put", "RUSTFS_WARP_SIZES": "64MiB",
+ "RUSTFS_WARP_DURATION": "99h", "RUSTFS_WARP_CONCURRENCY": "2",
+ "WARP_DURATION": "88h", "WARP_CONCURRENCY": "3", "WARP_METHODS": "mixed", "WARP_SIZES": "32MiB",
+ "LOG_FILE": str(self.directory / "suite.log")}
+ runner.env.update(runner.step_env(job, indent=4))
+ for step, number in (("Run benchmark (GET/PUT/MIXED)", "5"), ("Analyze results", "6")):
+ for code in (0, 42):
+ with self.subTest(methods=methods, sizes=sizes, step=step, exit=code):
+ runner.env["FAKE_BENCH_EXIT"] = str(code)
+ result = runner.run_step(step)
+ self.assertEqual(result.returncode, code, result.stderr)
+ invocation = json.loads(result.stdout)
+ expected = ["--step", number, "-y", "--log-file", runner.env["LOG_FILE"]]
+ self.assertEqual(invocation["args"], expected)
+ self.assertEqual(invocation["env"]["RUSTFS_BENCH_SCRIPT"], str(scripts / "rustfs_performance_testing.sh"))
+ self.assertEqual(invocation["env"]["RUSTFS_WARP_METHODS"], methods)
+ self.assertEqual(invocation["env"]["RUSTFS_WARP_SIZES"], sizes)
+ self.assertEqual(invocation["env"]["RUSTFS_WARP_DURATION"], duration)
+ self.assertEqual(invocation["env"]["RUSTFS_WARP_CONCURRENCY"], concurrency)
+ if number == "6":
+ self.assertEqual(invocation["env"]["WARP_METHODS"], methods)
+ self.assertEqual(invocation["env"]["WARP_SIZES"], sizes)
+ self.assertEqual(invocation["env"]["WARP_DURATION"], duration)
+ self.assertEqual(invocation["env"]["WARP_CONCURRENCY"], concurrency)
+
if __name__ == "__main__":
unittest.main()