Compare commits

..

14 Commits

Author SHA1 Message Date
Chris 2b243d8659 fix(scanner): clear transient cycle recovery on follower lock contention (1.0.2 backport) (#8596)
fix(scanner): clear transient cycle recovery on follower lock contention (#8566)

(cherry picked from commit 606d8a3a8a)

Co-authored-by: Ramakrishna Chilaka <49393831+RamakrishnaChilaka@users.noreply.github.com>
2026-10-16 04:05:23 +08:00
Chris c52570d385 fix(rebalance): wait for readiness and admit peer workers (1.0.2 backport) (#8593)
fix(rebalance): wait for readiness and admit peer workers (#8570)

* fix(rebalance): avoid contention when peers join the same run

* fix(rebalance): wait for fleet proof during initialization

* test(rebalance): check contextualized competing run errors

* fix(rebalance): retain typed worker admission lock errors

* test(rebalance): retry the init readiness 503 and pin the retry error shape (#8586)

The e2e harness only retried the pre-fix 500 carrying the fleet-proof
marker; the init stage now answers 503 + Retry-After after its own
readiness budget, so start_rebalance treats that as the same wait. The
missing-proof regression also asserts the typed error shape that the admin
handler's retry predicate depends on, not only the rendered message.

---------


(cherry picked from commit 7682a12b58)

Co-authored-by: cxymds <cxymds@gmail.com>
2026-10-15 18:11:29 +00:00
Chris e0b1635abf fix(multipart): retain commit guards and reclaim orphan staging (1.0.2 backport) (#8592)
fix(multipart): retain commit guards across request cancellation (#8571)

* fix(multipart): retain commit guards across request cancellation

* fix(multipart): preserve commit task error context

* fix(multipart): reclaim verified orphan upload staging (#8572)

* fix(multipart): reclaim verified orphan upload staging

* fix(multipart): share orphan guards across matching lock domains

* fix(multipart): preserve orphan cleanup task error context

* fix(ecstore): box guarded delete transition identity

* fix(multipart): keep NoSuchUpload when orphan recovery cannot prove absence (#8587)

Recovery is a side path of an idempotent Abort. An offline disk, lock
contention, partially overlapping lock domains or unreadable bucket
metadata all stop the staging-absence proof before any deletion; those
used to propagate as 503/500, where the base answered NoSuchUpload. They
now keep that answer and delete nothing; only failures inside the
deletion task remain request failures, because a retry can finish them.

---------


(cherry picked from commit 65f30bd5c1)

Co-authored-by: cxymds <cxymds@gmail.com>
2026-10-15 18:00:45 +00:00
Chris f0f2b0cf48 fix(replication): skip the replica ETag audit for encrypted sources (1.0.2 backport) (#8580)
Backport of #8573 (main commit 7a32ca2adb) to release/1.0.2. The five touched files were identical between release/1.0.2 and the main base the fix was written against, so this applied without conflicts.

New encrypted writes store a keyed HMAC fingerprint as the object ETag
(docs/operations/encrypted-etags.md). Replication forwards that ETag on
x-rustfs-source-etag, but a target that computes its own ETag (AWS S3 and
most compatible stores) answers the plaintext PUT with the MD5 of what it
received. Both values are 32 hex characters, so verify_single_part_replica
treated the fingerprint as a comparable MD5 and failed every single-part
SSE-S3 / SSE-KMS replica with "replica etag mismatch"; MRF and resync then
retried the object forever. RustFS and MinIO targets adopt the source ETag
and were unaffected.

Exempt sources whose stored ETag cannot describe the wire bytes: any
managed-SSE or SSE-C source, decided by the existing replication
source-encryption classifier. The checksum header on the replication PUT
remains the integrity check for those objects; plaintext sources keep the
audit unchanged.

Claude-Session: https://claude.ai/code/session_01CezCdjE9AdShFUi9tNk797


(cherry picked from commit 7a32ca2adb)

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-15 22:47:37 +08:00
Chris 0882e99c32 fix(s3): make an unsupported delete participant retryable and explain it (1.0.2 backport) (#8577)
Backport of #8574 (main commit d7156290d5) to release/1.0.2.

A user-bucket delete fans out through the delete-transaction RPC. When a
participant answers Unimplemented (a peer still running a release without
that RPC, or a protocol-version mismatch) the coordinator reports
StorageError::UnsupportedDisk, which had no S3 mapping: clients received a
500 InternalError whose message was the variant's Display text,
"drive does not support O_DIRECT". That text describes the startup drive
probe and sends an operator mid-upgrade after the wrong cause.

Map UnsupportedDisk to 503 ServiceUnavailable with a constant message that
names the protocol-version cause and the remedy (finish upgrading every
node, then retry). The typed source is retained for diagnostics and no
Retry-After is added, since retrying the same node does not help until the
cluster upgrade completes.

Conflict resolution against release/1.0.2: the DiskFull arm and the
capacity-failure test from #8553 are not on this branch and were not
carried over; only the UnsupportedDisk mapping, its message, the mapping
table row and the new unit test are included.

(cherry picked from commit d7156290d5)


Claude-Session: https://claude.ai/code/session_01CezCdjE9AdShFUi9tNk797

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-15 22:01:36 +08:00
Chris 802ac8a1bd test(heal): avoid needless collect in legacy root recovery test (1.0.2 backport) (#8551)
test(heal): avoid needless collect in legacy root recovery test
2026-10-14 08:22:12 -07:00
Chris 9408b90f13 chore(release): prepare 1.0.2 (#8549)
Co-authored-by: Hauser <housemecn@gmail.com>
2026-10-14 22:12:55 +08:00
Hauser 08ac1ec218 fix(s3): advise retry after typed lock timeouts (#8546)
Add a one-second Retry-After hint at the canonical S3 error conversion
only for ServiceUnavailable responses retaining a typed lock timeout.
Keep existing error codes, source chains, and lock budgets unchanged.

Cover actual HTTP/XML serialization, nested causes, private-resource
redaction, and unrelated disk/quorum/text failures. Document the response
contract and the limits of log-count and five-second timeout attribution.

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-10-14 20:52:14 +08:00
cxymds 388d1f17e6 fix(heal): safely recover missing bucket directories by default (#8517)
* fix(heal): safely recover missing bucket directories by default

* fix(heal): allow bound admin recovery before namespace checks
2026-10-14 03:57:09 -07:00
Henry Guo 22c093e66b feat(table-catalog): add opt-in periodic maintenance queueing (#8537) 2026-10-14 03:55:36 -07:00
Chris 24ab34a5ee fix(ci): restore main lint and test reliability (#8543)
* fix(ci): remove needless returns from Linux disk helpers

* fix(test): verify encrypted purge replicas by version body

* fix(test): expire namespace timeout after fixture preparation

* fix(ci): remove remaining needless collection and borrows

* fix(test): wait for active admin heal at interruption checkpoint
2026-10-14 18:28:10 +08:00
Hauser 9068d83400 fix(heal): unblock automatic replacement recovery (#8533)
* fix(heal): unblock automatic replacement recovery

Repair bucket publication directories before adopting survivor delete proofs.
Keep explicit empty-volume removal after proof settlement and preserve the
admitted bucket incarnation for administrator replay.

Discover unfinished intents across every local set and resume the durable
owner with its exact target scope before admitting new sibling replacements.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(heal): route scanner fixtures through the storage boundary

Expose the existing disk-layout bootstrap symbols as test-only aliases in
heal::storage_api and use that boundary from scanner recovery regressions.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* refactor(heal): index replacement discovery within each scan

Keep a single endpoint-keyed disk inventory with first-snapshot precedence
and deterministic discovery order. Index local target lookups and move
resume state into generation candidates instead of cloning it.

Retain fresh durable reads and the existing conflict and mutation fences.
Cover duplicate disk handles, endpoint uniqueness and discovery ordering.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(heal): retain committed journals in recovery fixtures

Build one certified survivor delete decision for both replacement and
empty-volume removal tests. Clear target contents without moving its root
inode so Linux disk handles observe the same empty target. Drain PUT tails
and assert the proof remains before the real Heal entry.

Expose the existing recovery state through the ECStore disk API for typed
fixture checks.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(heal): use server stack for replacement regression

The replacement regression polls the composed Heal future on libtest's
smaller calling-thread stack and aborts before reporting an assertion.
Run it on a bounded 8 MiB test thread and Tokio workers, matching the
debug server's stack budget. Keep all storage and object assertions intact.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-10-14 17:04:36 +08:00
Hauser 20fc897036 chore(deps): refresh crates and use Pyroscope 2.1.2 (#8540)
* chore(deps): refresh crates and use Pyroscope 2.1.2

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* chore(deps): refresh lockfile

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-10-14 16:13:57 +08:00
Hauser a86117590c fix(ci): prepare pinned Compose for Connect acceptance (#8531)
* fix(ci): prepare pinned Compose for Connect acceptance

Install the verified Compose CLI through one local action shared by all 11 connected acceptance entry points. Publish only checksum-verified binaries and keep daemon, provenance and timeout gates unchanged.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(ci): use daemon-visible Connect temporary inputs

Match the existing official Connect acceptance TMPDIR contract in all 11 real E2E invocations so daemon bind mounts can read generated inputs. Preserve key modes, read-only mounts and cleanup checks.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-10-14 10:42:35 +08:00
318 changed files with 9223 additions and 25209 deletions
+2 -2
View File
@@ -1,2 +1,2 @@
sha256-darwin=ff63177a6280d52ac29f4a14a0c1f7fdcf7cecd442c1f3d3bce0bbdffc0552a5
sha256-linux=2cc8bb449f52bdf5d9e88f75dc5c3a94df121a55aa436d5ec95311d42ec5fa0b
sha256-darwin=1520cd3027c3a6924afc6a204eddd5475bc30f0109d4c2847456a66177263244
sha256-linux=26155b7f1c5bf6bf596ff24a4136e6767ea4ee6461c32db7fbe1400791da2c28
+2 -2
View File
@@ -1,2 +1,2 @@
sha256-darwin=8ef80b7353d1e0c4f863a2aca7e996d23858b8c32b9edf8fd272076fa63a83d8
sha256-linux=99fe1cd128ade2df69b9e3ecf94b91021185a9769293420b1ba5d24760f2ea99
sha256-darwin=347a0d4e5e547d5388703062ce66a66167a845b6e57ec591ccb4c732a567c68d
sha256-linux=4f9efa47194279185bb68249e82c788fa147abe944b6c764050e3d47f31e7926
+1 -1
View File
@@ -1 +1 @@
sha256=bf77250d94604db7ba094c5d63bf8a284beb872983871bcfd0dc02e46ff53326
sha256=ba8f392b617cf9f2338142233a417e43927b28096635e628233ce8a60cf49d10
-5
View File
@@ -70,11 +70,6 @@ s3s-footprint-check: ## Check the s3s dependency footprint ratchet stays frozen
@echo "📦 Checking s3s footprint ratchet..."
./scripts/check_s3s_footprint.sh
.PHONY: test-count-ratchet-check
test-count-ratchet-check: ## Check no workspace crate lost tests or gained #[ignore] against the committed baseline
@echo "🧪 Checking per-crate test-count ratchet..."
./scripts/check_test_count_ratchet.sh
.PHONY: fips-wording-check
fips-wording-check: ## Check docs and crates/kms do not over-claim crypto capabilities
@echo "📣 Checking cryptographic capability wording guard..."
+2 -2
View File
@@ -20,11 +20,11 @@ planning-docs-check: ## Check that no planning-type documents are committed
./scripts/check_no_planning_docs.sh
.PHONY: pre-commit
pre-commit: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check error-other-ratchet-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check test-count-ratchet-check fips-wording-check embedded-secrets-check test-wiring-check doc-paths-check planning-docs-check quick-check ## Run fast pre-commit checks without clippy/full tests
pre-commit: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check error-other-ratchet-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check embedded-secrets-check test-wiring-check doc-paths-check planning-docs-check quick-check ## Run fast pre-commit checks without clippy/full tests
@echo "✅ All pre-commit checks passed!"
.PHONY: pre-pr
pre-pr: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check error-other-ratchet-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check test-count-ratchet-check fips-wording-check embedded-secrets-check test-wiring-check doc-paths-check planning-docs-check log-analyzer-rules-check offline-enrollment-e2e-check clippy-check test ## Run full pre-PR checks with clippy and tests
pre-pr: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check error-other-ratchet-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check embedded-secrets-check test-wiring-check doc-paths-check planning-docs-check log-analyzer-rules-check offline-enrollment-e2e-check clippy-check test ## Run full pre-PR checks with clippy and tests
@echo "✅ All pre-PR checks passed!"
.PHONY: dev-check
-2
View File
@@ -50,9 +50,7 @@ script-tests: ## Run shell script tests
./scripts/test_fuzz_runner.sh
./scripts/test_python_bin.sh
./scripts/check_embedded_secrets.sh --self-test
./scripts/check_s3s_footprint.sh --self-test
$(RUSTFS_PYTHON_BIN) ./scripts/check_test_wiring.py --self-test
$(RUSTFS_PYTHON_BIN) ./scripts/codemods/s3_error_to_s3_types.py --self-test
$(RUSTFS_PYTHON_BIN) ./scripts/test_e2e_binary.py
$(RUSTFS_PYTHON_BIN) ./scripts/test_release_build_workflow.py
$(RUSTFS_PYTHON_BIN) ./scripts/test_release_glibc.py
-9
View File
@@ -599,14 +599,6 @@ test-group = 'ecstore-serial-flaky'
# the PR-lane e2e coverage for scanner usage snapshots consumed by quota and
# admin surfaces. It uses the same single-node, random-port, isolated-temp-dir
# fixture as the existing smoke modules.
#
# Dual-stack differential (rustfs/backlog#2740): `dual_stack_differential::`
# starts two single-node servers from the same binary, one per
# RUSTFS_S3_STACK value, on random ports with isolated temp dirs, and runs one
# SDK script covering every operation of `impl S3 for FS` against both. Its
# comparator unit tests need no server. It meets the admission criteria
# unchanged; the module clause admits a new differential test without another
# filter edit.
[profile.e2e-smoke]
default-filter = """
package(e2e_test) & (
@@ -614,7 +606,6 @@ default-filter = """
| test(/^replication_extension_test::(test_replication_check_succeeds_with_remote_target|test_replication_check_rejects_target_without_object_lock|test_set_remote_target_rejects_unversioned_source_bucket|test_replication_check_rejects_unversioned_source_bucket|test_replication_check_rejects_missing_replication_config|test_replication_check_rejects_invalid_bucket|test_set_remote_target_rejects_same_bucket_on_same_deployment|test_set_remote_target_rejects_unversioned_target_bucket|test_set_remote_target_update_requires_arn|test_set_remote_target_update_rejects_missing_target|test_set_remote_target_rejects_invalid_target_url|test_set_remote_target_rejects_self_signed_https_target_without_skip_tls_verify|test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_pem|test_list_remote_targets_rejects_empty_bucket|test_list_remote_targets_rejects_invalid_bucket|test_remove_remote_target_rejects_missing_target|test_remove_remote_target_rejects_missing_arn|test_remove_remote_target_rejects_invalid_bucket|test_remove_remote_target_rejects_target_used_by_replication|test_delete_bucket_replication_removes_remote_target)$/)
| test(/^reliant::lifecycle::/)
| test(/^reliant::tiering::/)
| test(/^dual_stack_differential::/)
| test(/^upgrade_compatibility_test::upgrade_write_readiness_tests::/)
| test(/^on_demand_migration::(get_basic_test::(get_miss_pulls_inline_and_serves_locally_afterwards|head_miss_answers_from_the_source_without_persisting)|interaction_test::test_odm_admin_config_is_redacted_and_status_counts_match_the_source)$/)
)
-37
View File
@@ -1,37 +0,0 @@
# Files that may still reference s3s (`s3s::` or `s3s_sigv4`) when Phase 1 of
# the gateway migration ends (rustfs/backlog#2734 task T0.6, rustfs/backlog#2741).
# Consumed only by `scripts/check_s3s_footprint.sh --mode allowlist`.
#
# One entry per line: a repository-relative file path or a git `:(glob)`
# pathspec. `**` matches across directories, `*` stays within one path
# component, and a directory entry matches everything under it. Blank lines
# and lines starting with `#` are ignored; an entry may not contain whitespace,
# so a note belongs on its own comment line.
#
# Every Rust file listed by `git grep -l '\bs3s\(::\|_sigv4\)' -- '*.rs'` that
# no entry matches fails the allowlist mode. Task T1.9 switches CI to that
# mode; until then `--dry-run` prints the files still outside this list, which
# is the Phase 1 tracking list.
#
# Never add crates/** other than crates/e2e_test: the engine crates leave s3s
# entirely, and the script rejects such an entry. Only the s3s-facing edge of
# the rustfs binary and the e2e oracle keep it.
# rustfs binary: the s3s-facing edge that survives Phase 1
rustfs/src/storage/ecfs.rs
rustfs/src/storage/s3_compat/**
rustfs/src/storage/access.rs
rustfs/src/auth.rs
rustfs/src/error.rs
rustfs/src/storage_api.rs
rustfs/src/app/storage_api.rs
rustfs/src/admin/router.rs
rustfs/src/app/metadata_route.rs
rustfs/src/server/http.rs
rustfs/src/server/layer.rs
rustfs/src/server/hybrid.rs
rustfs/src/protocols/client.rs
# e2e oracle: the test target that verifies S3 behaviour against s3s
crates/e2e_test/src/fake_s3_target/**
crates/e2e_test/src/**
-17
View File
@@ -1,17 +0,0 @@
# Exact baseline for the direct s3s footprint, consumed only by
# scripts/check_s3s_footprint.sh (rustfs/backlog#1677; rustfs/backlog#2734 T0.4).
#
# Each value is the exact measurement of the tree, not a ceiling. CI fails when a
# measured counter differs from its line here in either direction:
# - a larger measurement means new code widened the s3s surface: route it
# through the gateway abstractions instead; never raise a value here;
# - a smaller measurement means the PR shrank the surface: lower the matching
# value to the measured number in the same PR.
#
# files Rust files referencing s3s paths, repo-wide minus crates/e2e_test
# s3_error_lines lines holding s3_error!, same scope, minus those the script
# header's rule clears (rustfs_s3_types's macro); fails closed
# ecstore_files Rust files referencing s3s paths under crates/ecstore/src
files=194
s3_error_lines=1565
ecstore_files=36
-63
View File
@@ -1,63 +0,0 @@
# Per-crate test-count baseline consumed by scripts/check_test_count_ratchet.sh
# (rustfs/backlog#2734 section 1 rule 3; T0.3 in rustfs/backlog#2738).
#
# One row per workspace member: <package>, tests, ignored, tab-separated.
# tests counts #[test], #[tokio::test], #[tokio::test(...)] and #[test_case(...)]
# attribute lines under the package directory; ignored counts #[ignore] and
# #[ignore = "..."]. Both columns are exact: CI fails when a crate's tests fall
# below or rise above its row, or when its ignored count changes. Raise a row in
# the PR that adds tests; lowering one is a reviewed statement that tests were
# intentionally removed. Regenerate with:
# scripts/check_test_count_ratchet.sh --update
e2e_test 902 18
rustfs 5653 62
rustfs-audit 70 0
rustfs-checksums 26 0
rustfs-common 25 0
rustfs-concurrency 27 0
rustfs-config 43 0
rustfs-connect-symbol-catalog 6 0
rustfs-credentials 24 0
rustfs-crypto 67 0
rustfs-data-usage 54 0
rustfs-ecstore 6495 1
rustfs-extension-schema 23 0
rustfs-filemeta 306 0
rustfs-heal 843 3
rustfs-heal-contracts 6 0
rustfs-iam 390 0
rustfs-io-core 49 0
rustfs-io-metrics 127 0
rustfs-keystone 29 0
rustfs-kms 715 25
rustfs-license 2 0
rustfs-lifecycle 152 0
rustfs-lock 135 0
rustfs-log-analyzer 89 0
rustfs-madmin 209 0
rustfs-notify 136 0
rustfs-object-capacity 92 0
rustfs-object-data-cache 140 0
rustfs-obs 367 0
rustfs-policy 681 0
rustfs-protocols 736 1
rustfs-protos 63 0
rustfs-replication 200 0
rustfs-rio 200 0
rustfs-rio-v2 34 6
rustfs-s3-client 65 0
rustfs-s3-ops 8 0
rustfs-s3-types 97 0
rustfs-s3select-api 206 0
rustfs-s3select-query 188 2
rustfs-scanner 997 20
rustfs-scanner-metrics 81 0
rustfs-security-governance 20 0
rustfs-signer 31 0
rustfs-storage-api 31 0
rustfs-targets 603 26
rustfs-test-utils 1 0
rustfs-tls-runtime 43 0
rustfs-trusted-proxies 68 0
rustfs-utils 217 0
rustfs-zip 30 0
-4
View File
@@ -86,10 +86,6 @@ runs:
shell: bash
run: ./scripts/check_s3s_footprint.sh
- name: Check per-crate test-count ratchet
shell: bash
run: ./scripts/check_test_count_ratchet.sh
- name: Check cryptographic capability wording
shell: bash
run: ./scripts/check_fips_wording.sh
@@ -0,0 +1,39 @@
# Copyright 2024 RustFS Team
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
name: Set up Docker Compose
description: Install the verified Compose CLI used by native Linux x86_64 Connect acceptance
runs:
using: composite
steps:
- name: Install verified Docker Compose CLI plugin
shell: bash
run: |
set -euo pipefail
[[ $(uname -s) == Linux && $(uname -m) == x86_64 ]]
plugin_dir="${DOCKER_CONFIG:-$HOME/.docker}/cli-plugins"
mkdir -p "$plugin_dir"
compose_download=$(mktemp "${RUNNER_TEMP:?}/connect-compose.XXXXXX")
compose_staged=$(mktemp "$plugin_dir/.docker-compose.XXXXXX")
trap 'rm -f "$compose_download" "$compose_staged"' EXIT
curl --fail --location --retry 3 \
https://github.com/docker/compose/releases/download/v2.40.3/docker-compose-linux-x86_64 \
--output "$compose_download"
printf '%s %s\n' \
dba9d98e1ba5bfe11d88c99b9bd32fc4a0624a30fafe68eea34d61a3e42fd372 \
"$compose_download" | sha256sum --check --strict
install -m 0755 "$compose_download" "$compose_staged"
mv -f "$compose_staged" "$plugin_dir/docker-compose"
[[ $(docker compose version --short) == 2.40.3 ]]
-9
View File
@@ -20,7 +20,6 @@ on:
paths:
- '**/Cargo.toml'
- '**/Cargo.lock'
- 'third-party/pyroscope/**'
- 'deny.toml'
- '.github/actions/**'
- '.github/workflows/**'
@@ -40,7 +39,6 @@ on:
paths:
- '**/Cargo.toml'
- '**/Cargo.lock'
- 'third-party/pyroscope/**'
- 'deny.toml'
- '.github/actions/**'
- '.github/workflows/**'
@@ -131,13 +129,6 @@ jobs:
- name: Run cargo-deny
run: cargo deny check --hide-inclusion-graph advisories sources bans licenses
- name: Test patched profiling dependency
env:
CARGO_TARGET_DIR: ${{ github.workspace }}/target/pyroscope-dependency
run: >-
cargo test --manifest-path third-party/pyroscope/Cargo.toml
--features backend-pprof-rs --lib --test rustfs_zstd
workflow-pin-report:
name: Workflow Pin Report
if: github.repository == 'rustfs/rustfs' && (github.event_name != 'pull_request' || github.event.action != 'closed')
+5 -96
View File
@@ -16,9 +16,7 @@ name: Continuous Integration
on:
push:
# gateway/integration receives the daily merge of main as a direct push
# (rustfs/backlog#2734), so its pushes must prove the result like main's.
branches: [ main, gateway/integration ]
branches: [ main ]
paths-ignore:
- "**.md"
- "docs/**"
@@ -750,8 +748,6 @@ jobs:
needs: [ build-rustfs-debug-binary ]
runs-on: sm-standard-2
timeout-minutes: 30
env:
RUSTFS_S3_STACK: legacy
steps:
- name: Checkout repository
uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
@@ -870,105 +866,18 @@ jobs:
path: ${{ runner.temp }}/rustfs-e2e-*/rustfs.log
retention-days: 3
# The same e2e-smoke suite against the same debug binary with the gateway
# S3 stack selected (rustfs/backlog#2734). Report-only: it is outside the
# Test and Lint aggregate and becomes required at T2.10. The archive and run
# steps must stay identical to e2e-tests; scripts/ci_gate.py enforces that.
e2e-tests-gateway:
name: e2e-smoke (gateway, report-only until T2.10)
needs: [ build-rustfs-debug-binary ]
runs-on: sm-standard-2
timeout-minutes: 30
continue-on-error: true
env:
RUSTFS_S3_STACK: gateway
steps:
- name: Checkout repository
uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
with:
persist-credentials: false
- name: Setup Rust environment
uses: ./.github/actions/setup
with:
rust-version: stable
cache-shared-key: ci-dev
cache-save-if: 'false'
install-build-packaging-tools: 'false'
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
python-version: "3.12"
- name: Install awscurl
run: |
python3 -m pip install --user --upgrade pip "awscurl==0.44"
echo "AWSCURL_PATH=$HOME/.local/bin/awscurl" >> "$GITHUB_ENV"
- name: Verify awscurl
run: test -x "$AWSCURL_PATH"
- name: Download debug binary
uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
with:
name: rustfs-debug-binary
path: target/debug
- name: Make binary executable
run: chmod +x ./target/debug/rustfs
- name: Archive e2e smoke test binaries
env:
NEXTEST_ARCHIVE: ${{ runner.temp }}/rustfs-e2e-smoke.tar.zst
NEXTEST_LISTING: ${{ runner.temp }}/rustfs-e2e-smoke-list.json
run: |
cargo nextest archive --profile e2e-smoke -p e2e_test --archive-file "${NEXTEST_ARCHIVE}"
cargo nextest list --profile e2e-smoke --archive-file "${NEXTEST_ARCHIVE}" --message-format json > "${NEXTEST_LISTING}"
python3 ./scripts/check_test_wiring.py --check-profile e2e-smoke "${NEXTEST_LISTING}"
./scripts/check_security_smoke_count.sh check "${NEXTEST_LISTING}"
- name: Run e2e smoke suite
env:
NEXTEST_ARCHIVE: ${{ runner.temp }}/rustfs-e2e-smoke.tar.zst
RUSTFS_E2E_LOG_DIR: ${{ runner.temp }}/rustfs-e2e-smoke-logs
SAMPLER_INTERVAL_SECS: "15"
run: |
./scripts/ci/resource_sampler.sh start e2e-smoke
trap './scripts/ci/resource_sampler.sh stop' EXIT
python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-smoke --archive-file "${NEXTEST_ARCHIVE}" \
--status-level all --final-status-level all --failure-output final
- name: Upload e2e smoke diagnostics
if: always()
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
with:
name: e2e-smoke-gateway-diagnostics-${{ github.run_number }}
path: |
${{ runner.temp }}/rustfs-e2e-smoke-logs/
${{ runner.temp }}/rustfs-e2e-smoke-list.json
if-no-files-found: warn
- name: Upload e2e smoke JUnit report
if: always()
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
with:
name: e2e-smoke-gateway-junit-${{ github.run_number }}
path: target/nextest/e2e-smoke/junit.xml
if-no-files-found: warn
e2e-full:
name: End-to-End Tests (full merge gate)
# Merge gate only (backlog#1149 ci-5): the never-automated user-visible
# suites — KMS, object_lock, multipart_auth, quota, checksum, encryption,
# security-boundary, ... — via the e2e-full nextest profile. Too heavy for
# every PR, so it is gated to main and gateway/integration pushes, the merge
# queue, and manual dispatch. protocols / the 7 cluster suites / replication
# / #[ignore] are owned by other lanes (see .config/nextest.toml profile.e2e-full).
# every PR, so it is gated to main pushes, the merge queue, and manual
# dispatch. protocols / the 7 cluster suites / replication / #[ignore] are
# owned by other lanes (see .config/nextest.toml profile.e2e-full).
if: >-
github.event_name == 'workflow_dispatch' ||
github.event_name == 'merge_group' ||
(github.event_name == 'push' && (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/gateway/integration'))
(github.event_name == 'push' && github.ref == 'refs/heads/main')
needs: [ build-rustfs-debug-binary ]
runs-on: sm-standard-2
timeout-minutes: 55
@@ -57,6 +57,9 @@ jobs:
ref: ${{ inputs.connect_sha }}
token: ${{ secrets.PF_TESTING_GH_TOKEN }}
- name: Set up Docker Compose
uses: ./.github/actions/setup-docker-compose
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
@@ -167,7 +170,7 @@ jobs:
connect-harness/web/node_modules/.bin/playwright install --with-deps chromium
make -C connect-harness e2e-connected-dispatch-check
(cd connect-harness && ./tests/connectivity/diagnostic-schedule-evidence.test.sh)
RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
TMPDIR="$GITHUB_WORKSPACE" RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
RUSTFS_WORKTREE="$GITHUB_WORKSPACE/rustfs-source" \
CONNECT_E2E_DIAGNOSTIC_SCHEDULE_EVIDENCE="$GITHUB_WORKSPACE/diagnostic-schedule-evidence.json" \
make -C connect-harness e2e-connected E2E_SCENARIO=diagnostic-schedule
@@ -57,6 +57,9 @@ jobs:
ref: ${{ inputs.connect_sha }}
token: ${{ secrets.PF_TESTING_GH_TOKEN }}
- name: Set up Docker Compose
uses: ./.github/actions/setup-docker-compose
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
@@ -204,7 +207,7 @@ jobs:
connect-harness/web/node_modules/.bin/playwright install --with-deps chromium
make -C connect-harness e2e-connected-dispatch-check
(cd connect-harness && ./tests/connectivity/health-service-evidence.test.sh)
RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
TMPDIR="$GITHUB_WORKSPACE" RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
RUSTFS_WORKTREE="$GITHUB_WORKSPACE/rustfs-source" \
CONNECT_E2E_HEALTH_SERVICE_EVIDENCE="$GITHUB_WORKSPACE/health-service-evidence.json" \
make -C connect-harness e2e-connected E2E_SCENARIO=health-service
@@ -73,6 +73,9 @@ jobs:
ref: ${{ inputs.connect_sha }}
token: ${{ secrets.PF_TESTING_GH_TOKEN }}
- name: Set up Docker Compose
uses: ./.github/actions/setup-docker-compose
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
@@ -159,7 +162,7 @@ jobs:
npm --prefix connect-harness/web ci
connect-harness/web/node_modules/.bin/playwright install --with-deps chromium
make -C connect-harness e2e-connected-dispatch-check
RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
TMPDIR="$GITHUB_WORKSPACE" RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
RUSTFS_WORKTREE="$GITHUB_WORKSPACE/rustfs-source" \
CONNECT_E2E_PERFORMANCE_DRIVE_EVIDENCE="$GITHUB_WORKSPACE/performance-drive-service-job-evidence.json" \
make -C connect-harness e2e-connected E2E_SCENARIO=performance-drive
@@ -73,6 +73,9 @@ jobs:
ref: ${{ inputs.connect_sha }}
token: ${{ secrets.PF_TESTING_GH_TOKEN }}
- name: Set up Docker Compose
uses: ./.github/actions/setup-docker-compose
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
@@ -159,7 +162,7 @@ jobs:
npm --prefix connect-harness/web ci
connect-harness/web/node_modules/.bin/playwright install --with-deps chromium
make -C connect-harness e2e-connected-dispatch-check
RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
TMPDIR="$GITHUB_WORKSPACE" RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
RUSTFS_WORKTREE="$GITHUB_WORKSPACE/rustfs-source" \
CONNECT_E2E_PERFORMANCE_NETWORK_EVIDENCE="$GITHUB_WORKSPACE/performance-network-service-job-evidence.json" \
make -C connect-harness e2e-connected E2E_SCENARIO=performance-network
@@ -77,6 +77,9 @@ jobs:
ref: ${{ inputs.connect_sha }}
token: ${{ secrets.PF_TESTING_GH_TOKEN }}
- name: Set up Docker Compose
uses: ./.github/actions/setup-docker-compose
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
@@ -178,7 +181,7 @@ jobs:
npm --prefix connect-harness/web ci
connect-harness/web/node_modules/.bin/playwright install --with-deps chromium
make -C connect-harness e2e-connected-dispatch-check
RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
TMPDIR="$GITHUB_WORKSPACE" RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
RUSTFS_WORKTREE="$GITHUB_WORKSPACE/rustfs-source" \
CONNECT_E2E_CPU_CATALOG="$GITHUB_WORKSPACE/binary/rustfs.cpu-symbol-catalog.json" \
CONNECT_E2E_CPU_CATALOG_SHA256="$CATALOG_SHA256" \
@@ -57,6 +57,9 @@ jobs:
ref: ${{ inputs.connect_sha }}
token: ${{ secrets.PF_TESTING_GH_TOKEN }}
- name: Set up Docker Compose
uses: ./.github/actions/setup-docker-compose
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
@@ -203,7 +206,7 @@ jobs:
connect-harness/web/node_modules/.bin/playwright install --with-deps chromium
make -C connect-harness e2e-connected-dispatch-check
(cd connect-harness && ./tests/connectivity/profile-memory-evidence.test.sh)
RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
TMPDIR="$GITHUB_WORKSPACE" RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
RUSTFS_WORKTREE="$GITHUB_WORKSPACE/rustfs-source" \
CONNECT_E2E_PROFILE_MEMORY_EVIDENCE="$GITHUB_WORKSPACE/profile-memory-evidence.json" \
make -C connect-harness e2e-connected E2E_SCENARIO=profile-memory
@@ -81,6 +81,9 @@ jobs:
ref: ${{ inputs.connect_sha }}
token: ${{ secrets.PF_TESTING_GH_TOKEN }}
- name: Set up Docker Compose
uses: ./.github/actions/setup-docker-compose
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
@@ -206,7 +209,7 @@ jobs:
npm --prefix connect-harness/web ci
connect-harness/web/node_modules/.bin/playwright install --with-deps chromium
make -C connect-harness e2e-connected-dispatch-check
RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
TMPDIR="$GITHUB_WORKSPACE" RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
RUSTFS_WORKTREE="$GITHUB_WORKSPACE/rustfs-source" \
CONNECT_E2E_PROFILE_THREADS_EVIDENCE="$GITHUB_WORKSPACE/profile-threads-service-job-evidence.json" \
make -C connect-harness e2e-connected E2E_SCENARIO=profile-threads
@@ -73,6 +73,9 @@ jobs:
ref: ${{ inputs.connect_sha }}
token: ${{ secrets.PF_TESTING_GH_TOKEN }}
- name: Set up Docker Compose
uses: ./.github/actions/setup-docker-compose
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
@@ -159,7 +162,7 @@ jobs:
npm --prefix connect-harness/web ci
connect-harness/web/node_modules/.bin/playwright install --with-deps chromium
make -C connect-harness e2e-connected-dispatch-check
RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
TMPDIR="$GITHUB_WORKSPACE" RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
RUSTFS_WORKTREE="$GITHUB_WORKSPACE/rustfs-source" \
CONNECT_E2E_TOP_API_EVIDENCE="$GITHUB_WORKSPACE/top-api-service-job-evidence.json" \
make -C connect-harness e2e-connected E2E_SCENARIO=top-api
@@ -68,6 +68,9 @@ jobs:
ref: ${{ inputs.connect_sha }}
token: ${{ secrets.PF_TESTING_GH_TOKEN }}
- name: Set up Docker Compose
uses: ./.github/actions/setup-docker-compose
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
@@ -218,7 +221,7 @@ jobs:
npm --prefix connect-harness/web ci
connect-harness/web/node_modules/.bin/playwright install --with-deps chromium
make -C connect-harness e2e-connected-dispatch-check
RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
TMPDIR="$GITHUB_WORKSPACE" RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
RUSTFS_WORKTREE="$GITHUB_WORKSPACE/rustfs-source" \
CONNECT_E2E_TOP_DISK_EVIDENCE="$GITHUB_WORKSPACE/top-disk-service-job-evidence.json" \
make -C connect-harness e2e-connected E2E_SCENARIO=top-disk
@@ -73,6 +73,9 @@ jobs:
ref: ${{ inputs.connect_sha }}
token: ${{ secrets.PF_TESTING_GH_TOKEN }}
- name: Set up Docker Compose
uses: ./.github/actions/setup-docker-compose
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
@@ -159,7 +162,7 @@ jobs:
npm --prefix connect-harness/web ci
connect-harness/web/node_modules/.bin/playwright install --with-deps chromium
make -C connect-harness e2e-connected-dispatch-check
RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
TMPDIR="$GITHUB_WORKSPACE" RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
RUSTFS_WORKTREE="$GITHUB_WORKSPACE/rustfs-source" \
CONNECT_E2E_TOP_LOCKS_EVIDENCE="$GITHUB_WORKSPACE/top-locks-service-job-evidence.json" \
make -C connect-harness e2e-connected E2E_SCENARIO=top-locks
@@ -73,6 +73,9 @@ jobs:
ref: ${{ inputs.connect_sha }}
token: ${{ secrets.PF_TESTING_GH_TOKEN }}
- name: Set up Docker Compose
uses: ./.github/actions/setup-docker-compose
- name: Set up Python
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
with:
@@ -159,7 +162,7 @@ jobs:
npm --prefix connect-harness/web ci
connect-harness/web/node_modules/.bin/playwright install --with-deps chromium
make -C connect-harness e2e-connected-dispatch-check
RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
TMPDIR="$GITHUB_WORKSPACE" RUSTFS_BINARY="$GITHUB_WORKSPACE/binary/rustfs" \
RUSTFS_WORKTREE="$GITHUB_WORKSPACE/rustfs-source" \
CONNECT_E2E_TOP_RPC_EVIDENCE="$GITHUB_WORKSPACE/top-rpc-service-job-evidence.json" \
make -C connect-harness e2e-connected E2E_SCENARIO=top-rpc
+1 -1
View File
@@ -155,7 +155,7 @@ default build (lifecycle:
- ⚠️ PARTIALLY VIOLATED: serving-side `s3s` references remain in ecstore
(bucket metadata/replication/lifecycle DTOs and error mapping). The
count is ratcheted shrink-only by `scripts/check_s3s_footprint.sh`
(the `ecstore_files` line of `.config/s3s-footprint-baseline.txt`; the `object_lock` module was converted to
(`S3S_ECSTORE_FILES_BASELINE`; the `object_lock` module was converted to
storage-level types as the first ratchet step). Target state: the
baseline reaches zero and ecstore's `Cargo.toml` drops `s3s`.
Generated
+85 -95
View File
@@ -3708,9 +3708,9 @@ dependencies = [
[[package]]
name = "deranged"
version = "0.5.8"
version = "0.5.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c"
checksum = "9e9de72ce2ad1f90dc62fa25f0f430ef85eb4b0d8fa0be4f30373bc40a21d28e"
dependencies = [
"serde_core",
]
@@ -3800,9 +3800,9 @@ dependencies = [
[[package]]
name = "dial9"
version = "0.5.3"
version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b35f95885a2167602b457c77caf898670c02eb0aaba72bef76ac5b0faef86e14"
checksum = "734474ca4e814f893385c8fdea63688a9e6034ea19a5ec9846f8f515a69c9bed"
dependencies = [
"bon",
"dial9-core",
@@ -3817,9 +3817,9 @@ dependencies = [
[[package]]
name = "dial9-core"
version = "0.5.3"
version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "82f826078c782919cf9a4fb0b2b6755dab24d6c908b5181469b23cf6eb8f6fff"
checksum = "4ff5074fab1f810a73d58336275eb163e795ab0ae923fcc8437ba58bf2994e33"
dependencies = [
"ahash",
"arc-swap",
@@ -3851,9 +3851,9 @@ dependencies = [
[[package]]
name = "dial9-metrique"
version = "0.5.3"
version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9c59e35e7b38ce255d3dc6310b48c3ea54307c42a0589eda44ea91ba24eadf67"
checksum = "3dd61f8f3d44db27e7a5e970c216bf4c19db66ebb68b7f58db2549f010c3a02e"
dependencies = [
"dial9-core",
"dial9-trace-format",
@@ -3865,9 +3865,9 @@ dependencies = [
[[package]]
name = "dial9-perf-self-profile"
version = "0.5.3"
version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3c429b209bc07612b2e60267e8bca80cd2b249f498e9a0b83f7caf514d5f5388"
checksum = "d320bef9be4ef9c20ca9800cbfc369c319304b5e11d0a7aa931f5482aa5c1bd8"
dependencies = [
"blazesym",
"bon",
@@ -3884,9 +3884,9 @@ dependencies = [
[[package]]
name = "dial9-tokio-telemetry"
version = "0.5.3"
version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "aee46c0a97dd88ba151f9688c79de60c41cd7efc6855e6eb369af06326cba8f3"
checksum = "1a98e1e673f2dd9e223811832be7a4344ddfe48282a2e7efbf1289e47e4322ce"
dependencies = [
"bon",
"bytes",
@@ -4030,7 +4030,7 @@ checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"
[[package]]
name = "e2e_test"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"anyhow",
"async-compression",
@@ -4310,9 +4310,9 @@ dependencies = [
[[package]]
name = "equivalent"
version = "1.0.2"
version = "1.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
checksum = "00d174d5400e5e8fd687ad1049e2f578285fa914201b1af7e8b112a4546bd826"
[[package]]
name = "erased-serde"
@@ -6052,9 +6052,9 @@ checksum = "a4933f3f57a8e9d9da04db23fb153356ecaf00cbd14aee46279c33dc80925c37"
[[package]]
name = "lapin"
version = "4.12.1"
version = "4.12.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1f6674e18d68e810f66ea962f8080f1ded75f813c1cd3fae9ed53bab760dfcfd"
checksum = "3a585eb3060f00f09a1d231e3030afe59f2c2d63f794787ab1dd87d2f72df6a3"
dependencies = [
"amq-protocol",
"async-rs",
@@ -8529,7 +8529,9 @@ dependencies = [
[[package]]
name = "pyroscope"
version = "2.1.1"
version = "2.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "357782a4af9b6b43bd3a9caa64dd802967498b1fce1d4e0bda552267d62de7cf"
dependencies = [
"aligned-vec",
"backtrace",
@@ -8544,6 +8546,7 @@ dependencies = [
"once_cell",
"prost 0.14.4",
"reqwest",
"ruzstd",
"serde_json",
"smallvec",
"spin 0.12.3",
@@ -9519,7 +9522,7 @@ dependencies = [
[[package]]
name = "rustfs"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"aes-gcm",
"anyhow",
@@ -9678,7 +9681,7 @@ dependencies = [
[[package]]
name = "rustfs-audit"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"const-str",
"futures",
@@ -9700,7 +9703,7 @@ dependencies = [
[[package]]
name = "rustfs-checksums"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"base64-simd",
"bytes",
@@ -9716,7 +9719,7 @@ dependencies = [
[[package]]
name = "rustfs-common"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"arc-swap",
"hotpath",
@@ -9731,7 +9734,7 @@ dependencies = [
[[package]]
name = "rustfs-concurrency"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"hotpath",
"insta",
@@ -9744,7 +9747,7 @@ dependencies = [
[[package]]
name = "rustfs-config"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"const-str",
"hotpath",
@@ -9754,7 +9757,7 @@ dependencies = [
[[package]]
name = "rustfs-connect-symbol-catalog"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"hex-simd",
"object 0.40.0",
@@ -9766,7 +9769,7 @@ dependencies = [
[[package]]
name = "rustfs-credentials"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"base64-simd",
"hmac 0.13.0",
@@ -9780,7 +9783,7 @@ dependencies = [
[[package]]
name = "rustfs-crypto"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"aes-gcm",
"argon2",
@@ -9801,7 +9804,7 @@ dependencies = [
[[package]]
name = "rustfs-data-usage"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"hotpath",
"rmp-serde",
@@ -9811,7 +9814,7 @@ dependencies = [
[[package]]
name = "rustfs-ecstore"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"arc-swap",
"async-channel",
@@ -9944,7 +9947,7 @@ dependencies = [
[[package]]
name = "rustfs-extension-schema"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"hotpath",
"serde",
@@ -9954,7 +9957,7 @@ dependencies = [
[[package]]
name = "rustfs-filemeta"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"arc-swap",
"base64-simd",
@@ -9983,7 +9986,7 @@ dependencies = [
[[package]]
name = "rustfs-heal"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"async-trait",
"base64-simd",
@@ -10025,7 +10028,7 @@ dependencies = [
[[package]]
name = "rustfs-heal-contracts"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"serde",
"serde_json",
@@ -10035,7 +10038,7 @@ dependencies = [
[[package]]
name = "rustfs-iam"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"arc-swap",
"async-trait",
@@ -10085,7 +10088,7 @@ dependencies = [
[[package]]
name = "rustfs-io-core"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"bytes",
"hotpath",
@@ -10097,7 +10100,7 @@ dependencies = [
[[package]]
name = "rustfs-io-metrics"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"criterion",
"hotpath",
@@ -10161,7 +10164,7 @@ dependencies = [
[[package]]
name = "rustfs-keystone"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"bytes",
"futures",
@@ -10188,7 +10191,7 @@ dependencies = [
[[package]]
name = "rustfs-kms"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"aes-gcm",
"anyhow",
@@ -10238,14 +10241,14 @@ dependencies = [
[[package]]
name = "rustfs-license"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"thiserror 2.0.21",
]
[[package]]
name = "rustfs-lifecycle"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"async-trait",
"hotpath",
@@ -10267,7 +10270,7 @@ dependencies = [
[[package]]
name = "rustfs-lock"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"async-trait",
"compact_str",
@@ -10290,7 +10293,7 @@ dependencies = [
[[package]]
name = "rustfs-log-analyzer"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"chrono",
"flate2",
@@ -10309,7 +10312,7 @@ dependencies = [
[[package]]
name = "rustfs-madmin"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"hotpath",
"http 1.5.0",
@@ -10319,6 +10322,7 @@ dependencies = [
"reqwest",
"rmp-serde",
"rustfs-signer",
"s3s",
"serde",
"serde_json",
"sysinfo",
@@ -10346,7 +10350,7 @@ dependencies = [
[[package]]
name = "rustfs-notify"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"arc-swap",
"async-trait",
@@ -10381,7 +10385,7 @@ dependencies = [
[[package]]
name = "rustfs-object-capacity"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"criterion",
"futures",
@@ -10400,7 +10404,7 @@ dependencies = [
[[package]]
name = "rustfs-object-data-cache"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"bytes",
"criterion",
@@ -10417,7 +10421,7 @@ dependencies = [
[[package]]
name = "rustfs-obs"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"chrono",
"crossbeam-channel",
@@ -10480,7 +10484,7 @@ dependencies = [
[[package]]
name = "rustfs-policy"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"async-trait",
"base64-simd",
@@ -10511,7 +10515,7 @@ dependencies = [
[[package]]
name = "rustfs-protocols"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"async-compression",
"async-trait",
@@ -10543,7 +10547,6 @@ dependencies = [
"rustfs-keystone",
"rustfs-policy",
"rustfs-rio",
"rustfs-s3-types",
"rustfs-storage-api",
"rustfs-test-utils",
"rustfs-tls-runtime",
@@ -10558,7 +10561,6 @@ dependencies = [
"sha2 0.11.1",
"socket2",
"subtle",
"sync_wrapper",
"tempfile",
"thiserror 2.0.21",
"time",
@@ -10575,7 +10577,7 @@ dependencies = [
[[package]]
name = "rustfs-protos"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"flatbuffers",
"hotpath",
@@ -10600,7 +10602,7 @@ dependencies = [
[[package]]
name = "rustfs-replication"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"byteorder",
"bytes",
@@ -10618,7 +10620,7 @@ dependencies = [
[[package]]
name = "rustfs-rio"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"aes-gcm",
"arc-swap",
@@ -10662,7 +10664,7 @@ dependencies = [
[[package]]
name = "rustfs-rio-v2"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"aes-gcm",
"bytes",
@@ -10685,7 +10687,7 @@ dependencies = [
[[package]]
name = "rustfs-s3-client"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"base64-simd",
"bytes",
@@ -10706,7 +10708,6 @@ dependencies = [
"rustfs-checksums",
"rustfs-config",
"rustfs-rio",
"rustfs-s3-types",
"rustfs-signer",
"rustfs-storage-api",
"rustfs-tls-runtime",
@@ -10730,7 +10731,7 @@ dependencies = [
[[package]]
name = "rustfs-s3-ops"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"hotpath",
"rustfs-s3-types",
@@ -10738,26 +10739,16 @@ dependencies = [
[[package]]
name = "rustfs-s3-types"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"bytes",
"futures-core",
"hotpath",
"http 1.5.0",
"http-body",
"http-body-util",
"hyper",
"hyper-util",
"pin-project-lite",
"s3s",
"serde",
"serde_json",
"tokio",
]
[[package]]
name = "rustfs-s3select-api"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"arc-swap",
"async-compression",
@@ -10776,7 +10767,6 @@ dependencies = [
"parking_lot",
"rustfs-common",
"rustfs-ecstore",
"rustfs-s3-types",
"rustfs-storage-api",
"rustfs-test-utils",
"s3s",
@@ -10794,7 +10784,7 @@ dependencies = [
[[package]]
name = "rustfs-s3select-query"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"async-trait",
"bytes",
@@ -10814,7 +10804,7 @@ dependencies = [
[[package]]
name = "rustfs-scanner"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"async-trait",
"bytes",
@@ -10861,7 +10851,7 @@ dependencies = [
[[package]]
name = "rustfs-scanner-metrics"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"chrono",
"jiff",
@@ -10876,7 +10866,7 @@ dependencies = [
[[package]]
name = "rustfs-security-governance"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"hotpath",
"thiserror 2.0.21",
@@ -10884,16 +10874,15 @@ dependencies = [
[[package]]
name = "rustfs-signer"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"base64-simd",
"bytes",
"hotpath",
"http 1.5.0",
"http-body",
"hyper",
"rustfs-s3-types",
"rustfs-utils",
"s3s",
"serde_urlencoded",
"thiserror 2.0.21",
"time",
@@ -10903,7 +10892,7 @@ dependencies = [
[[package]]
name = "rustfs-storage-api"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"async-trait",
"hotpath",
@@ -10918,7 +10907,7 @@ dependencies = [
[[package]]
name = "rustfs-targets"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"arc-swap",
"async-nats",
@@ -10951,6 +10940,7 @@ dependencies = [
"rustfs-utils",
"rustls",
"rustls-native-certs",
"s3s",
"serde",
"serde_json",
"sha2 0.11.1",
@@ -10971,7 +10961,7 @@ dependencies = [
[[package]]
name = "rustfs-test-utils"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"hotpath",
"rustfs-data-usage",
@@ -10987,7 +10977,7 @@ dependencies = [
[[package]]
name = "rustfs-tls-runtime"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"arc-swap",
"hotpath",
@@ -11026,7 +11016,7 @@ dependencies = [
[[package]]
name = "rustfs-trusted-proxies"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"async-trait",
"axum",
@@ -11063,7 +11053,7 @@ dependencies = [
[[package]]
name = "rustfs-utils"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"base64-simd",
"blake2",
@@ -11105,7 +11095,7 @@ dependencies = [
[[package]]
name = "rustfs-zip"
version = "1.0.1"
version = "1.0.2"
dependencies = [
"async-compression",
"futures",
@@ -11220,9 +11210,9 @@ dependencies = [
[[package]]
name = "rustls-native-certs"
version = "0.8.4"
version = "0.8.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d"
checksum = "4a073f5dc7469f984c52ad2752b63b0807745133b6de880b7b64c1ac4c48aec4"
dependencies = [
"openssl-probe",
"rustls-pki-types",
@@ -12730,9 +12720,9 @@ dependencies = [
[[package]]
name = "tokio"
version = "1.53.2"
version = "1.53.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e95f91fcc7a621e8b030f6aa23c71fe9838ae2fb4d8118b75602a328f5144044"
checksum = "ce3335fa71841cda333a58d7615b03901380ecf09d59b3296d21f8bbac0dde4e"
dependencies = [
"bytes",
"libc",
@@ -14094,18 +14084,18 @@ dependencies = [
[[package]]
name = "zerocopy"
version = "0.8.62"
version = "0.8.63"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "86502bf56ac7c77571a32e2647bb2a15894565e981fb2a48d7bde2d91c965a9d"
checksum = "e5fe1f8f1b06191a00962174c61aa5005e0bb391a6d80d07e24d115c01a92ed8"
dependencies = [
"zerocopy-derive",
]
[[package]]
name = "zerocopy-derive"
version = "0.8.62"
version = "0.8.63"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5457206954b06561e2608c7e19cf58b1926586d999c246eebe4502f7e2039d1a"
checksum = "863ad3ac83293fb4d740aedbfdc9240dd8d1a50c1099acd76ce80ce7c7230c7f"
dependencies = [
"proc-macro2",
"quote",
@@ -14188,9 +14178,9 @@ dependencies = [
[[package]]
name = "zip"
version = "9.0.2"
version = "9.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9e2fc56a6707efb066c40aead2409ebb1224e6dde24a074e9fbc78ced500507b"
checksum = "c2d98561b612a84ae525331a7a094429ef8a5772514ed507ff6db5f54badc971"
dependencies = [
"aes 0.9.3",
"bzip2",
+57 -68
View File
@@ -48,7 +48,7 @@ members = [
"crates/replication", # Replication contracts and wire formats
"crates/concurrency", # Concurrency management for RustFS - timeout, locking, backpressure, and I/O scheduling
"crates/s3-client", # S3 client for engine-side consumption of remote S3 endpoints (tiering, transition targets)
"crates/s3-types", # S3 event and error type definitions
"crates/s3-types", # S3 event type definitions
"crates/s3-ops", # S3 operation definitions and mapping
"crates/s3select-api", # S3 Select API interface
"crates/s3select-query", # S3 Select query engine
@@ -67,7 +67,6 @@ members = [
"crates/io-core", # Buffered Bytes, mmap-then-copy, and aligned pread I/O helpers
"crates/zip", # ZIP file handling and compression
]
exclude = ["third-party/pyroscope"]
resolver = "3"
[workspace.package]
@@ -75,7 +74,7 @@ edition = "2024"
license = "Apache-2.0"
repository = "https://github.com/rustfs/rustfs"
rust-version = "1.99.0"
version = "1.0.1"
version = "1.0.2"
homepage = "https://rustfs.com"
description = "RustFS is a high-performance distributed object storage software built using Rust, one of the most popular languages worldwide. "
keywords = ["RustFS", "Minio", "object-storage", "filesystem", "s3"]
@@ -92,56 +91,56 @@ redundant_clone = "warn"
[workspace.dependencies]
# RustFS Internal Crates
rustfs = { path = "./rustfs", version = "1.0.1" }
rustfs-heal = { path = "crates/heal", version = "1.0.1" }
rustfs-heal-contracts = { path = "crates/heal-contracts", version = "1.0.1" }
rustfs-scanner-metrics = { path = "crates/scanner-metrics", version = "1.0.1" }
rustfs-audit = { path = "crates/audit", version = "1.0.1" }
rustfs-checksums = { path = "crates/checksums", version = "1.0.1" }
rustfs-common = { path = "crates/common", version = "1.0.1" }
rustfs-data-usage = { path = "crates/data-usage", version = "1.0.1" }
rustfs-config = { path = "./crates/config", version = "1.0.1" }
rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.1" }
rustfs-credentials = { path = "crates/credentials", version = "1.0.1" }
rustfs-crypto = { path = "crates/crypto", version = "1.0.1" }
rustfs-ecstore = { path = "crates/ecstore", version = "1.0.1" }
rustfs-filemeta = { path = "crates/filemeta", version = "1.0.1" }
rustfs-iam = { path = "crates/iam", version = "1.0.1" }
rustfs-keystone = { path = "crates/keystone", version = "1.0.1" }
rustfs-license = { path = "crates/license", version = "1.0.1" }
rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.1" }
rustfs-kms = { path = "crates/kms", version = "1.0.1" }
rustfs-lock = { path = "crates/lock", version = "1.0.1" }
rustfs-madmin = { path = "crates/madmin", version = "1.0.1" }
rustfs-notify = { path = "crates/notify", version = "1.0.1" }
rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.1" }
rustfs-io-core = { path = "crates/io-core", version = "1.0.1" }
rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.1" }
rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.1", default-features = false }
rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.1" }
rustfs-obs = { path = "crates/obs", version = "1.0.1" }
rustfs-policy = { path = "crates/policy", version = "1.0.1" }
rustfs-protos = { path = "crates/protos", version = "1.0.1" }
rustfs-protocols = { path = "crates/protocols", version = "1.0.1" }
rustfs-replication = { path = "crates/replication", version = "1.0.1" }
rustfs-rio = { path = "crates/rio", version = "1.0.1" }
rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.1" }
rustfs-s3-client = { path = "crates/s3-client", version = "1.0.1" }
rustfs-s3-types = { path = "crates/s3-types", version = "1.0.1" }
rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.1" }
rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.1" }
rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.1" }
rustfs-scanner = { path = "crates/scanner", version = "1.0.1" }
rustfs-security-governance = { path = "crates/security-governance", version = "1.0.1" }
rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.1" }
rustfs-signer = { path = "crates/signer", version = "1.0.1" }
rustfs-storage-api = { path = "crates/storage-api", version = "1.0.1" }
rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.1" }
rustfs-targets = { path = "crates/targets", version = "1.0.1" }
rustfs-test-utils = { path = "crates/test-utils", version = "1.0.1" }
rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.1" }
rustfs-utils = { path = "crates/utils", version = "1.0.1" }
rustfs-zip = { path = "./crates/zip", version = "1.0.1" }
rustfs = { path = "./rustfs", version = "1.0.2" }
rustfs-heal = { path = "crates/heal", version = "1.0.2" }
rustfs-heal-contracts = { path = "crates/heal-contracts", version = "1.0.2" }
rustfs-scanner-metrics = { path = "crates/scanner-metrics", version = "1.0.2" }
rustfs-audit = { path = "crates/audit", version = "1.0.2" }
rustfs-checksums = { path = "crates/checksums", version = "1.0.2" }
rustfs-common = { path = "crates/common", version = "1.0.2" }
rustfs-data-usage = { path = "crates/data-usage", version = "1.0.2" }
rustfs-config = { path = "./crates/config", version = "1.0.2" }
rustfs-concurrency = { path = "./crates/concurrency", version = "1.0.2" }
rustfs-credentials = { path = "crates/credentials", version = "1.0.2" }
rustfs-crypto = { path = "crates/crypto", version = "1.0.2" }
rustfs-ecstore = { path = "crates/ecstore", version = "1.0.2" }
rustfs-filemeta = { path = "crates/filemeta", version = "1.0.2" }
rustfs-iam = { path = "crates/iam", version = "1.0.2" }
rustfs-keystone = { path = "crates/keystone", version = "1.0.2" }
rustfs-license = { path = "crates/license", version = "1.0.2" }
rustfs-lifecycle = { path = "crates/lifecycle", version = "1.0.2" }
rustfs-kms = { path = "crates/kms", version = "1.0.2" }
rustfs-lock = { path = "crates/lock", version = "1.0.2" }
rustfs-madmin = { path = "crates/madmin", version = "1.0.2" }
rustfs-notify = { path = "crates/notify", version = "1.0.2" }
rustfs-io-metrics = { path = "crates/io-metrics", version = "1.0.2" }
rustfs-io-core = { path = "crates/io-core", version = "1.0.2" }
rustfs-object-capacity = { path = "crates/object-capacity", version = "1.0.2" }
rustfs-object-data-cache = { path = "crates/object-data-cache", version = "1.0.2", default-features = false }
rustfs-log-analyzer = { path = "crates/log-analyzer", version = "1.0.2" }
rustfs-obs = { path = "crates/obs", version = "1.0.2" }
rustfs-policy = { path = "crates/policy", version = "1.0.2" }
rustfs-protos = { path = "crates/protos", version = "1.0.2" }
rustfs-protocols = { path = "crates/protocols", version = "1.0.2" }
rustfs-replication = { path = "crates/replication", version = "1.0.2" }
rustfs-rio = { path = "crates/rio", version = "1.0.2" }
rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.2" }
rustfs-s3-client = { path = "crates/s3-client", version = "1.0.2" }
rustfs-s3-types = { path = "crates/s3-types", version = "1.0.2" }
rustfs-s3-ops = { path = "crates/s3-ops", version = "1.0.2" }
rustfs-s3select-api = { path = "crates/s3select-api", version = "1.0.2" }
rustfs-s3select-query = { path = "crates/s3select-query", version = "1.0.2" }
rustfs-scanner = { path = "crates/scanner", version = "1.0.2" }
rustfs-security-governance = { path = "crates/security-governance", version = "1.0.2" }
rustfs-extension-schema = { path = "crates/extension-schema", version = "1.0.2" }
rustfs-signer = { path = "crates/signer", version = "1.0.2" }
rustfs-storage-api = { path = "crates/storage-api", version = "1.0.2" }
rustfs-trusted-proxies = { path = "crates/trusted-proxies", version = "1.0.2" }
rustfs-targets = { path = "crates/targets", version = "1.0.2" }
rustfs-test-utils = { path = "crates/test-utils", version = "1.0.2" }
rustfs-tls-runtime = { path = "crates/tls-runtime", version = "1.0.2" }
rustfs-utils = { path = "crates/utils", version = "1.0.2" }
rustfs-zip = { path = "./crates/zip", version = "1.0.2" }
# Async Runtime and Networking
async-channel = "2.5.0"
@@ -157,7 +156,7 @@ futures-lite = "2.6.1"
futures-util = "0.3.34"
pollster = "1.0.1"
pulsar = { default-features = false, version = "6.9.0" }
lapin = { default-features = false, version = "4.12.1" }
lapin = { default-features = false, version = "4.12.2" }
hyper = { version = "1.12.0" }
h2 = { version = "0.4.20" }
hyper-rustls = { default-features = false, version = "0.27.10" }
@@ -170,7 +169,7 @@ minlz = "1.2.3"
reqwest = "0.13.5"
rustfs-kafka-async = { version = "1.4.0" }
socket2 = { version = "0.6.5" }
tokio = { version = "1.53.2" }
tokio = { version = "1.53.3" }
tokio-rustls = { default-features = false, version = "0.26.6" }
tokio-stream = { version = "0.1.19" }
tokio-test = "0.4.6"
@@ -325,10 +324,6 @@ rust-embed = { version = "8.13.0" }
rustc-hash = { version = "2.1.3" }
s3s = { git = "https://github.com/s3s-project/s3s", rev = "5761ddfe4c509fcac87e3b99fd5bd9e176b442c0", features = ["minio"] }
s3s-sigv4 = { git = "https://github.com/s3s-project/s3s", rev = "5761ddfe4c509fcac87e3b99fd5bd9e176b442c0" }
# The same s3s pin under the name the legacy edge of the gateway migration uses
# (crates/s3-types, feature `compat-s3s`). Keep `rev` identical to `s3s` above.
# DELETE BY T4.2 (rustfs/backlog#2784).
legacy-s3s = { package = "s3s", git = "https://github.com/s3s-project/s3s", rev = "5761ddfe4c509fcac87e3b99fd5bd9e176b442c0", features = ["minio"] }
serial_test = "4.0.1"
shadow-rs = { default-features = false, version = "2.0.0" }
siphasher = "1.0.4"
@@ -337,7 +332,6 @@ compact_str = "0.10.0"
snap = "1.1.2"
starshard = { version = "2.4.0" }
strum = { version = "0.28.0" }
sync_wrapper = "1.0.2"
sysinfo = "0.39.6"
temp-env = "0.3.6"
tempfile = "3.27.0"
@@ -360,13 +354,13 @@ winapi-util = "0.1.11"
windows = { version = "0.62.2" }
windows-sys = "0.61.2"
xxhash-rust = { version = "0.8.19" }
zip = "9.0.2"
zip = "9.0.3"
zstd = "0.14.2"
# Observability and Metrics
metrics = "0.24.6"
metrics-util = "0.20"
dial9 = { version = "0.5.3", default-features = false }
dial9 = { version = "0.5.4", default-features = false }
opentelemetry = { version = "0.33.1" }
opentelemetry-appender-tracing = { version = "0.33.1" }
opentelemetry-otlp = { version = "0.33.1" }
@@ -374,7 +368,7 @@ opentelemetry-proto = { version = "0.33.1", default-features = false, features =
opentelemetry_sdk = { version = "0.33.1" }
opentelemetry-semantic-conventions = { version = "0.33.1" }
opentelemetry-stdout = { version = "0.33.1" }
pyroscope = { version = "2.1.1" }
pyroscope = { version = "2.1.2" }
# FTP and SFTP
libunftp = { version = "0.23.1" }
@@ -427,8 +421,3 @@ inherits = "release"
inherits = "release"
debug = true
strip = "none"
# Remove this patch when Pyroscope releases support for object 0.40 or later.
# object 0.39 requires ruzstd 0.8, affected by RUSTSEC-2026-0344.
[patch.crates-io]
pyroscope = { path = "third-party/pyroscope" }
+6 -4
View File
@@ -860,10 +860,12 @@ pub(crate) async fn start_rebalance(cluster: &RustFSTestClusterEnvironment) -> T
if status.is_success() {
break response;
}
if status == StatusCode::INTERNAL_SERVER_ERROR
&& response.contains("requires a live fleet capability proof")
&& Instant::now() < deadline
{
// The init stage answers 503 + Retry-After while the cluster capability
// proof is still missing; older binaries surfaced the same condition as
// a 500 carrying the proof marker.
let waiting_for_readiness = status == StatusCode::SERVICE_UNAVAILABLE
|| (status == StatusCode::INTERNAL_SERVER_ERROR && response.contains("requires a live fleet capability proof"));
if waiting_for_readiness && Instant::now() < deadline {
sleep(Duration::from_secs(1)).await;
continue;
}
@@ -1,512 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! What one stack answered, and how two answers are compared.
//!
//! An [`Observation`] keeps only what a client acts on: the HTTP status, the S3 error code, the
//! allowlisted response headers and a few typed fields decoded from the response. The error
//! message is kept for diagnostics and never compared: message wording is judged by the wire
//! corpus, not here. The comparator is pure so it can be tested without a server.
use serde::Serialize;
use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};
/// Response headers compared exactly (after [`normalized_header_value`]).
const HEADER_ALLOWLIST: [&str; 7] = [
"content-type",
"x-amz-version-id",
"etag",
"x-amz-delete-marker",
"content-range",
"accept-ranges",
"x-amz-request-charged",
];
/// Every response header with this prefix is compared as well.
const HEADER_ALLOWLIST_PREFIX: &str = "x-amz-server-side-encryption";
/// Placeholder for a field one side did not send.
const ABSENT: &str = "<absent>";
/// The lowercase header name if the comparator looks at this header.
pub(crate) fn compared_header(name: &str) -> Option<String> {
let name = name.to_ascii_lowercase();
(HEADER_ALLOWLIST.contains(&name.as_str()) || name.starts_with(HEADER_ALLOWLIST_PREFIX)).then_some(name)
}
/// A version ID is minted per process, so its value never matches across two servers; only
/// whether it is the literal `null` (an unversioned write) carries meaning.
fn normalized_header_value(name: &str, value: &str) -> String {
if name == "x-amz-version-id" && value != "null" {
"<version-id>".to_owned()
} else {
value.to_owned()
}
}
/// One request's outcome as a client sees it.
#[derive(Clone, Debug, Serialize)]
pub(crate) struct Observation {
/// The S3 operation, named as in the `s3s::S3` trait (`GetObject`).
pub op: &'static str,
/// Which request of the script this was (`normal`, `negative`, or a setup label).
pub case: &'static str,
/// HTTP status; 0 when no response arrived.
pub status: u16,
/// The S3 error code, or the SDK's error kind when there was no service error.
pub error_code: Option<String>,
/// Allowlisted headers only, lowercase names, normalized values.
pub headers: BTreeMap<String, String>,
/// Typed values decoded from the response body, never free text.
pub fields: BTreeMap<String, String>,
/// Diagnostic only; never compared.
#[serde(skip)]
pub message: Option<String>,
}
impl Observation {
/// Builds an observation, keeping only the allowlisted headers.
pub(crate) fn new<'h>(
op: &'static str,
case: &'static str,
status: u16,
error_code: Option<String>,
headers: impl IntoIterator<Item = (&'h str, &'h str)>,
) -> Self {
let headers = headers
.into_iter()
.filter_map(|(name, value)| compared_header(name).map(|name| (name, value)))
.map(|(name, value)| {
let value = normalized_header_value(&name, value);
(name, value)
})
.collect();
Self {
op,
case,
status,
error_code,
headers,
fields: BTreeMap::new(),
message: None,
}
}
pub(crate) fn with_field(mut self, name: &str, value: impl Into<String>) -> Self {
self.fields.insert(name.to_owned(), value.into());
self
}
pub(crate) fn is_success(&self) -> bool {
(200..300).contains(&self.status)
}
}
/// One field on which the two stacks answered differently.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub(crate) struct Difference {
pub op: String,
pub case: String,
pub field: String,
pub a: String,
pub b: String,
}
/// Every difference between two transcripts of the same script, matched by `(op, case)`.
pub(crate) fn compare(a: &[Observation], b: &[Observation]) -> Vec<Difference> {
let mut differences = Vec::new();
let (index_a, duplicates_a) = index(a);
let (index_b, duplicates_b) = index(b);
for (op, case) in duplicates_a.union(&duplicates_b) {
let count = |side: &[Observation]| {
side.iter()
.filter(|observation| (observation.op, observation.case) == (*op, *case))
.count()
.to_string()
};
differences.push(difference(op, case, "duplicate_case", count(a), count(b)));
}
let keys: BTreeSet<_> = index_a.keys().chain(index_b.keys()).copied().collect();
for key @ (op, case) in keys {
match (index_a.get(&key), index_b.get(&key)) {
(Some(a), Some(b)) => compare_one(a, b, &mut differences),
(a, b) => {
let presence = |side: Option<&&Observation>| if side.is_some() { "present" } else { ABSENT }.to_owned();
differences.push(difference(op, case, "observation", presence(a), presence(b)));
}
}
}
differences
}
type Key = (&'static str, &'static str);
fn index(side: &[Observation]) -> (BTreeMap<Key, &Observation>, BTreeSet<Key>) {
let mut index = BTreeMap::new();
let mut duplicates = BTreeSet::new();
for observation in side {
let key = (observation.op, observation.case);
if index.insert(key, observation).is_some() {
duplicates.insert(key);
}
}
(index, duplicates)
}
fn difference(op: &str, case: &str, field: &str, a: String, b: String) -> Difference {
Difference {
op: op.to_owned(),
case: case.to_owned(),
field: field.to_owned(),
a,
b,
}
}
fn compare_one(a: &Observation, b: &Observation, differences: &mut Vec<Difference>) {
let mut field = |name: String, value_a: Option<&str>, value_b: Option<&str>| {
if value_a != value_b {
let show = |value: Option<&str>| value.unwrap_or(ABSENT).to_owned();
differences.push(difference(a.op, a.case, &name, show(value_a), show(value_b)));
}
};
field(
"status".to_owned(),
Some(a.status.to_string().as_str()),
Some(b.status.to_string().as_str()),
);
field("error_code".to_owned(), a.error_code.as_deref(), b.error_code.as_deref());
let headers: BTreeSet<_> = a.headers.keys().chain(b.headers.keys()).collect();
for name in headers {
field(
format!("header.{name}"),
a.headers.get(name).map(String::as_str),
b.headers.get(name).map(String::as_str),
);
}
let fields: BTreeSet<_> = a.fields.keys().chain(b.fields.keys()).collect();
for name in fields {
field(
format!("body.{name}"),
a.fields.get(name).map(String::as_str),
b.fields.get(name).map(String::as_str),
);
}
}
/// Operations observed on both sides.
pub(crate) fn compared_ops(a: &[Observation], b: &[Observation]) -> BTreeSet<&'static str> {
let ops_b: BTreeSet<_> = b.iter().map(|observation| observation.op).collect();
a.iter()
.map(|observation| observation.op)
.filter(|op| ops_b.contains(op))
.collect()
}
/// `(op, case)` pairs observed on both sides.
pub(crate) fn compared_cases(a: &[Observation], b: &[Observation]) -> usize {
let cases_b: BTreeSet<_> = b.iter().map(|observation| (observation.op, observation.case)).collect();
a.iter()
.map(|observation| (observation.op, observation.case))
.collect::<BTreeSet<_>>()
.intersection(&cases_b)
.count()
}
/// The JSON report written for every run.
#[derive(Debug, Serialize)]
pub(crate) struct Report {
pub schema: u32,
pub run: String,
pub stack_a: String,
pub stack_b: String,
pub region_a: Option<String>,
pub region_b: Option<String>,
pub compared_ops: usize,
pub compared_cases: usize,
pub differences: Vec<Difference>,
}
impl Report {
/// Writes the report as `<directory>/<run>.json` and returns its path.
pub(crate) fn write_to(&self, directory: &Path) -> std::io::Result<PathBuf> {
std::fs::create_dir_all(directory)?;
let path = directory.join(format!("{}.json", self.run));
let json = serde_json::to_vec_pretty(self).map_err(std::io::Error::other)?;
std::fs::write(&path, json)?;
Ok(path)
}
}
/// One line per difference, for a failing assertion.
pub(crate) fn render(differences: &[Difference]) -> String {
differences
.iter()
.map(|d| format!("({}, {}, {}, a={:?}, b={:?})", d.op, d.case, d.field, d.a, d.b))
.collect::<Vec<_>>()
.join("\n")
}
#[cfg(test)]
mod tests {
use super::*;
fn ok(op: &'static str) -> Observation {
Observation::new(op, "normal", 200, None, [("content-type", "application/xml")])
}
fn fields(differences: &[Difference]) -> Vec<&str> {
differences.iter().map(|d| d.field.as_str()).collect()
}
#[test]
fn identical_transcripts_have_no_difference() {
let a = vec![
ok("HeadBucket"),
ok("GetBucketLocation").with_field("location_constraint", "us-east-1"),
];
let b = a.clone();
assert_eq!(compare(&a, &b), Vec::new());
assert_eq!(compared_ops(&a, &b).len(), 2);
}
#[test]
fn a_status_difference_is_reported() {
let a = vec![ok("PutBucketAcl")];
let b = vec![Observation::new(
"PutBucketAcl",
"normal",
403,
None,
[("content-type", "application/xml")],
)];
assert_eq!(
compare(&a, &b),
vec![Difference {
op: "PutBucketAcl".into(),
case: "normal".into(),
field: "status".into(),
a: "200".into(),
b: "403".into(),
}]
);
}
#[test]
fn an_error_code_difference_is_reported() {
let a = vec![Observation::new("GetObject", "negative", 404, Some("NoSuchKey".into()), [])];
let b = vec![Observation::new(
"GetObject",
"negative",
404,
Some("NoSuchBucket".into()),
[],
)];
let differences = compare(&a, &b);
assert_eq!(fields(&differences), ["error_code"]);
assert_eq!((differences[0].a.as_str(), differences[0].b.as_str()), ("NoSuchKey", "NoSuchBucket"));
}
#[test]
fn an_error_code_on_one_side_only_is_reported() {
let a = vec![Observation::new("HeadBucket", "negative", 404, Some("NotFound".into()), [])];
let b = vec![Observation::new("HeadBucket", "negative", 404, None, [])];
let differences = compare(&a, &b);
assert_eq!(fields(&differences), ["error_code"]);
assert_eq!(differences[0].b, ABSENT);
}
#[test]
fn an_allowlisted_header_difference_is_reported() {
let a = vec![Observation::new("HeadObject", "normal", 200, None, [("ETag", "\"aa\"")])];
let b = vec![Observation::new("HeadObject", "normal", 200, None, [("etag", "\"bb\"")])];
assert_eq!(fields(&compare(&a, &b)), ["header.etag"]);
}
#[test]
fn a_header_sent_by_one_side_only_is_reported() {
let a = vec![Observation::new(
"DeleteObject",
"normal",
204,
None,
[("x-amz-delete-marker", "true")],
)];
let b = vec![Observation::new("DeleteObject", "normal", 204, None, [])];
let differences = compare(&a, &b);
assert_eq!(fields(&differences), ["header.x-amz-delete-marker"]);
assert_eq!((differences[0].a.as_str(), differences[0].b.as_str()), ("true", ABSENT));
}
#[test]
fn every_server_side_encryption_header_is_compared() {
let header = "x-amz-server-side-encryption-customer-key-md5";
let a = vec![Observation::new("PutObject", "sse-c", 200, None, [(header, "one")])];
let b = vec![Observation::new("PutObject", "sse-c", 200, None, [(header, "two")])];
assert_eq!(fields(&compare(&a, &b)), [format!("header.{header}")]);
}
#[test]
fn headers_outside_the_allowlist_are_not_compared() {
let a = vec![Observation::new(
"ListBuckets",
"normal",
200,
None,
[
("date", "Mon"),
("x-amz-request-id", "1"),
("server", "a"),
("content-length", "10"),
],
)];
let b = vec![Observation::new(
"ListBuckets",
"normal",
200,
None,
[
("date", "Tue"),
("x-amz-request-id", "2"),
("server", "b"),
("content-length", "11"),
],
)];
assert_eq!(compare(&a, &b), Vec::new());
assert!(a[0].headers.is_empty(), "non-allowlisted headers were kept: {:?}", a[0].headers);
}
#[test]
fn distinct_version_ids_are_not_a_difference() {
let a = vec![Observation::new(
"PutObject",
"normal",
200,
None,
[("x-amz-version-id", "0b1c")],
)];
let b = vec![Observation::new(
"PutObject",
"normal",
200,
None,
[("x-amz-version-id", "9f8e")],
)];
assert_eq!(compare(&a, &b), Vec::new());
}
#[test]
fn a_null_version_id_against_a_real_one_is_reported() {
let a = vec![Observation::new(
"PutObject",
"normal",
200,
None,
[("x-amz-version-id", "null")],
)];
let b = vec![Observation::new(
"PutObject",
"normal",
200,
None,
[("x-amz-version-id", "9f8e")],
)];
assert_eq!(fields(&compare(&a, &b)), ["header.x-amz-version-id"]);
}
#[test]
fn a_body_field_difference_is_reported() {
let a = vec![ok("GetBucketLocation").with_field("location_constraint", "us-east-1")];
let b = vec![ok("GetBucketLocation").with_field("location_constraint", "eu-west-1")];
let differences = compare(&a, &b);
assert_eq!(fields(&differences), ["body.location_constraint"]);
assert_eq!((differences[0].a.as_str(), differences[0].b.as_str()), ("us-east-1", "eu-west-1"));
}
#[test]
fn a_body_field_on_one_side_only_is_reported() {
let a = vec![ok("ListObjectsV2").with_field("keys", "[]")];
let b = vec![ok("ListObjectsV2")];
assert_eq!(fields(&compare(&a, &b)), ["body.keys"]);
}
#[test]
fn error_messages_are_never_compared() {
let mut a = Observation::new("GetObject", "negative", 404, Some("NoSuchKey".into()), []);
let mut b = a.clone();
a.message = Some("The specified key does not exist.".into());
b.message = Some("Object not found".into());
assert_eq!(compare(&[a], &[b]), Vec::new());
}
#[test]
fn a_case_observed_on_one_side_only_is_reported() {
let a = vec![ok("HeadBucket"), ok("ListBuckets")];
let b = vec![ok("HeadBucket")];
let differences = compare(&a, &b);
assert_eq!(fields(&differences), ["observation"]);
assert_eq!(differences[0].op, "ListBuckets");
assert_eq!(compared_ops(&a, &b).into_iter().collect::<Vec<_>>(), ["HeadBucket"]);
}
#[test]
fn a_case_recorded_twice_is_reported() {
let a = vec![ok("HeadBucket"), ok("HeadBucket")];
let b = vec![ok("HeadBucket")];
assert_eq!(fields(&compare(&a, &b)), ["duplicate_case"]);
}
#[test]
fn compared_ops_counts_operations_not_cases() {
let negative = Observation::new("HeadBucket", "negative", 404, Some("NotFound".into()), []);
let a = vec![ok("HeadBucket"), negative, ok("ListBuckets")];
let b = a.clone();
assert_eq!(compared_ops(&a, &b).len(), 2);
}
#[test]
fn the_report_lists_each_difference_as_op_field_a_b() {
let directory = std::env::temp_dir().join(format!("dual-stack-report-{}", uuid::Uuid::new_v4()));
let report = Report {
schema: 1,
run: "unit".into(),
stack_a: "legacy".into(),
stack_b: "legacy".into(),
region_a: None,
region_b: Some("eu-west-1".into()),
compared_ops: 1,
compared_cases: 1,
differences: compare(
&[ok("GetBucketLocation").with_field("location_constraint", "us-east-1")],
&[ok("GetBucketLocation").with_field("location_constraint", "eu-west-1")],
),
};
let path = report.write_to(&directory).expect("write the report");
let json: serde_json::Value = serde_json::from_slice(&std::fs::read(&path).expect("read the report")).expect("parse");
let _ = std::fs::remove_dir_all(&directory);
assert_eq!(path.file_name().and_then(|name| name.to_str()), Some("unit.json"));
assert_eq!(json["compared_ops"], 1);
assert_eq!(
json["differences"],
serde_json::json!([{
"op": "GetBucketLocation",
"case": "normal",
"field": "body.location_constraint",
"a": "us-east-1",
"b": "eu-west-1",
}])
);
}
}
@@ -1,197 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Dual-stack differential: one `rustfs` binary, two processes, one SDK script (rustfs/backlog#2740).
//!
//! Stack A always runs `RUSTFS_S3_STACK=legacy`. Stack B runs the value of `RUSTFS_E2E_STACK_B`
//! (default `legacy` until the gateway stack is assembled), and `RUSTFS_E2E_STACK_B_REGION`, when
//! set, starts B in another region so that a difference can be injected by hand. Both servers
//! answer the script in `scenarios`, which covers every operation of `impl S3 for FS`; `compare`
//! reports each `(op, field, a, b)` on which the answers differ, and every run writes its report
//! to `<target>/dual-stack/<run>.json`. Two processes, not two embedded servers: bucket metadata
//! is process-global, so two stacks in one process would read each other's buckets.
mod compare;
mod scenarios;
use crate::common::{RustFSTestEnvironment, init_logging, workspace_root};
use compare::{Difference, Observation, Report, compare, compared_cases, compared_ops, render};
use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};
type BoxError = Box<dyn std::error::Error + Send + Sync>;
const STACK_A: &str = "legacy";
const ENV_STACK_B: &str = "RUSTFS_E2E_STACK_B";
const ENV_STACK_B_REGION: &str = "RUSTFS_E2E_STACK_B_REGION";
/// Becomes `gateway` once rustfs/backlog#2734 T2.5 assembles the gateway stack.
const DEFAULT_STACK_B: &str = "legacy";
/// The region the negative control gives stack A explicitly, so an inherited `RUSTFS_REGION`
/// cannot erase the injected difference.
const CONTROL_REGION_A: &str = "us-east-1";
const CONTROL_REGION_B: &str = "eu-west-1";
/// `normal` cases the legacy stack must answer with a 2xx. The comparison alone cannot tell a
/// script that exercises real behavior from one in which both stacks fail the same way. Five of
/// the 73 fail on legacy by design of this script: PutBucketReplication (no registered remote
/// target), GetBucketReplication (so no configuration), PutBucketNotificationConfiguration (no
/// notification target configured), RestoreObject (the object is not archived) and
/// GetObjectTorrent (legacy answers `NoSuchKey`).
const MIN_NORMAL_SUCCESSES: usize = 68;
struct PairConfig {
stack_b: String,
region_a: Option<String>,
region_b: Option<String>,
}
impl PairConfig {
fn from_env() -> Self {
Self {
stack_b: std::env::var(ENV_STACK_B).unwrap_or_else(|_| DEFAULT_STACK_B.to_owned()),
region_a: None,
region_b: std::env::var(ENV_STACK_B_REGION).ok(),
}
}
}
/// Starts one server on `stack`, overriding its region only when asked.
async fn start_stack(stack: &str, region: Option<&str>) -> Result<RustFSTestEnvironment, BoxError> {
let mut env = RustFSTestEnvironment::new().await?;
let mut vars = vec![("RUSTFS_S3_STACK", stack)];
if let Some(region) = region {
vars.push(("RUSTFS_REGION", region));
}
env.start_rustfs_server_without_cleanup_with_env(&vars).await?;
Ok(env)
}
/// Starts stack A and stack B side by side, each with its own port and data directory.
async fn spawn_pair(config: &PairConfig) -> Result<(RustFSTestEnvironment, RustFSTestEnvironment), BoxError> {
tokio::try_join!(
start_stack(STACK_A, config.region_a.as_deref()),
start_stack(&config.stack_b, config.region_b.as_deref())
)
}
fn report_directory() -> PathBuf {
let target = std::env::var_os("CARGO_TARGET_DIR").map_or_else(|| PathBuf::from("target"), PathBuf::from);
let target = if target.is_absolute() {
target
} else {
workspace_root().join(target)
};
target.join("dual-stack")
}
struct Run {
report: Report,
path: PathBuf,
a: Vec<Observation>,
}
/// Runs the script against both stacks concurrently and writes the report.
async fn run_differential(label: &str, config: &PairConfig) -> Run {
let (server_a, server_b) = spawn_pair(config).await.expect("start stack A and stack B");
let (a, b) = tokio::join!(Box::pin(scenarios::run(&server_a)), Box::pin(scenarios::run(&server_b)));
let millis = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map_or(0, |elapsed| elapsed.as_millis());
let report = Report {
schema: 1,
run: format!("{label}-{millis}-{}", std::process::id()),
stack_a: STACK_A.to_owned(),
stack_b: config.stack_b.clone(),
region_a: config.region_a.clone(),
region_b: config.region_b.clone(),
compared_ops: compared_ops(&a, &b).len(),
compared_cases: compared_cases(&a, &b),
differences: compare(&a, &b),
};
let path = report.write_to(&report_directory()).expect("write the dual-stack report");
Run { report, path, a }
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn both_stacks_answer_every_operation_identically() {
init_logging();
let config = PairConfig::from_env();
let run = run_differential("self-comparison", &config).await;
let report = &run.report;
let mut observed: Vec<_> = run.a.iter().map(|observation| observation.op).collect();
observed.sort_unstable();
observed.dedup();
assert_eq!(
observed,
scenarios::OPERATIONS,
"the script must drive exactly the operations of impl S3 for FS"
);
assert_eq!(report.compared_ops, scenarios::OPERATIONS.len(), "report: {}", run.path.display());
assert!(
report.differences.is_empty(),
"stack {} and stack {} differ on {} of {} cases (report: {}):\n{}",
report.stack_a,
report.stack_b,
report.differences.len(),
report.compared_cases,
run.path.display(),
render(&report.differences)
);
let not_4xx: Vec<_> = run
.a
.iter()
.filter(|observation| observation.case == "negative" && !(400..500).contains(&observation.status))
.collect();
assert!(not_4xx.is_empty(), "negative cases must fail with a 4xx on {STACK_A}: {not_4xx:#?}");
let successes = run
.a
.iter()
.filter(|observation| observation.case == "normal" && observation.is_success())
.count();
assert!(
successes >= MIN_NORMAL_SUCCESSES,
"only {successes} normal cases succeeded on {STACK_A}: {:#?}",
run.a
);
}
#[tokio::test]
async fn differential_detects_an_injected_difference() {
init_logging();
let config = PairConfig {
stack_b: STACK_A.to_owned(),
region_a: Some(CONTROL_REGION_A.to_owned()),
region_b: Some(CONTROL_REGION_B.to_owned()),
};
let run = run_differential("injected-region", &config).await;
let expected = Difference {
op: "GetBucketLocation".to_owned(),
case: "normal".to_owned(),
field: "body.location_constraint".to_owned(),
a: CONTROL_REGION_A.to_owned(),
b: CONTROL_REGION_B.to_owned(),
};
assert!(
run.report.differences.contains(&expected),
"B started in {CONTROL_REGION_B} but GetBucketLocation was not reported (report: {}):\n{}",
run.path.display(),
render(&run.report.differences)
);
}
}
File diff suppressed because it is too large Load Diff
@@ -1969,11 +1969,16 @@ mod tests {
let status_body = signed_admin_post(&status_url, None, &cluster.access_key, &cluster.secret_key).await?;
let status: serde_json::Value = serde_json::from_str(&status_body)
.map_err(|err| format!("pre-interrupt background heal status is not JSON ({err}): {status_body}"))?;
if background_enabled || only_admin_heal_is_active(&status) {
let admin_active = background_rejoin_heal_evidence
|| (status["state"].as_str() == Some("active")
&& status["healOperations"]["activeBySource"]["admin"].as_u64() == Some(1));
if admin_active && (background_enabled || only_admin_heal_is_active(&status)) {
break status;
}
if Instant::now() >= partial_deadline {
return Err(format!("non-admin heal did not drain before the interruption deadline: {status}").into());
return Err(
format!("heal tasks did not reach the required interruption state before the deadline: {status}").into(),
);
}
sleep(Duration::from_millis(50)).await;
};
@@ -2667,7 +2672,12 @@ mod tests {
}
}
async fn verify_two_node_directory_recovery(remove_bucket: bool, scan_mode: u8) -> Result<(), Box<dyn Error + Send + Sync>> {
async fn verify_two_node_directory_recovery(
remove_bucket: bool,
scan_mode: u8,
bucket_scoped: bool,
recreate: bool,
) -> Result<(), Box<dyn Error + Send + Sync>> {
init_logging();
let mut cluster = RustFSTestClusterEnvironment::new(2).await?;
// Isolate explicit administrator recovery from scanner, MRF, and disk healing.
@@ -2714,27 +2724,13 @@ mod tests {
tokio::fs::create_dir(&removed).await?;
if remove_bucket {
tokio::fs::rename(&target_bucket, removed.join(bucket)).await?;
let status = two_node_heal_status(&cluster, None, false, scan_mode).await?;
assert_eq!(
status["summary"], "stopped",
"missing bucket must not look like an empty successful scan: {status}"
);
assert_eq!(status["outcome"]["execution"]["state"], "aborted");
assert_eq!(status["outcome"]["coverage"], "partial");
assert_eq!(status["outcome"]["counters"]["processed"], 0);
assert!(!target_bucket.exists(), "default discovery must not silently recreate a bucket");
let scoped_default = two_node_heal_status(&cluster, Some(bucket), false, scan_mode).await?;
assert_eq!(
scoped_default["summary"], "stopped",
"default namespace quorum protection must remain: {scoped_default}"
);
} else {
for (index, (key, _)) in objects.iter().enumerate() {
tokio::fs::rename(target_bucket.join(key), removed.join(index.to_string())).await?;
}
}
let status = two_node_heal_status(&cluster, remove_bucket.then_some(bucket), remove_bucket, scan_mode).await?;
assert_eq!(status["summary"], "finished", "explicit recovery should complete: {status}");
let status = two_node_heal_status(&cluster, bucket_scoped.then_some(bucket), recreate, scan_mode).await?;
assert_eq!(status["summary"], "finished", "administrator recovery should complete: {status}");
assert_eq!(status["outcome"]["execution"]["state"], "completed");
assert_eq!(status["outcome"]["coverage"], "complete");
for ((key, _), before) in objects.iter().zip(&original) {
@@ -2773,16 +2769,26 @@ mod tests {
#[tokio::test]
async fn test_two_node_bucket_discovery_and_explicit_recreate() -> Result<(), Box<dyn Error + Send + Sync>> {
verify_two_node_directory_recovery(true, 2).await
verify_two_node_directory_recovery(true, 2, true, true).await
}
#[tokio::test]
async fn test_two_node_missing_bucket_default_cluster_recovery() -> Result<(), Box<dyn Error + Send + Sync>> {
verify_two_node_directory_recovery(true, 2, false, false).await
}
#[tokio::test]
async fn test_two_node_missing_bucket_default_scoped_recovery() -> Result<(), Box<dyn Error + Send + Sync>> {
verify_two_node_directory_recovery(true, 2, true, false).await
}
#[tokio::test]
async fn test_two_node_object_directory_recovery_normal() -> Result<(), Box<dyn Error + Send + Sync>> {
verify_two_node_directory_recovery(false, 1).await
verify_two_node_directory_recovery(false, 1, false, false).await
}
#[tokio::test]
async fn test_two_node_object_directory_recovery_deep() -> Result<(), Box<dyn Error + Send + Sync>> {
verify_two_node_directory_recovery(false, 2).await
verify_two_node_directory_recovery(false, 2, false, false).await
}
}
-6
View File
@@ -31,12 +31,6 @@ pub mod chaos;
#[cfg(test)]
pub mod fake_s3_target;
// Dual-stack differential (rustfs/backlog#2740): one binary, two processes on
// different RUSTFS_S3_STACK values, the same SDK script for every S3 operation,
// compared on status, error code, allowlisted headers and typed body fields.
#[cfg(test)]
mod dual_stack_differential;
// On-demand migration (backlog#2147): shared two-server environment, admin
// wrappers, and the harness self-test (backlog#2151). Behavior scenarios are
// added by later ODM tasks.
@@ -3945,7 +3945,15 @@ async fn test_bucket_replication_version_purge_of_non_inline_object_releases_sou
(sse_key, sse_put.version_id().ok_or("SSE PUT omitted version ID")?.to_string()),
(large_key, large_put.version_id().ok_or("large PUT omitted version ID")?.to_string()),
];
assert_replication_converged(&source_client, &source_bucket, &target_client, &target_bucket).await?;
// Managed-SSE listings may omit ETags. Check the replicated version's body
// before deleting it rather than comparing keyless listing fingerprints.
for (key, version_id) in &purged {
let source_body = get_version_body(&source_client, &source_bucket, key, version_id).await?;
let expected_sha256: [u8; 32] = Sha256::digest(&source_body).into();
wait_for_replicated_sha256(&target_client, &target_bucket, key, expected_sha256).await?;
let target_body = get_version_body(&target_client, &target_bucket, key, version_id).await?;
assert_eq!(target_body, source_body, "replicated version body differs for {key}/{version_id}");
}
for (key, version_id) in &purged {
source_client
@@ -3956,10 +3964,6 @@ async fn test_bucket_replication_version_purge_of_non_inline_object_releases_sou
.send()
.await?;
}
assert_replication_converged(&source_client, &source_bucket, &target_client, &target_bucket).await?;
let target_state = list_replication_state(&target_client, &target_bucket).await?;
assert!(target_state.is_empty(), "target retained an explicitly purged version: {target_state:?}");
// The purge state is applied on the source asynchronously after the target
// acknowledges the delete; only then does the retained version go away and
// the bucket become deletable. A listing that is empty while DeleteBucket
@@ -3984,6 +3988,9 @@ async fn test_bucket_replication_version_purge_of_non_inline_object_releases_sou
}
}
let target_state = list_replication_state(&target_client, &target_bucket).await?;
assert!(target_state.is_empty(), "target retained an explicitly purged version: {target_state:?}");
Ok(())
}
+5 -4
View File
@@ -425,10 +425,11 @@ pub mod disk {
BATCH_READ_VERSION_MAX_ITEMS, BUCKET_META_PREFIX, BatchReadVersionItem, BatchReadVersionReq, BatchReadVersionResp,
CheckPartsResp, ConditionalFileUpdate, DELETE_RECOVERY_PROTOCOL_VERSION, DeleteOptions, DeleteRecoveryAction,
DeleteRecoveryCursor, DeleteRecoveryDigest, DeleteRecoveryInspection, DeleteRecoveryPage, DeleteRecoveryRpcOperation,
DeleteRecoveryRpcReply, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, DiskStore, FileInfoVersions,
FileReader, FileWriter, HEALING_MARKER_PATH, NsScannerOpenRequest, OldCurrentSize, PartTransactionAction,
RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE,
SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk, validate_batch_read_version_item_count,
DeleteRecoveryRpcReply, DeleteRecoveryState, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption,
DiskStore, FileInfoVersions, FileReader, FileWriter, HEALING_MARKER_PATH, NsScannerOpenRequest, OldCurrentSize,
PartTransactionAction, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
STORAGE_FORMAT_FILE, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk,
validate_batch_read_version_item_count,
};
pub use bytes::Bytes;
pub use endpoint::Endpoint;
@@ -14277,11 +14277,7 @@ mod tests {
#[tokio::test]
#[serial]
async fn tier_free_version_recovery_production_entrypoint_enqueues_seeded_item() {
let (disk_paths, ecstore) =
temp_env::async_with_vars([(super::ENV_TIER_FREE_VERSION_RECOVERY_ENABLED, Some("false"))], async {
setup_test_env().await
})
.await;
let (disk_paths, ecstore) = setup_test_env().await;
let runtime_state = install_unconsumed_runtime_expiry_worker(&ecstore, 1).await;
let recovery_rx = {
let state = runtime_state.read().await;
+5 -23
View File
@@ -439,8 +439,6 @@ pub struct BucketMetadata {
pub bucket_incarnation_id: Uuid,
pub(crate) bucket_incarnation_sidecar: bool,
pub(crate) bucket_creation_committed: bool,
// A legacy default is not proof of an interrupted creation.
pub(crate) bucket_creation_commit_record_present: bool,
pub policy_config_json: Vec<u8>,
pub notification_config_xml: Vec<u8>,
pub lifecycle_config_xml: Vec<u8>,
@@ -516,7 +514,6 @@ impl Default for BucketMetadata {
bucket_incarnation_id: Uuid::nil(),
bucket_incarnation_sidecar: false,
bucket_creation_committed: false,
bucket_creation_commit_record_present: false,
policy_config_json: Default::default(),
notification_config_xml: Default::default(),
lifecycle_config_xml: Default::default(),
@@ -755,10 +752,7 @@ impl BucketMetadata {
"TableBucketConfigUpdatedAt" => self.table_bucket_config_updated_at = read_msgp_time_value(rd)?,
"DurabilityConfigUpdatedAt" => self.durability_config_updated_at = read_msgp_time_value(rd)?,
"OnDemandMigrationConfigUpdatedAt" => self.on_demand_migration_config_updated_at = read_msgp_time_value(rd)?,
"BucketCreationCommitted" => {
self.bucket_creation_committed = read_msgp_bool(rd)?;
self.bucket_creation_commit_record_present = true;
}
"BucketCreationCommitted" => self.bucket_creation_committed = read_msgp_bool(rd)?,
other => {
tracing::debug!(field = %other, "BucketMetadata decode_from: skipping unknown field");
skip_msgp_value(rd)?;
@@ -771,10 +765,8 @@ impl BucketMetadata {
/// Encode to msgp bytes. Field order follows MinIO BucketMetadata for compatibility.
pub fn encode_to<W: Write>(&self, wr: &mut W) -> Result<()> {
// Preserve the absence of a legacy commit record across config rewrites.
let commit_record_present = self.bucket_creation_commit_record_present || self.bucket_creation_committed;
// Map size: MinIO fields (25) + RustFS extensions (21) + optional commit record.
let map_len: u32 = 46 + u32::from(commit_record_present);
// Map size: MinIO fields (25) + RustFS extensions (22)
let map_len: u32 = 47;
rmp::encode::write_map_len(wr, map_len)?;
// MinIO field order (same as Go struct)
@@ -855,10 +847,8 @@ impl BucketMetadata {
write_msgp_time(wr, self.durability_config_updated_at)?;
rmp::encode::write_str(wr, "OnDemandMigrationConfigUpdatedAt")?;
write_msgp_time(wr, self.on_demand_migration_config_updated_at)?;
if commit_record_present {
rmp::encode::write_str(wr, "BucketCreationCommitted")?;
rmp::encode::write_bool(wr, self.bucket_creation_committed)?;
}
rmp::encode::write_str(wr, "BucketCreationCommitted")?;
rmp::encode::write_bool(wr, self.bucket_creation_committed)?;
Ok(())
}
@@ -1701,13 +1691,6 @@ mod test {
let mut bm = BucketMetadata::unmarshal(&body[4..]).expect("unmarshal recovered blob");
assert_eq!(bm.name, "interop");
assert!(!bm.bucket_creation_committed, "legacy metadata must default to uncommitted");
assert!(!bm.bucket_creation_commit_record_present, "legacy metadata has no creation-commit record");
let rewritten = BucketMetadata::unmarshal(&bm.marshal_msg().expect("rewrite legacy metadata"))
.expect("read rewritten legacy metadata");
assert!(
!rewritten.bucket_creation_commit_record_present,
"rewriting legacy metadata must not fabricate a creation intent"
);
bm.parse_all_configs().expect("parse recovered configs");
assert!(bm.lifecycle_config.is_some());
}
@@ -1725,7 +1708,6 @@ mod test {
assert_eq!(bm.name, new.name);
assert!(new.bucket_creation_committed);
assert!(new.bucket_creation_commit_record_present);
assert!(!bm.bucket_incarnation_id.is_nil());
assert_eq!(bm.bucket_incarnation_id, new.bucket_incarnation_id);
}
+30 -283
View File
@@ -525,6 +525,36 @@ pub(crate) async fn get_bucket_incarnation_id_in(ctx: &crate::runtime::instance:
sys.get_bucket_incarnation_id_from_disk(bucket).await
}
/// Directory loss is recoverable only for a committed, non-retired generation.
/// The caller retains the bucket lifecycle fence through the ensuing reconstruction.
pub(crate) async fn validate_bucket_reconstruction_in(
ctx: &crate::runtime::instance::InstanceContext,
bucket: &str,
expected: Uuid,
) -> Result<()> {
let api = bucket_metadata_sys_of(ctx)?.read().await.object_store();
let lock = api
.new_ns_lock(RUSTFS_META_BUCKET, &bucket_metadata_transaction_lock_key(bucket))
.await?;
let guard = lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
let (metadata, persisted) = load_bucket_metadata_parse_with_presence(api.clone(), bucket, true).await?;
if !persisted
|| !metadata.bucket_incarnation_sidecar
|| expected.is_nil()
|| metadata.bucket_incarnation_id != expected
|| !metadata.bucket_creation_committed
{
return Err(Error::PreconditionFailed);
}
if crate::bucket::retirement::is_retired(api, bucket, expected).await? {
return Err(Error::BucketNotFound(bucket.to_owned()));
}
if guard.is_lock_lost() {
return Err(Error::PreconditionFailed);
}
Ok(())
}
/// Validate against disk while retaining an already-held Object Lock fence.
pub(crate) async fn get_bucket_incarnation_id_for_options_in(
ctx: &crate::runtime::instance::InstanceContext,
@@ -1619,27 +1649,9 @@ struct MetadataPublishGuard {
_guard: tokio::sync::OwnedMutexGuard<MetadataPublishLockState>,
}
#[derive(Debug, Clone)]
pub(crate) struct PendingBucketCreationProof {
pub(crate) generation: Uuid,
pub(crate) intent: Vec<u8>,
pub(crate) versioning_requested: bool,
pub(crate) witnesses: Vec<crate::disk::os::NativeBucketCreationWitness>,
}
impl PendingBucketCreationProof {
pub(crate) fn matches_metadata(&self, metadata: &BucketMetadata) -> Result<bool> {
Ok(!metadata.bucket_creation_committed
&& metadata.bucket_incarnation_sidecar
&& metadata.bucket_incarnation_id == self.generation
&& metadata.marshal_msg()? == self.intent)
}
}
#[derive(Debug, Clone)]
pub struct BucketMetadataSys {
metadata_map: Arc<RwLock<HashMap<String, Arc<BucketMetadata>>>>,
pending_bucket_creations: Arc<RwLock<HashMap<String, PendingBucketCreationProof>>>,
/// Serializes metadata-map commits and their derived cache updates for one
/// bucket. Namespace locks, when present, are acquired before this lock.
metadata_publish_locks: Arc<MetadataPublishLockRegistry>,
@@ -1683,7 +1695,6 @@ impl BucketMetadataSys {
pub fn new(api: Arc<ECStore>) -> Self {
Self {
metadata_map: Arc::new(RwLock::new(HashMap::new())),
pending_bucket_creations: Arc::new(RwLock::new(HashMap::new())),
metadata_publish_locks: Arc::new(MetadataPublishLockRegistry {
locks: StdMutex::new(HashMap::new()),
}),
@@ -1714,81 +1725,6 @@ impl BucketMetadataSys {
.expect("bucket metadata object store should still be live")
}
pub(crate) async fn pending_bucket_creation(&self, bucket: &str) -> Option<PendingBucketCreationProof> {
self.pending_bucket_creations.read().await.get(bucket).cloned()
}
pub(crate) async fn grant_pending_bucket_creation(&self, bucket: &str, proof: PendingBucketCreationProof) {
self.pending_bucket_creations.write().await.insert(bucket.to_owned(), proof);
}
pub(crate) async fn revoke_pending_bucket_creation(&self, bucket: &str) {
self.pending_bucket_creations.write().await.remove(bucket);
}
pub(crate) async fn consume_pending_bucket_creation(
&self,
bucket: &str,
metadata: &BucketMetadata,
witnesses: &[crate::disk::os::NativeBucketCreationWitness],
) -> Result<bool> {
let mut pending = self.pending_bucket_creations.write().await;
let Some(proof) = pending.get(bucket) else {
return Ok(false);
};
if !proof.matches_metadata(metadata)?
|| proof.witnesses.len() != witnesses.len()
|| !proof
.witnesses
.iter()
.zip(witnesses)
.all(|(old, current)| old.matches(current))
|| !crate::disk::os::consume_native_creation_witnesses(witnesses)
{
return Ok(false);
}
pending.remove(bucket);
Ok(true)
}
#[cfg(test)]
pub(crate) async fn refresh_bucket_for_test(&self, bucket: &str) -> Result<()> {
let expected = self.metadata_map.read().await.get(bucket).cloned();
let lock = self.object_store().new_ns_lock(bucket, bucket).await?;
let guard = lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
self.load_bucket_under_namespace(bucket, MetadataLoadMode::Refresh, expected.as_ref(), &guard)
.await
}
#[cfg(test)]
pub(crate) async fn publish_persisted_metadata_for_test(&self, bucket: &str) -> Result<()> {
let expected = self.metadata_map.read().await.get(bucket).cloned();
let lock = self.object_store().new_ns_lock(bucket, bucket).await?;
let guard = lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
let (metadata, persisted) = await_bucket_namespace_operation(
Some(&guard),
bucket,
"test cached metadata load",
load_bucket_metadata_parse_with_presence(self.object_store(), bucket, true),
)
.await?;
self.publish_if_unchanged(bucket, expected.as_ref(), metadata, persisted, &guard)
.await
}
#[cfg(test)]
pub(crate) async fn migrate_bucket_creation_commit_for_test(&self, bucket: &str) -> Result<()> {
self.migrate_bucket_creation_commit(bucket).await.map(|_| ())
}
#[cfg(test)]
pub(crate) async fn legacy_migration_reports_missing_bucket_for_test(&self, bucket: &str) -> Result<bool> {
Ok(matches!(
self.migrate_legacy_metadata(bucket).await?,
BucketMetadataAuthority::MissingBucket
))
}
fn object_store_if_live(&self) -> Option<Arc<ECStore>> {
self.api.upgrade()
}
@@ -1979,7 +1915,6 @@ impl BucketMetadataSys {
expected: Option<&Arc<BucketMetadata>>,
namespace_guard: &rustfs_lock::NamespaceLockGuard,
) -> Result<()> {
self.revoke_pending_bucket_creation(bucket).await;
if !await_bucket_namespace_operation(
Some(namespace_guard),
bucket,
@@ -2063,7 +1998,6 @@ impl BucketMetadataSys {
persisted: bool,
namespace_guard: &rustfs_lock::NamespaceLockGuard,
) -> Result<()> {
self.revoke_pending_bucket_creation(bucket).await;
if !persisted {
let _publish_guard = self
.lock_metadata_publish(bucket, namespace_guard, "refreshed bucket metadata absence publish")
@@ -2125,7 +2059,6 @@ impl BucketMetadataSys {
}
pub async fn set(&self, bucket: String, bm: Arc<BucketMetadata>) {
self.revoke_pending_bucket_creation(&bucket).await;
if !is_meta_bucketname(&bucket) {
let publish_lock = self.metadata_publish_lock(&bucket);
let _publish_guard = publish_lock.lock().await;
@@ -2147,7 +2080,6 @@ impl BucketMetadataSys {
if is_meta_bucketname(bucket) {
return false;
}
self.revoke_pending_bucket_creation(bucket).await;
let publish_lock = self.metadata_publish_lock(bucket);
let _publish_guard = publish_lock.lock().await;
let mut map = self.metadata_map.write().await;
@@ -2165,7 +2097,6 @@ impl BucketMetadataSys {
}
async fn _reset(&mut self) {
self.pending_bucket_creations.write().await.clear();
let mut map = self.metadata_map.write().await;
map.clear();
drop(map);
@@ -2304,7 +2235,6 @@ impl BucketMetadataSys {
/// server's metadata never leaks into the ambient (first) instance.
pub(crate) async fn persist_and_set(&self, bm: BucketMetadata) -> Result<()> {
let mut bm = bm;
self.revoke_pending_bucket_creation(&bm.name).await;
bm.save_with_store(self.object_store()).await?;
self.set(bm.name.clone(), Arc::new(bm)).await;
@@ -2313,7 +2243,6 @@ impl BucketMetadataSys {
}
async fn persist_new_and_set(&self, mut bm: BucketMetadata) -> Result<()> {
self.revoke_pending_bucket_creation(&bm.name).await;
bm.bucket_creation_committed = true;
bm.save_with_store_committed(self.object_store()).await?;
save_bucket_incarnation(self.object_store(), &bm.name, bm.bucket_incarnation_id).await?;
@@ -2323,9 +2252,7 @@ impl BucketMetadataSys {
}
async fn persist_new_bucket_metadata_intent(&self, mut bm: BucketMetadata) -> Result<()> {
self.revoke_pending_bucket_creation(&bm.name).await;
bm.bucket_creation_committed = false;
bm.bucket_creation_commit_record_present = true;
bm.save_with_store(self.object_store()).await?;
save_bucket_incarnation(self.object_store(), &bm.name, bm.bucket_incarnation_id).await?;
bm.bucket_incarnation_sidecar = true;
@@ -2335,7 +2262,6 @@ impl BucketMetadataSys {
}
async fn commit_bucket_metadata(&self, mut bm: BucketMetadata) -> Result<()> {
self.revoke_pending_bucket_creation(&bm.name).await;
bm.parse_all_configs()?;
bm.bucket_incarnation_sidecar = true;
// The caller holds the full bucket creation fence (lifecycle, metadata
@@ -2384,7 +2310,6 @@ impl BucketMetadataSys {
bucket: &str,
namespace_guard: &rustfs_lock::NamespaceLockGuard,
) -> Result<()> {
self.revoke_pending_bucket_creation(bucket).await;
let expected = self.metadata_map.read().await.get(bucket).cloned();
if !self
.bucket_exists(bucket, namespace_guard, "peer bucket metadata existence check")
@@ -2441,7 +2366,6 @@ impl BucketMetadataSys {
#[cfg(test)]
self.lazy_disk_loads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
self.revoke_pending_bucket_creation(bucket).await;
let lock = self.object_store().new_ns_lock(bucket, bucket).await?;
let guard = lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
#[cfg(test)]
@@ -2750,7 +2674,6 @@ impl BucketMetadataSys {
/// revalidated at write quorum, so a stale snapshot can neither revert an
/// acknowledged configuration update nor outlive a delete/recreate.
async fn migrate_bucket_creation_commit(&self, bucket: &str) -> Result<BucketMetadataAuthority> {
self.revoke_pending_bucket_creation(bucket).await;
let transaction_lock = self
.object_store()
.new_ns_lock(RUSTFS_META_BUCKET, &bucket_metadata_transaction_lock_key(bucket))
@@ -2828,7 +2751,6 @@ impl BucketMetadataSys {
&self,
bucket: &str,
) -> Result<BucketMetadataAuthority> {
self.revoke_pending_bucket_creation(bucket).await;
#[cfg(test)]
if self.object_lock_disk_read_errors.write().await.remove(bucket) {
return Err(Error::other(format!("injected Object Lock metadata disk read failure: {bucket}")));
@@ -3221,7 +3143,6 @@ mod tests {
};
use crate::bucket::target::{BucketTarget, BucketTargetType, Credentials};
use crate::config::com::read_config;
use crate::disk::{DiskAPI as _, RUSTFS_META_TMP_BUCKET};
use crate::storage_api_contracts::bucket::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions};
use byteorder::{ByteOrder as _, LittleEndian};
use serial_test::serial;
@@ -3276,180 +3197,6 @@ mod tests {
),
];
#[tokio::test]
#[serial(storage_class_env)]
async fn persistence_entry_failures_revoke_pending_creation_before_cache_publish() {
let (dirs, store) = isolated_store_over_temp_disks().await;
init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let sys = bucket_metadata_sys_of(&store.ctx).expect("instance metadata system");
let sys = sys.read().await.clone();
let options = MakeBucketOptions {
lock_enabled: true,
..Default::default()
};
for entry in ["save", "persist", "persist-new", "commit"] {
let bucket = format!("revoke-{entry}-{}", Uuid::new_v4());
store.fail_next_bucket_creation_after_intent_for_test(&bucket);
store
.make_bucket(&bucket, &options)
.await
.expect_err("stop the real producer before physical publication");
sys.pending_bucket_creation(&bucket)
.await
.expect("the producer must leave a real live creation permission");
let (intent, persisted) = get_config_from_disk_with_presence_in(&store.ctx, &bucket)
.await
.expect("read the original producer intent");
assert!(persisted, "{entry}: the complete original intent must be persisted");
let original_intent = intent.marshal_msg().expect("encode the original intent");
// Keep the creation fence order used by the real producer.
let lifecycle_guard = store
.acquire_bucket_lifecycle_write_lock(&bucket)
.await
.expect("hold the bucket lifecycle fence");
let transaction_guard = acquire_bucket_metadata_transaction_lock_in(&store.ctx, &bucket)
.await
.expect("hold the metadata transaction fence");
let namespace_lock = store
.new_ns_lock(&bucket, &bucket)
.await
.expect("create the exact bucket namespace lock");
let namespace_guard = namespace_lock
.get_write_lock(crate::set_disk::get_lock_acquire_timeout())
.await
.expect("hold the exact bucket namespace fence");
// A file at each native staging root causes actual IO to fail before
// metadata can reach the cache-publishing set call. Preserve the
// original staging directories so no fixture data is discarded.
for dir in &dirs {
let staging = dir.path().join(RUSTFS_META_TMP_BUCKET);
let saved = dir.path().join(".rustfs.sys/revocation-saved-tmp");
std::fs::rename(&staging, &saved).expect("preserve the native staging directory");
std::fs::write(&staging, b"block metadata staging with a non-directory")
.expect("install an actual non-directory IO obstruction");
}
let result = match entry {
"save" => sys.save(intent).await,
"persist" => sys.persist_and_set(intent).await,
"persist-new" => sys.persist_new_and_set(intent).await,
"commit" => sys.commit_bucket_metadata(intent).await,
_ => unreachable!(),
};
for dir in &dirs {
let staging = dir.path().join(RUSTFS_META_TMP_BUCKET);
let saved = dir.path().join(".rustfs.sys/revocation-saved-tmp");
std::fs::remove_file(&staging).expect("remove only the fixture's staging obstruction");
std::fs::rename(&saved, &staging).expect("restore the complete original staging directory");
}
drop((namespace_guard, transaction_guard, lifecycle_guard));
result.expect_err("native persistence must fail before its downstream cache set");
assert!(
sys.pending_bucket_creation(&bucket).await.is_none(),
"{entry}: the entry revoker must retire live permission even when persistence fails"
);
let (after_failure, present) = get_config_from_disk_with_presence_in(&store.ctx, &bucket)
.await
.expect("read the intent after the failed persistence entry");
assert!(present, "{entry}: a failed persistence entry must retain the original metadata");
assert_eq!(
after_failure
.marshal_msg()
.expect("encode the original intent after persistence failure"),
original_intent,
"{entry}: native IO failure must not change the original generation or complete intent"
);
let error = store
.make_bucket(&bucket, &options)
.await
.expect_err("a failed ordinary persistence entry cannot preserve creation permission");
assert!(error.to_string().contains("administrator recovery"), "{entry}: {error}");
let (after_retry, _) = get_config_from_disk_with_presence_in(&store.ctx, &bucket)
.await
.expect("read the original intent after the rejected retry");
assert_eq!(
after_retry.marshal_msg().expect("encode the intent after the rejected retry"),
original_intent,
"{entry}: the rejected retry must preserve the complete original intent"
);
for disk in store.pools[0].disk_set[0].disks.read().await.iter() {
assert!(
matches!(
disk.as_ref().expect("native disk").stat_volume(&bucket).await,
Err(crate::disk::error::DiskError::VolumeNotFound)
),
"{entry}: failed persistence and the rejected retry must not publish a physical volume"
);
}
}
}
#[tokio::test]
#[serial(storage_class_env)]
async fn metadata_reset_revokes_real_pending_creation_without_changing_persisted_intent() {
let (_dirs, store) = isolated_store_over_temp_disks().await;
init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let bucket = format!("revoke-reset-{}", Uuid::new_v4());
let options = MakeBucketOptions {
lock_enabled: true,
..Default::default()
};
store.fail_next_bucket_creation_after_intent_for_test(&bucket);
store
.make_bucket(&bucket, &options)
.await
.expect_err("stop the real producer before physical publication");
let sys = bucket_metadata_sys_of(&store.ctx).expect("instance metadata system");
let mut sys = sys.read().await.clone();
sys.pending_bucket_creation(&bucket)
.await
.expect("the real producer must leave live creation permission before reset");
let (intent, persisted) = get_config_from_disk_with_presence_in(&store.ctx, &bucket)
.await
.expect("read the original intent before cache reset");
assert!(persisted, "cache reset starts from a complete real persisted intent");
let original_intent = intent.marshal_msg().expect("encode the original reset intent");
sys._reset().await;
assert!(
sys.pending_bucket_creation(&bucket).await.is_none(),
"reset must revoke the producer-issued permission shared with the instance metadata system"
);
let (after_reset, present) = get_config_from_disk_with_presence_in(&store.ctx, &bucket)
.await
.expect("read the original intent after cache reset");
assert!(present, "cache reset must retain the persisted intent");
assert_eq!(
after_reset.marshal_msg().expect("encode the intent after cache reset"),
original_intent,
"cache reset must leave the original generation and complete persisted intent unchanged"
);
let error = store
.make_bucket(&bucket, &options)
.await
.expect_err("reset cannot regrant the producer's old creation permission");
assert!(error.to_string().contains("administrator recovery"), "{error}");
let (after_retry, _) = get_config_from_disk_with_presence_in(&store.ctx, &bucket)
.await
.expect("read the intent after the reset-induced retry rejection");
assert_eq!(
after_retry
.marshal_msg()
.expect("encode the intent after the rejected reset retry"),
original_intent,
"the reset-induced rejection must preserve the original complete intent"
);
for disk in store.pools[0].disk_set[0].disks.read().await.iter() {
assert!(
matches!(
disk.as_ref().expect("native disk").stat_volume(&bucket).await,
Err(crate::disk::error::DiskError::VolumeNotFound)
),
"cache reset and the rejected retry must not publish a physical volume"
);
}
}
#[tokio::test]
async fn g_d3_003_new_writer_replication_loads_without_fail_closed_state() {
let (dirs, store) = isolated_store_over_temp_disks().await;
@@ -65,9 +65,9 @@ use super::replication_target_boundary::{
SsecPassthroughGate, TargetClient, VersionIdentityCapability, is_replication_target_offline_error,
replication_action_for_target_head, replication_complete_multipart_options, replication_delete_marker_purge_remove_options,
replication_delete_remove_options, replication_force_delete_remove_options, replication_object_is_ssec_encrypted,
replication_put_object_header_size, replication_put_object_options, replication_target_head_is_newer_null_version,
resolve_read_api_version_id, ssec_passthrough_evidence_present, ssec_passthrough_gate, version_identity_capability_from_put,
version_identity_drifted,
replication_put_object_header_size, replication_put_object_options, replication_source_etag_is_comparable,
replication_target_head_is_newer_null_version, resolve_read_api_version_id, ssec_passthrough_evidence_present,
ssec_passthrough_gate, version_identity_capability_from_put, version_identity_drifted,
};
use super::replication_versioning_boundary::ReplicationVersioningStore;
use super::runtime_boundary as runtime_sources;
@@ -349,15 +349,22 @@ fn replica_etag_verification_enabled() -> bool {
/// target that stores a transformed payload (e.g. undecoded `aws-chunked`
/// frames, #6853) returns the ETag of what it actually wrote. Reporting
/// COMPLETED over such a replica is silent corruption, so a decidable
/// mismatch fails the replication instead. An SSE-C ciphertext passthrough
/// transfer is exempt: the wire bytes are ciphertext while the source ETag is
/// the plaintext MD5, and that path has its own HEAD-back audit.
/// mismatch fails the replication instead. Encrypted sources are exempt: an
/// SSE-C ciphertext passthrough sends ciphertext while the source ETag is the
/// plaintext MD5 (that path has its own HEAD-back audit), and a managed-SSE
/// (SSE-S3 / SSE-KMS) source stores a keyed fingerprint, or a legacy plaintext
/// MD5, that the ETag of the target's re-encrypted replica can neither confirm
/// nor refute. The checksum header on the replication PUT stays the integrity
/// check for those objects.
fn verify_single_part_replica(
object_info: &ObjectInfo,
response: &RemotePutObjectResponse,
ciphertext_passthrough: bool,
) -> std::result::Result<(), std::io::Error> {
if ciphertext_passthrough || !replica_etag_verification_enabled() {
if ciphertext_passthrough
|| !replication_source_etag_is_comparable(&object_info.user_defined)
|| !replica_etag_verification_enabled()
{
return Ok(());
}
if single_part_replica_etag_mismatch(object_info.etag.as_deref(), response.etag.as_deref()) {
@@ -5870,6 +5877,57 @@ mod tests {
use super::super::replication_filemeta_boundary::ReplicateTargetDecision;
use super::super::replication_object_decision_boundary::ReplicationMultipartPlanError;
#[test]
fn verify_single_part_replica_exempts_managed_sse_sources() {
// A new encrypted write stores a keyed fingerprint; a target that
// computes its own ETag answers with the MD5 of the plaintext it
// received. Both are 32 hex characters, so without the exemption the
// audit would fail every SSE-S3 / SSE-KMS replica (rustfs/backlog).
let fingerprint = "00112233445566778899aabbccddeeff";
let replica_md5 = "ffeeddccbbaa99887766554433221100";
let response = RemotePutObjectResponse {
version_id: None,
etag: Some(replica_md5.to_string()),
};
for sse in ["AES256", "aws:kms"] {
let object_info = ObjectInfo {
bucket: "source".to_string(),
name: "encrypted".to_string(),
etag: Some(fingerprint.to_string()),
user_defined: HashMap::from([("x-amz-server-side-encryption".to_string(), sse.to_string())]).into(),
..Default::default()
};
verify_single_part_replica(&object_info, &response, false)
.unwrap_or_else(|err| panic!("{sse}: a managed-SSE source must not be audited against the replica MD5: {err}"));
}
}
#[test]
fn verify_single_part_replica_still_fails_plaintext_mismatch() {
if !replica_etag_verification_enabled() {
return;
}
let object_info = ObjectInfo {
bucket: "source".to_string(),
name: "plain".to_string(),
etag: Some("00112233445566778899aabbccddeeff".to_string()),
..Default::default()
};
let mismatch = RemotePutObjectResponse {
version_id: None,
etag: Some("ffeeddccbbaa99887766554433221100".to_string()),
};
let err = verify_single_part_replica(&object_info, &mismatch, false)
.expect_err("a plaintext source whose replica MD5 differs must fail the replication");
assert_eq!(err.to_string(), REPLICA_ETAG_MISMATCH_ERROR);
let matching = RemotePutObjectResponse {
version_id: None,
etag: object_info.etag.clone(),
};
verify_single_part_replica(&object_info, &matching, false).expect("matching plaintext MD5s pass the audit");
}
#[test]
fn multipart_read_plan_preserves_legacy_plain_part_ranges() {
const MIB: usize = 1024 * 1024;
@@ -157,6 +157,19 @@ pub(crate) fn replication_object_is_ssec_encrypted(user_defined: &HashMap<String
rustfs_replication::is_ssec_encrypted(user_defined)
}
/// Whether the stored ETag of a replication source can be compared with the
/// ETag a target returns for the bytes it received. Only an unencrypted source
/// stores the MD5 of the bytes that go on the wire. A managed-SSE (SSE-S3 /
/// SSE-KMS) or SSE-C source stores a keyed fingerprint (new writes) or a legacy
/// plaintext MD5, and the target re-encrypts what it receives, so neither
/// equality nor inequality of the two ETags proves anything about the replica.
pub(crate) fn replication_source_etag_is_comparable(user_defined: &HashMap<String, String>) -> bool {
matches!(
classify_replication_source_encryption(user_defined),
ReplicationSourceEncryption::Plaintext
)
}
/// HeadObjectOutput adapter over the pure SSE-C passthrough evidence
/// judgment owned by `rustfs-replication`: extract the echoed
/// customer-algorithm header and let the crate-owned policy decide.
@@ -578,6 +591,22 @@ mod tests {
use time::Duration;
use uuid::Uuid;
#[test]
fn replication_source_etag_is_comparable_only_for_plaintext_sources() {
assert!(replication_source_etag_is_comparable(&HashMap::new()));
for (key, value) in [
(metadata_keys::SERVER_SIDE_ENCRYPTION, "AES256"),
(metadata_keys::SERVER_SIDE_ENCRYPTION, "aws:kms"),
(SSEC_ALGORITHM_HEADER, "AES256"),
] {
let metadata = HashMap::from([(key.to_string(), value.to_string())]);
assert!(
!replication_source_etag_is_comparable(&metadata),
"{key}={value} must not expose its stored ETag to the replica audit"
);
}
}
/// backlog#1735 A3b SSE key-case finding. Every RustFS writer persists the
/// SSE intent as lowercase `x-amz-server-side-encryption`
/// (`encryption_material_to_metadata`; header-derived keys come from a
+90
View File
@@ -3691,6 +3691,55 @@ fn activation_pool_meta_lock_error(err: rustfs_lock::LockError) -> Error {
}
}
fn rebalance_worker_read_lock_error(err: rustfs_lock::LockError, object: &str) -> Error {
match err {
rustfs_lock::LockError::QuorumNotReached { required, achieved } => Error::NamespaceLockQuorumUnavailable {
mode: "read",
bucket: RUSTFS_META_BUCKET.to_string(),
object: object.to_string(),
required,
achieved,
},
other => Error::Lock(other),
}
}
/// Read-only admission to an already durable rebalance run. Keeping this
/// separate from the activation writer prevents admission from authorizing a
/// pool or rebalance metadata write under shared namespace locks.
pub(crate) struct PoolRebalanceWorkerFence {
pool_meta_guard: rustfs_lock::NamespaceLockGuard,
rebalance_meta_guard: rustfs_lock::NamespaceLockGuard,
fleet_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
}
impl PoolRebalanceWorkerFence {
pub(crate) fn set_fleet_proof(
&mut self,
fleet_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
) {
self.fleet_proof = fleet_proof;
}
pub(crate) fn ensure_held(&self) -> Result<()> {
if self.pool_meta_guard.is_lock_lost()
|| self.pool_meta_guard.is_released()
|| self.rebalance_meta_guard.is_lock_lost()
|| self.rebalance_meta_guard.is_released()
{
return Err(Error::other("worker admission lock lost before rebalance activation"));
}
if self
.fleet_proof
.as_ref()
.is_some_and(|proof| !crate::services::notification_sys::cross_pool_fence_fleet_proof_matches(proof))
{
return Err(Error::other(POOL_ACTIVATION_FLEET_PROOF_EXPIRED));
}
Ok(())
}
}
pub(crate) struct PoolRebalanceActivationFence {
pool_meta_guard: Arc<rustfs_lock::NamespaceLockGuard>,
rebalance_meta_guard: Arc<rustfs_lock::NamespaceLockGuard>,
@@ -3762,6 +3811,35 @@ tokio::task_local! {
pub(crate) static REBALANCE_ACTIVATION_LOCK_ATTEMPT: Arc<tokio::sync::Notify>;
}
pub(crate) async fn acquire_pool_rebalance_worker_locks<S>(pool: Arc<S>) -> Result<PoolRebalanceWorkerFence>
where
S: crate::storage_api_contracts::namespace::NamespaceLocking<
Error = Error,
NamespaceLock = rustfs_lock::NamespaceLockWrapper,
>,
{
// Match activation and entry admission: rebalance.bin -> pool.bin. Joining
// the same durable run must coexist with entries retaining their run fence.
let rebalance_meta_lock = pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?;
#[cfg(test)]
let _ = REBALANCE_ACTIVATION_LOCK_ATTEMPT.try_with(|attempted| attempted.notify_one());
let rebalance_meta_guard = rebalance_meta_lock
.get_read_lock(get_lock_acquire_timeout())
.await
.map_err(|err| rebalance_worker_read_lock_error(err, REBAL_META_NAME))?;
let pool_meta_lock = pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?;
let pool_meta_guard = pool_meta_lock
.get_read_lock(get_lock_acquire_timeout())
.await
.map_err(|err| rebalance_worker_read_lock_error(err, POOL_META_NAME))?;
Ok(PoolRebalanceWorkerFence {
pool_meta_guard,
rebalance_meta_guard,
fleet_proof: None,
})
}
pub(crate) async fn acquire_pool_rebalance_activation_locks<S>(
pool: Arc<S>,
fleet_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
@@ -12430,6 +12508,18 @@ impl ECStore {
Ok(pool_meta)
}
pub(crate) async fn load_runtime_pool_meta_under_worker_fence(
&self,
write_state: &mut PoolMetaWriteState,
worker_fence: &PoolRebalanceWorkerFence,
operation: &str,
) -> Result<PoolMeta> {
worker_fence.ensure_held()?;
let pool_meta = self.load_runtime_pool_meta_observing(write_state, operation).await?;
worker_fence.ensure_held()?;
Ok(pool_meta)
}
pub(crate) async fn load_runtime_pool_meta_under_activation_fence(
&self,
write_state: &mut PoolMetaWriteState,
-40
View File
@@ -72,38 +72,6 @@ type WalkOptions = StorageWalkOptions<fn(&FileInfo) -> bool>;
const LIST_MULTIPART_SETS_CONCURRENCY: usize = 4;
#[cfg(test)]
type TestSetLockers = HashMap<usize, Vec<Arc<dyn rustfs_lock::LockClient>>>;
#[cfg(test)]
static TEST_SET_LOCKERS: std::sync::LazyLock<std::sync::Mutex<TestSetLockers>> =
std::sync::LazyLock::new(|| std::sync::Mutex::new(HashMap::new()));
/// Keep a fixture's real locker domain scoped to its owning instance.
#[cfg(test)]
pub(crate) struct ScopedSetLockers(Arc<InstanceContext>);
#[cfg(test)]
impl ScopedSetLockers {
pub(crate) fn install(ctx: Arc<InstanceContext>, lockers: Vec<Arc<dyn rustfs_lock::LockClient>>) -> Self {
let key = Arc::as_ptr(&ctx) as usize;
let mut domains = TEST_SET_LOCKERS.lock().expect("test set lockers mutex");
assert!(!domains.contains_key(&key), "the instance already owns a test locker domain");
domains.insert(key, lockers);
Self(ctx)
}
}
#[cfg(test)]
impl Drop for ScopedSetLockers {
fn drop(&mut self) {
TEST_SET_LOCKERS
.lock()
.expect("test set lockers mutex")
.remove(&(Arc::as_ptr(&self.0) as usize));
}
}
fn is_idempotent_delete_prefix_error(err: &Error) -> bool {
is_err_object_not_found(err) || is_err_strict_volume_not_found(err)
}
@@ -305,14 +273,6 @@ impl Sets {
}
}
#[cfg(test)]
let pool_lockers = TEST_SET_LOCKERS
.lock()
.expect("test set lockers mutex")
.get(&(Arc::as_ptr(&instance_ctx) as usize))
.cloned()
.unwrap_or_else(|| pool_lockers.clone());
let set_disks = SetDisks::new_with_instance_ctx(
runtime_sources::local_node_name().await,
Arc::new(RwLock::new(set_drive)),
-9
View File
@@ -2527,15 +2527,6 @@ impl DiskAPI for LocalDiskWrapper {
})
}
async fn bucket_creation_witness(&self, volume: &str) -> Result<Option<super::os::NativeBucketCreationWitness>> {
self.track_disk_health_with_op(
"bucket_creation_witness",
|| async { self.disk.bucket_creation_witness(volume).await },
Duration::ZERO,
)
.await
}
async fn make_volume(&self, volume: &str) -> Result<()> {
// Scoped heal must drain directory creation before releasing its lifecycle owner.
let timeout = if crate::store::bucket_heal_scope(volume).is_some() {
-7
View File
@@ -117,9 +117,6 @@ pub const O_APPEND: FileMode = 0x00400;
// create_new: bool,
pub async fn open_file(path: impl AsRef<Path>, mode: FileMode) -> io::Result<File> {
if mode & O_CREATE != 0 {
super::os::record_native_publication(path.as_ref());
}
let base_opts = match mode & (O_RDONLY | O_WRONLY | O_RDWR) {
O_RDONLY => get_readonly_options(),
O_WRONLY => get_writeonly_options(),
@@ -163,7 +160,6 @@ pub fn lstat_std(path: impl AsRef<Path>) -> io::Result<Metadata> {
}
pub async fn make_dir_all(path: impl AsRef<Path>) -> io::Result<()> {
super::os::record_native_publication(path.as_ref());
fs::create_dir_all(path.as_ref()).await
}
@@ -215,18 +211,15 @@ pub fn remove_all_std(path: impl AsRef<Path>) -> io::Result<()> {
}
pub async fn mkdir(path: impl AsRef<Path>) -> io::Result<()> {
super::os::record_native_publication(path.as_ref());
fs::create_dir(path.as_ref()).await
}
pub async fn rename(from: impl AsRef<Path>, to: impl AsRef<Path>) -> io::Result<()> {
super::os::record_native_publication(to.as_ref());
fs::rename(from, to).await
}
#[cfg(any(not(windows), test))]
pub fn rename_std(from: impl AsRef<Path>, to: impl AsRef<Path>) -> io::Result<()> {
super::os::record_native_publication(to.as_ref());
std::fs::rename(from, to)
}
-220
View File
@@ -286,7 +286,6 @@ fn snapshot_part_transaction_file(src: &Path, backup: &Path, absent: &Path) -> s
}
fn restore_part_transaction_file(current: &Path, backup: &Path, absent: &Path, restore: &Path) -> std::io::Result<()> {
os::record_native_publication(current);
match std::fs::symlink_metadata(backup) {
Ok(metadata) if metadata.is_file() => {
remove_file_if_exists(restore)?;
@@ -323,7 +322,6 @@ async fn write_delete_rollback_file(
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<()> {
let backup_dir = object_dir.join(rollback_dir.to_string());
os::record_native_publication(&backup_dir);
let path = backup_dir.join(name);
let lease = os::acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await;
let data = data.to_vec();
@@ -2332,9 +2330,6 @@ impl AsyncWrite for DirectWriter {
fn open_direct_writer(file_path: &Path, state: &DirectIoWriteState) -> Result<Option<DirectWriter>> {
use std::os::unix::fs::OpenOptionsExt;
#[cfg(test)]
os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::DirectWriteOpen, file_path);
os::record_native_publication(file_path);
let open_result = std::fs::OpenOptions::new()
.create(true)
.write(true)
@@ -4107,7 +4102,6 @@ impl LocalIoBackend for StdBackend {
if is_direct_io_write_enabled() && self.direct_io_write.supported.load(Ordering::Relaxed) {
let write_state = self.direct_io_write.clone();
let direct_path = file_path.clone();
os::record_native_publication(&direct_path);
let direct = tokio::task::spawn_blocking(move || open_direct_writer(&direct_path, &write_state))
.await
.map_err(|err| DiskError::other(format!("O_DIRECT open task failed: {err}")))??;
@@ -5355,7 +5349,6 @@ async fn build_local_io_backend(root: PathBuf) -> Arc<dyn LocalIoBackend> {
pub struct LocalDisk {
pub root: PathBuf,
publication_root: os::PublicationRoot,
publication_history: Option<os::NativePublicationRegistration>,
/// I/O root pinned to the mount instance that was opened while the disk
/// was initialized. On Linux this is `/proc/self/fd/<dirfd>/.`; resolving
/// paths beneath it keeps repair I/O on that mount even if the configured
@@ -5710,10 +5703,6 @@ impl LocalDisk {
#[cfg(not(target_os = "linux"))]
let io_root = Self::open_mount_lease(&root)?;
let publication_history = os::register_native_publication_root(&root, &io_root);
#[cfg(all(test, target_os = "linux"))]
os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::BeforeDataUsageLayout, &root);
ensure_data_usage_layout(&io_root)
.await
.map_err(|e| e.narrow_to_disk().unwrap_or_else(DiskError::other))
@@ -5851,7 +5840,6 @@ impl LocalDisk {
let mut disk = Self {
root: root.clone(),
publication_root,
publication_history,
io_root: io_root.clone(),
#[cfg(target_os = "linux")]
mount_lease,
@@ -8076,7 +8064,6 @@ impl LocalDisk {
};
if let Some(rollback_dir) = rollback_dir {
let rollback_path = object_dir.join(rollback_dir.to_string());
os::record_native_publication(&rollback_path);
if let Err(err) = fs::create_dir_all(&rollback_path).await {
let err: DiskError = to_file_error(err).into();
if reserved_version_delete {
@@ -8606,7 +8593,6 @@ impl LocalDisk {
}
InternalBuf::Owned(buf) => {
let path = file_path.to_path_buf();
os::record_native_publication(&path);
if let Some(parent) = path.parent()
&& parent != skip_parent
{
@@ -8615,7 +8601,6 @@ impl LocalDisk {
let root = self.publication_root.clone();
tokio::task::spawn_blocking(move || {
os::record_native_publication(&path);
#[cfg(test)]
run_owned_file_write_before_open(&path);
@@ -10400,9 +10385,6 @@ impl DiskAPI for LocalDisk {
}
async fn close(&self) -> Result<()> {
if let Some(history) = self.publication_history.as_ref() {
history.invalidate();
}
// This disk instance is being retired (e.g. replaced by renew_disk on
// reconnect). Drop its cached descriptors so a replacement instance's
// invalidations are never defeated by this one continuing to serve stale
@@ -10560,7 +10542,6 @@ impl DiskAPI for LocalDisk {
let file_path = self.io_get_object_path(volume, path)?;
let path = path.to_string();
let sync_metadata = effective_durability(volume).syncs_commit_metadata();
os::record_native_publication(&file_path);
return Ok(tokio::task::spawn_blocking(move || {
// A persistent directory lock bounds metadata growth. Removing
// per-target lock files can split flock ownership across inodes.
@@ -10949,7 +10930,6 @@ impl DiskAPI for LocalDisk {
.map_err(to_file_error)?;
}
}
os::record_native_publication(&transaction_path);
tokio::task::spawn_blocking(move || {
let source = std::fs::symlink_metadata(&src_file_path).map_err(to_file_error)?;
if !source.is_file() {
@@ -11021,7 +11001,6 @@ impl DiskAPI for LocalDisk {
}
let durability = effective_durability(volume);
os::record_native_publication(&current_data_path);
tokio::task::spawn_blocking(move || {
match std::fs::symlink_metadata(&transaction_path) {
Ok(metadata) if metadata.is_dir() => {}
@@ -11632,27 +11611,12 @@ impl DiskAPI for LocalDisk {
}
#[tracing::instrument(level = "trace", skip_all)]
async fn bucket_creation_witness(&self, volume: &str) -> Result<Option<os::NativeBucketCreationWitness>> {
let Some(history) = self.publication_history.as_ref() else {
return Ok(None);
};
let Some(disk_id) = self.get_disk_id().await? else {
return Ok(None);
};
Ok(history.witness(volume, disk_id, Arc::clone(&self.io_backend)))
}
async fn make_volume(&self, volume: &str) -> Result<()> {
if !Self::is_valid_volname(volume) {
return Err(Error::other("Invalid arguments specified"));
}
let volume_dir = self.io_get_bucket_path(volume)?;
#[cfg(all(test, not(windows)))]
os::prepared_publication_test_hooks::run(
os::prepared_publication_test_hooks::Stage::CreateDirectory,
&self.root.join(volume),
);
// Volume creation is a mutation boundary, so it must observe the live
// filesystem rather than a potentially stale existence-cache entry.
@@ -12385,190 +12349,6 @@ mod test {
use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
use tracing_subscriber::fmt::MakeWriter;
#[cfg(unix)]
#[tokio::test]
async fn owned_direct_child_open_attempt_consumes_native_history() {
let dir = tempfile::tempdir().expect("native disk directory");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("disk endpoint");
let disk = LocalDisk::new(&endpoint, false).await.expect("native local disk");
let bucket = format!("owned-open-attempt-{}", Uuid::new_v4());
let witness = disk
.publication_history
.as_ref()
.expect("native registration")
.witness(&bucket, Uuid::new_v4(), Arc::clone(&disk.io_backend))
.expect("fresh history");
disk.write_all_public(&bucket, "direct-child", Bytes::from_static(b"payload"))
.await
.expect_err("the direct child has no bucket parent");
assert!(
!os::validate_native_creation_witnesses(&[witness]),
"even a failed O_CREATE in the real owned executor must permanently consume native history"
);
}
#[cfg(target_os = "linux")]
#[tokio::test]
async fn synchronous_direct_open_attempt_consumes_native_history() {
let dir = tempfile::tempdir().expect("native disk directory");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("disk endpoint");
let disk = LocalDisk::new(&endpoint, false).await.expect("native local disk");
let bucket = format!("direct-io-open-attempt-{}", Uuid::new_v4());
let witness = disk
.publication_history
.as_ref()
.expect("native registration")
.witness(&bucket, Uuid::new_v4(), Arc::clone(&disk.io_backend))
.expect("fresh history");
let path = disk.io_get_object_path(&bucket, "direct-child").expect("direct I/O path");
assert!(
open_direct_writer(&path, &DirectIoWriteState::new()).is_err(),
"the actual synchronous O_DIRECT entry must observe the missing bucket parent"
);
assert!(
!os::validate_native_creation_witnesses(&[witness]),
"a directly callable O_CREATE helper cannot retain retry permission after a failed publication attempt"
);
}
#[cfg(target_os = "linux")]
#[tokio::test]
async fn linux_queued_direct_open_cancellation_retains_started_history() {
use crate::disk::os::prepared_publication_test_hooks as hooks;
use std::sync::mpsc;
let dir = tempfile::tempdir().expect("native disk directory");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("disk endpoint");
let disk = LocalDisk::new(&endpoint, false).await.expect("native local disk");
let backend = Arc::new(StdBackend::new_without_fd_cache(disk.io_root().to_path_buf()));
let actual_backend: Arc<dyn LocalIoBackend> = backend.clone();
let history = disk.publication_history.as_ref().expect("native registration");
let bucket = format!("native-queued-dio-{}", Uuid::new_v4());
let disk_id = Uuid::new_v4();
let witness = history
.witness(&bucket, disk_id, Arc::clone(&actual_backend))
.expect("fresh native history");
let path = local_disk_object_path(disk.io_root(), "", &bucket).expect("first-level DIO path");
let write_state = Arc::clone(&backend.direct_io_write);
let idle_state_owners = Arc::strong_count(&write_state);
let (entered_tx, entered_rx) = mpsc::channel();
let (release_tx, release_rx) = mpsc::channel();
let _hook = hooks::install_at(hooks::Stage::DirectWriteOpen, &path, move || {
entered_tx.send(()).expect("signal real DIO executor entry");
let _ = release_rx.recv();
});
temp_env::async_with_vars([(ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE, Some("true"))], async {
let operation_backend = Arc::clone(&backend);
let operation_bucket = bucket.clone();
let operation = tokio::spawn(async move {
operation_backend
.open_write("", &operation_bucket, WriteMode::Truncate { size_hint: 0 })
.await
});
tokio::task::spawn_blocking(move || entered_rx.recv_timeout(Duration::from_secs(30)))
.await
.expect("DIO entry waiter")
.expect("the real blocking executor must enter");
assert!(
!os::validate_native_creation_witnesses(std::slice::from_ref(&witness)),
"Started must be recorded before the queued DIO worker performs its own history lookup"
);
assert!(!path.exists(), "the keyed pause must precede the actual O_DIRECT open");
operation.abort();
match operation.await {
Err(error) if error.is_cancelled() => {}
Err(error) => panic!("DIO waiter failed instead of cancelling: {error}"),
Ok(_) => panic!("the paused DIO waiter must be cancelled"),
}
release_tx.send(()).expect("release the real DIO open");
timeout(Duration::from_secs(30), async {
while Arc::strong_count(&write_state) > idle_state_owners {
tokio::task::yield_now().await;
}
})
.await
.expect("the detached DIO executor must drain");
})
.await;
assert!(path.is_file(), "cancelling the waiter must not discard the queued native open");
std::fs::remove_file(&path).expect("remove the file published by the cancelled DIO waiter");
assert!(
history.witness(&bucket, disk_id, actual_backend).is_none(),
"executor drain and physical removal must not regrant native history"
);
}
#[cfg(target_os = "linux")]
#[tokio::test]
async fn linux_mount_lease_alias_publication_consumes_same_history() {
let dir = tempfile::tempdir().expect("native disk directory");
let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("disk endpoint");
let disk = LocalDisk::new(&endpoint, false).await.expect("native local disk");
let history = disk.publication_history.as_ref().expect("native registration");
let disk_id = Uuid::new_v4();
for (alias, root) in [("root", disk.root.as_path()), ("io", disk.io_root())] {
let bucket = format!("native-alias-{alias}-{}", Uuid::new_v4());
let witness = history
.witness(&bucket, disk_id, Arc::clone(&disk.io_backend))
.expect("fresh native history through both real root aliases");
super::super::fs::mkdir(root.join(&bucket))
.await
.expect("publish through the actual directory entry");
assert!(
!os::validate_native_creation_witnesses(std::slice::from_ref(&witness)),
"{alias}: publication through either physical-root alias must consume the same history"
);
std::fs::remove_dir(disk.root.join(&bucket)).expect("remove the physically published directory");
assert!(
history.witness(&bucket, disk_id, Arc::clone(&disk.io_backend)).is_none(),
"{alias}: removal through the other alias must not regrant native history"
);
}
}
#[cfg(target_os = "linux")]
#[tokio::test]
async fn linux_native_history_is_registered_before_data_usage_layout() {
use crate::disk::os::prepared_publication_test_hooks as hooks;
let dir = tempfile::tempdir().expect("native disk directory");
let root = std::fs::canonicalize(dir.path()).expect("canonical disk root");
let bucket = format!("native-before-layout-{}", Uuid::new_v4());
let source = root.join(format!("layout-source-{}", Uuid::new_v4()));
std::fs::create_dir(&source).expect("prepare an actual directory for publication");
let destination = root.join(&bucket);
let publication_path = destination.clone();
let _hook = hooks::install_at(hooks::Stage::BeforeDataUsageLayout, &root, move || {
super::super::fs::rename_std(&source, &publication_path)
.expect("publish through the native entry before data usage layout");
});
let endpoint = Endpoint::try_from(root.to_str().expect("UTF-8 disk path")).expect("disk endpoint");
let disk = LocalDisk::new(&endpoint, false)
.await
.expect("native local disk with real startup layout");
let history = disk.publication_history.as_ref().expect("native registration");
let disk_id = Uuid::new_v4();
let untouched = format!("native-layout-untouched-{}", Uuid::new_v4());
assert!(
history.witness(&untouched, disk_id, Arc::clone(&disk.io_backend)).is_some(),
"startup layout must keep an untouched bucket's native history known"
);
assert!(
history.witness(&bucket, disk_id, Arc::clone(&disk.io_backend)).is_none(),
"registration must capture real publication before startup layout finishes"
);
std::fs::remove_dir(destination).expect("remove the physically published directory");
assert!(
history.witness(&bucket, disk_id, Arc::clone(&disk.io_backend)).is_none(),
"startup completion and physical removal must retain Started"
);
}
#[cfg(unix)]
#[tokio::test]
async fn read_dir_entries_skip_symlinks_without_truncating_listing() {
-12
View File
@@ -491,13 +491,6 @@ impl DiskAPI for Disk {
}
}
async fn bucket_creation_witness(&self, volume: &str) -> Result<Option<os::NativeBucketCreationWitness>> {
match self {
Disk::Local(local_disk) => local_disk.bucket_creation_witness(volume).await,
Disk::Remote(_) => Ok(None),
}
}
#[tracing::instrument(level = "trace", skip_all)]
async fn make_volume(&self, volume: &str) -> Result<()> {
match self {
@@ -1292,11 +1285,6 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
// NSScanner
// Volume operations.
/// Unknown or remote backends cannot attest process-local publication history.
async fn bucket_creation_witness(&self, _volume: &str) -> Result<Option<os::NativeBucketCreationWitness>> {
Ok(None)
}
async fn make_volume(&self, volume: &str) -> Result<()>;
async fn make_volumes(&self, volume: Vec<&str>) -> Result<()>;
async fn list_volumes(&self) -> Result<Vec<VolumeInfo>>;
+10 -595
View File
@@ -21,7 +21,7 @@ use futures::TryStreamExt;
use parking_lot::Mutex;
use rustfs_utils::path::SLASH_SEPARATOR;
use std::{
collections::{HashMap, HashSet, VecDeque},
collections::{HashMap, VecDeque},
io,
path::{Component, Path, PathBuf},
sync::{Arc, LazyLock, Weak},
@@ -276,14 +276,6 @@ pub(crate) mod prepared_publication_test_hooks {
Rollback,
#[cfg(test)]
DirFsync,
#[cfg(test)]
CreateDirectory,
#[cfg(test)]
NativeRootIdentity,
#[cfg(all(test, target_os = "linux"))]
DirectWriteOpen,
#[cfg(all(test, target_os = "linux"))]
BeforeDataUsageLayout,
}
type Hook = Box<dyn FnOnce() + Send>;
@@ -1500,226 +1492,6 @@ fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'stati
}
static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
#[cfg(windows)]
type NativeRootIdentity = (u64, [u8; 16]);
#[cfg(not(windows))]
type NativeRootIdentity = (u64, u64);
// Publication history outlives disk handles, cleanup, and metadata snapshots.
// Reusing a pathname or descriptor for another physical root is not new proof.
#[derive(Debug, Default)]
struct NativePublicationHistory {
roots: HashMap<NativeRootIdentity, NativeRootHistory>,
aliases: HashMap<PathBuf, HashSet<NativeRootIdentity>>,
}
#[derive(Debug, Default)]
struct NativeRootHistory {
unknown: bool,
started: HashSet<String>,
}
static NATIVE_PUBLICATION_HISTORY: LazyLock<Mutex<NativePublicationHistory>> =
LazyLock::new(|| Mutex::new(NativePublicationHistory::default()));
#[derive(Clone, Debug)]
pub(crate) struct NativePublicationRegistration {
identity: NativeRootIdentity,
root: PathBuf,
io_root: PathBuf,
}
/// Opaque, process-local observation. Unsupported disks cannot construct it.
#[derive(Clone, Debug)]
pub struct NativeBucketCreationWitness {
registration: NativePublicationRegistration,
backend: Arc<dyn super::local::LocalIoBackend>,
bucket: String,
disk_id: uuid::Uuid,
}
fn native_root_identity(path: &Path) -> Option<NativeRootIdentity> {
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
#[cfg(test)]
prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::NativeRootIdentity, path);
let metadata = std::fs::metadata(path).ok()?;
metadata.is_dir().then(|| (metadata.dev(), metadata.ino()))
}
#[cfg(windows)]
{
// Resolve and validate the same directory anchor used for publication;
// keep its volume and complete 128-bit file ID in the history key.
let (_, directory) = open_windows_publication_root(path).ok()?;
windows_file_identity(directory.handle.as_ref()).ok()
}
#[cfg(all(not(unix), not(windows)))]
{
let _ = path;
None
}
}
pub(crate) fn register_native_publication_root(root: &Path, io_root: &Path) -> Option<NativePublicationRegistration> {
let identity = native_root_identity(root)?;
if native_root_identity(io_root) != Some(identity) {
return None;
}
let mut history = NATIVE_PUBLICATION_HISTORY.lock();
history.roots.entry(identity).or_default();
for alias in [root, io_root] {
let alias: PathBuf = alias.components().collect();
let previous = history.aliases.entry(alias).or_default();
let collisions: Vec<_> = previous.iter().copied().filter(|old| *old != identity).collect();
previous.insert(identity);
if !collisions.is_empty() {
history.roots.get_mut(&identity)?.unknown = true;
for old in collisions {
history.roots.get_mut(&old)?.unknown = true;
}
}
}
Some(NativePublicationRegistration {
identity,
root: root.to_path_buf(),
io_root: io_root.to_path_buf(),
})
}
/// Record before queueing filesystem work. Cancellation never removes Started.
pub(crate) fn record_native_publication(path: &Path) {
let aliases: Vec<_> = {
let history = NATIVE_PUBLICATION_HISTORY.lock();
path.ancestors()
.filter(|alias| history.aliases.contains_key(*alias))
.map(Path::to_path_buf)
.collect()
};
// Physical lookups must never hold the process-wide publication mutex.
let observations: Vec<_> = aliases
.into_iter()
.filter_map(|alias| {
let relative = path.strip_prefix(&alias).ok()?;
let mut components = relative.components().filter(|part| !matches!(part, Component::CurDir));
let first = components.next()?;
let bucket = match first {
Component::Normal(bucket) if components.all(|part| matches!(part, Component::Normal(_))) => {
bucket.to_str().map(str::to_owned)
}
_ => None,
};
let physical = native_root_identity(&alias);
Some((alias, physical, bucket))
})
.collect();
let mut history = NATIVE_PUBLICATION_HISTORY.lock();
for (alias, physical, bucket) in observations {
// Aliases only accumulate identities. Include registration concurrent
// with the stat, rather than using a stale captured identity list.
let identities = history.aliases.get(&alias).cloned().unwrap_or_default();
if physical.is_none() || identities.iter().any(|identity| Some(*identity) != physical) {
for identity in identities {
history.roots.entry(identity).or_default().unknown = true;
}
if let Some(identity) = physical {
history.roots.entry(identity).or_default().unknown = true;
}
} else if let Some(identity) = physical {
let root = history.roots.entry(identity).or_default();
match bucket {
Some(bucket) => {
root.started.insert(bucket);
}
None => root.unknown = true,
}
}
}
}
impl NativePublicationRegistration {
pub(crate) fn invalidate(&self) {
if let Some(root) = NATIVE_PUBLICATION_HISTORY.lock().roots.get_mut(&self.identity) {
root.unknown = true;
}
}
pub(crate) fn witness(
&self,
bucket: &str,
disk_id: uuid::Uuid,
backend: Arc<dyn super::local::LocalIoBackend>,
) -> Option<NativeBucketCreationWitness> {
if disk_id.is_nil()
|| !matches!(Path::new(bucket).components().next(), Some(Component::Normal(_)))
|| Path::new(bucket).components().count() != 1
{
return None;
}
let witness = NativeBucketCreationWitness {
registration: self.clone(),
backend,
bucket: bucket.to_owned(),
disk_id,
};
validate_native_creation_witnesses(std::slice::from_ref(&witness)).then_some(witness)
}
}
impl NativeBucketCreationWitness {
pub(crate) fn matches(&self, other: &Self) -> bool {
self.registration.identity == other.registration.identity
&& Arc::ptr_eq(&self.backend, &other.backend)
&& self.disk_id == other.disk_id
&& self.bucket == other.bucket
}
fn physical_root_is_current(&self) -> bool {
let root = native_root_identity(&self.registration.root);
let io_root = native_root_identity(&self.registration.io_root);
if root == Some(self.registration.identity) && io_root == Some(self.registration.identity) {
return true;
}
let mut history = NATIVE_PUBLICATION_HISTORY.lock();
history.roots.entry(self.registration.identity).or_default().unknown = true;
for identity in [root, io_root].into_iter().flatten() {
history.roots.entry(identity).or_default().unknown = true;
}
false
}
fn valid_in(&self, history: &NativePublicationHistory) -> bool {
history
.roots
.get(&self.registration.identity)
.is_some_and(|root| !root.unknown && !root.started.contains(&self.bucket))
}
}
pub(crate) fn validate_native_creation_witnesses(witnesses: &[NativeBucketCreationWitness]) -> bool {
if witnesses.is_empty() || !witnesses.iter().all(NativeBucketCreationWitness::physical_root_is_current) {
return false;
}
let history = NATIVE_PUBLICATION_HISTORY.lock();
witnesses.iter().all(|witness| witness.valid_in(&history))
}
pub(crate) fn consume_native_creation_witnesses(witnesses: &[NativeBucketCreationWitness]) -> bool {
if witnesses.is_empty() || !witnesses.iter().all(NativeBucketCreationWitness::physical_root_is_current) {
return false;
}
let mut history = NATIVE_PUBLICATION_HISTORY.lock();
if !witnesses.iter().all(|witness| witness.valid_in(&history)) {
return false;
}
for witness in witnesses {
if let Some(root) = history.roots.get_mut(&witness.registration.identity) {
root.started.insert(witness.bucket.clone());
}
}
true
}
type NamespaceMutationLock = AsyncMutex<()>;
type NamespaceMutationLockRegistry = HashMap<PathBuf, Weak<NamespaceMutationLock>>;
static DISK_NAMESPACE_MUTATION_LOCKS: LazyLock<Mutex<NamespaceMutationLockRegistry>> =
@@ -2422,7 +2194,6 @@ pub(in crate::disk) async fn rename_with_namespace_owner(
dst: &Path,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> io::Result<()> {
record_native_publication(dst);
if namespace_owner.is_none() {
return tokio::fs::rename(src, dst).await;
}
@@ -2444,19 +2215,12 @@ pub(in crate::disk) async fn create_dir_all_with_namespace_owner(
path: &Path,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> io::Result<()> {
record_native_publication(path);
if namespace_owner.is_none() {
return tokio::fs::create_dir_all(path).await;
}
let path = path.to_path_buf();
let lease = acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await;
run_blocking_namespace_operation(lease, move || {
#[cfg(all(test, not(windows)))]
prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::CreateDirectory, &path);
record_native_publication(&path);
std::fs::create_dir_all(path)
})
.await
run_blocking_namespace_operation(lease, move || std::fs::create_dir_all(path)).await
}
#[tracing::instrument(name = "rename_all", level = "debug", skip_all)]
@@ -2467,7 +2231,6 @@ pub(crate) async fn rename_all_with_owner(
publication_root: &PublicationRoot,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<()> {
record_native_publication(dst_file_path.as_ref());
let lease = acquire_namespace_mutation_lease_with_owner(dst_file_path.as_ref(), namespace_owner).await;
rename_all_with_lease(src_file_path, dst_file_path, base_dir, publication_root, lease).await
}
@@ -2479,7 +2242,6 @@ pub(crate) async fn rename_all_with_lease(
publication_root: &PublicationRoot,
lease: Arc<NamespaceMutationLease>,
) -> Result<()> {
record_native_publication(dst_file_path.as_ref());
reliable_rename_inner_with_lease(
src_file_path.as_ref().to_path_buf(),
dst_file_path.as_ref().to_path_buf(),
@@ -2504,7 +2266,6 @@ pub(crate) async fn rename_reconciled_metadata(
owner: Option<Arc<dyn Send + Sync>>,
authority: Arc<crate::bucket::lifecycle::legacy_transition_state_reconcile::TransitionStateReconcileAuthority>,
) -> Result<()> {
record_native_publication(&destination);
let lease = acquire_namespace_mutation_lease_with_owner(&destination, owner).await;
run_blocking_namespace_operation(lease, move || {
let preparation = prepare_rename_with_retry(&source, &destination, &base_dir, &publication_root)?;
@@ -2665,7 +2426,7 @@ pub(crate) fn read_destination_file_with_commit_guard(
let Some(parent) = directories.destination.get() else {
return Ok(None);
};
return match open_linux_relative(&parent.file, Path::new(name), OFlags::RDONLY, Mode::empty()) {
match open_linux_relative(&parent.file, Path::new(name), OFlags::RDONLY, Mode::empty()) {
Ok(mut file) => {
let mut data = Vec::new();
std::io::Read::read_to_end(&mut file, &mut data)?;
@@ -2673,7 +2434,7 @@ pub(crate) fn read_destination_file_with_commit_guard(
}
Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(None),
Err(err) => Err(err),
};
}
}
#[cfg(windows)]
{
@@ -2743,9 +2504,9 @@ pub(crate) async fn rename_all_with_prepared_source(
{
#[cfg(any(test, feature = "test-util"))]
prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::PreparedRename, &dst_file_path);
return commit_guard
commit_guard
.directories
.rename(&src_file_path, &dst_file_path, Some(&prepared_source));
.rename(&src_file_path, &dst_file_path, Some(&prepared_source))
}
#[cfg(not(target_os = "linux"))]
{
@@ -2781,7 +2542,7 @@ pub(crate) async fn rename_all_with_commit_guard(
let destination = dst_file_path.as_ref().to_path_buf();
let guard = commit_guard.clone();
let _ = (base_dir, publication_root);
return run_blocking_namespace_operation(lease, move || {
run_blocking_namespace_operation(lease, move || {
guard.directories.prepare_destination()?;
#[cfg(test)]
prepared_publication_test_hooks::run_rename_destination(&source, &destination);
@@ -2794,7 +2555,7 @@ pub(crate) async fn rename_all_with_commit_guard(
})
.await
.map_err(to_file_error)
.map_err(DiskError::from);
.map_err(DiskError::from)
}
#[cfg(not(target_os = "linux"))]
{
@@ -2826,7 +2587,6 @@ pub(crate) async fn rename_all_ignore_missing_source_with_owner(
publication_root: &PublicationRoot,
namespace_owner: Option<Arc<dyn Send + Sync>>,
) -> Result<()> {
record_native_publication(dst_file_path.as_ref());
let src_file_path = src_file_path.as_ref();
let lease = acquire_namespace_mutation_lease_with_owner(dst_file_path.as_ref(), namespace_owner).await;
match reliable_rename_inner_with_lease(
@@ -2881,7 +2641,6 @@ async fn reliable_rename_inner(
publication_root: &PublicationRoot,
warn_on_missing_source: bool,
) -> io::Result<()> {
record_native_publication(dst_file_path.as_ref());
let src_file_path = src_file_path.as_ref().to_path_buf();
let dst_file_path = dst_file_path.as_ref().to_path_buf();
let base_dir = base_dir.as_ref().to_path_buf();
@@ -2905,7 +2664,6 @@ async fn reliable_rename_inner_with_lease(
warn_on_missing_source: bool,
lease: Arc<NamespaceMutationLease>,
) -> io::Result<()> {
record_native_publication(&dst_file_path);
let operation = {
let src_file_path = src_file_path.clone();
let dst_file_path = dst_file_path.clone();
@@ -3328,7 +3086,6 @@ fn rename_into_existing_parent(
parent_guard: Option<&ExistingBaseDirectoryGuard>,
#[cfg(target_os = "linux")] source_parent: &std::fs::File,
) -> io::Result<()> {
record_native_publication(dst_file_path);
use rustix::fs::renameat;
let Some(parent_guard) = parent_guard else {
@@ -3382,7 +3139,6 @@ fn rename_into_existing_parent(
parent_guard: Option<&ExistingBaseDirectoryGuard>,
source: &winapi_util::Handle,
) -> io::Result<()> {
record_native_publication(dst_file_path);
use std::{
mem::size_of,
os::windows::{ffi::OsStrExt, io::AsRawHandle},
@@ -4134,7 +3890,7 @@ pub(crate) fn prepare_rename_commit_guard(
Err(err) if err.kind() == io::ErrorKind::NotFound => {}
Err(err) => return Err(err),
}
return Ok(RenameCommitGuard {
Ok(RenameCommitGuard {
directories: Arc::new(LinuxCommitDirectories {
root: publication_root.clone(),
source_path: source_parent.to_path_buf(),
@@ -4143,7 +3899,7 @@ pub(crate) fn prepare_rename_commit_guard(
destination,
destination_base: destination_base.to_path_buf(),
}),
});
})
}
#[cfg(windows)]
{
@@ -4990,7 +4746,6 @@ pub(crate) fn mkdir_all_below_existing_base_std(
base_dir: &Path,
publication_root: &PublicationRoot,
) -> io::Result<ExistingBaseDirectoryGuard> {
record_native_publication(dir_path);
let relative = existing_base_directory_suffix(dir_path, base_dir)?;
#[cfg(unix)]
@@ -6557,77 +6312,6 @@ mod tests {
assert_eq!(err.raw_os_error(), Some(disk_full));
}
#[cfg(windows)]
#[test]
fn windows_native_root_identity_matches_directory_handles() {
let temp = tempdir().expect("create native identity fixture");
let first = temp.path().join("first");
let second = temp.path().join("second");
std::fs::create_dir(&first).expect("create first root");
std::fs::create_dir(&second).expect("create second root");
let (_, first_anchor) = open_windows_publication_root(&first).expect("open first directory anchor");
let (_, second_anchor) = open_windows_publication_root(&second).expect("open second directory anchor");
let first_identity = windows_file_identity(first_anchor.handle.as_ref()).expect("query full first identity");
let second_identity = windows_file_identity(second_anchor.handle.as_ref()).expect("query full second identity");
assert_eq!(native_root_identity(&first), Some(first_identity));
assert_eq!(native_root_identity(&first.join(".")), Some(first_identity));
assert_eq!(native_root_identity(&second), Some(second_identity));
assert_ne!(first_identity, second_identity, "distinct physical roots must have distinct identities");
}
#[cfg(windows)]
#[tokio::test]
async fn windows_native_creation_witness_retains_started_history() {
let temp = tempdir().expect("create native witness fixture");
let root = std::fs::canonicalize(temp.path()).expect("canonical native root");
let _publication_root = PublicationRoot::new(&root).expect("retain native directory anchor");
let registration = register_native_publication_root(&root, &root.join(".")).expect("register same physical root");
let backend: Arc<dyn super::super::local::LocalIoBackend> = Arc::new(super::super::local::StdBackend::new(root.clone()));
let disk_id = uuid::Uuid::new_v4();
let witness = registration
.witness("published", disk_id, Arc::clone(&backend))
.expect("fresh native witness");
let bucket = root.join("published");
super::super::fs::mkdir(&bucket)
.await
.expect("publish through the actual native entry");
assert!(bucket.is_dir(), "the production entry must create the directory");
assert!(!validate_native_creation_witnesses(std::slice::from_ref(&witness)));
std::fs::remove_dir(&bucket).expect("remove published directory");
assert!(registration.witness("published", disk_id, Arc::clone(&backend)).is_none());
let reconnected = register_native_publication_root(&root, &root).expect("reconnect same physical root");
assert!(reconnected.witness("published", disk_id, Arc::clone(&backend)).is_none());
let unstarted = reconnected
.witness("unstarted", disk_id, Arc::clone(&backend))
.expect("untouched bucket keeps its witness");
assert!(consume_native_creation_witnesses(std::slice::from_ref(&unstarted)));
assert!(!validate_native_creation_witnesses(std::slice::from_ref(&unstarted)));
assert!(reconnected.witness("unstarted", disk_id, backend).is_none());
}
#[cfg(windows)]
#[test]
fn windows_native_creation_witness_rejects_unknown_roots() {
let temp = tempdir().expect("create unknown identity fixture");
let root = temp.path().join("root");
let other = temp.path().join("other");
let file = temp.path().join("file");
std::fs::create_dir(&root).expect("create native root");
std::fs::create_dir(&other).expect("create distinct native root");
std::fs::write(&file, b"not a directory").expect("create regular file");
assert!(native_root_identity(&temp.path().join("missing")).is_none());
assert!(native_root_identity(&file).is_none());
assert!(register_native_publication_root(&root, &other).is_none());
let registration = register_native_publication_root(&root, &root).expect("register actual native root");
let backend: Arc<dyn super::super::local::LocalIoBackend> = Arc::new(super::super::local::StdBackend::new(root));
registration.invalidate();
assert!(registration.witness("unstarted", uuid::Uuid::new_v4(), backend).is_none());
}
#[cfg(windows)]
#[test]
fn windows_rename_all_legacy_file_identity_distinguishes_sources() {
@@ -8816,272 +8500,3 @@ mod fd_relative_tests {
}
}
}
#[cfg(all(test, unix))]
mod native_creation_history_tests {
use super::*;
struct NativeFixture {
_temp: tempfile::TempDir,
root: PathBuf,
registration: NativePublicationRegistration,
backend: Arc<dyn super::super::local::LocalIoBackend>,
disk_id: uuid::Uuid,
}
impl NativeFixture {
fn new() -> Self {
let temp = tempfile::tempdir().expect("native root");
let root = std::fs::canonicalize(temp.path()).expect("canonical root");
let registration = register_native_publication_root(&root, &root).expect("native registration");
let backend = Arc::new(super::super::local::StdBackend::new(root.clone()));
Self {
_temp: temp,
root,
registration,
backend,
disk_id: uuid::Uuid::new_v4(),
}
}
fn witness(&self, bucket: &str) -> Option<NativeBucketCreationWitness> {
self.registration.witness(bucket, self.disk_id, Arc::clone(&self.backend))
}
}
#[tokio::test]
async fn every_native_directory_publication_entry_invalidates_retry() {
let fixture = NativeFixture::new();
for operation in [
"recursive",
"single",
"open",
"rename",
"rename_std",
"fallback",
"owned_mkdir",
"owned_rename",
"handle_relative",
] {
let bucket = format!("native-{operation}-{}", uuid::Uuid::new_v4());
let path = fixture.root.join(&bucket);
let witness = fixture.witness(&bucket).expect("fresh native witness");
match operation {
"recursive" => super::super::fs::make_dir_all(path.join("object/child"))
.await
.expect("recursive mkdir"),
"single" => super::super::fs::mkdir(&path).await.expect("single mkdir"),
"open" => {
let file = super::super::fs::open_file(&path, super::super::fs::O_CREATE | super::super::fs::O_WRONLY)
.await
.expect("create file");
drop(file);
}
"rename" | "rename_std" | "owned_rename" => {
let source = fixture.root.join(format!("source-{}", uuid::Uuid::new_v4()));
std::fs::create_dir(&source).expect("rename source");
match operation {
"rename" => super::super::fs::rename(&source, &path).await.expect("async rename"),
"rename_std" => super::super::fs::rename_std(&source, &path).expect("sync rename"),
_ => rename_with_namespace_owner(&source, &path, Some(Arc::new(())))
.await
.expect("owned rename"),
}
}
"fallback" => make_dir_all(path.join("object/child"), &path)
.await
.expect("implicit volume fallback"),
"owned_mkdir" => create_dir_all_with_namespace_owner(&path.join("child"), Some(Arc::new(())))
.await
.expect("owned recursive mkdir"),
"handle_relative" => {
let publication_root = PublicationRoot::new(&fixture.root).expect("publication root");
let guard = mkdir_all_below_existing_base_std(&path.join("child"), &fixture.root, &publication_root)
.expect("first-level handle-relative mkdir");
drop(guard);
}
_ => unreachable!(),
}
assert!(
!validate_native_creation_witnesses(std::slice::from_ref(&witness)),
"{operation}: old proof remains live"
);
if path.is_dir() {
std::fs::remove_dir_all(&path).expect("remove published directory");
} else {
std::fs::remove_file(&path).expect("remove published file");
}
assert!(fixture.witness(&bucket).is_none(), "{operation}: removal must not regrant proof");
}
}
#[tokio::test]
async fn cancelled_namespace_waiter_cannot_restore_native_proof() {
let fixture = NativeFixture::new();
let bucket = format!("native-cancel-{}", uuid::Uuid::new_v4());
let path = fixture.root.join(&bucket);
let witness = fixture.witness(&bucket).expect("fresh witness");
let lease = acquire_namespace_mutation_lease(&path).await;
let mut attempt = Box::pin(create_dir_all_with_namespace_owner(&path, Some(Arc::new(()))));
assert!(matches!(futures::poll!(&mut attempt), std::task::Poll::Pending));
assert!(
!validate_native_creation_witnesses(std::slice::from_ref(&witness)),
"Started must precede admission wait"
);
assert!(!path.exists(), "the namespace wait must still precede the syscall");
drop(attempt);
drop(lease);
assert!(fixture.witness(&bucket).is_none(), "cancellation cannot undo an admitted attempt");
}
#[tokio::test]
async fn queued_mkdir_cancellation_retains_lease_and_consumed_history() {
use futures::FutureExt;
use std::sync::mpsc;
let fixture = NativeFixture::new();
let bucket = format!("native-queued-cancel-{}", uuid::Uuid::new_v4());
let path = fixture.root.join(&bucket);
let witness = fixture.witness(&bucket).expect("fresh witness");
let (entered_tx, entered_rx) = mpsc::channel();
let (release_tx, release_rx) = mpsc::channel();
let _hook = prepared_publication_test_hooks::install_at(
prepared_publication_test_hooks::Stage::CreateDirectory,
&path,
move || {
entered_tx.send(()).expect("signal real executor entry");
let _ = release_rx.recv();
},
);
let operation_path = path.clone();
let operation =
tokio::spawn(async move { create_dir_all_with_namespace_owner(&operation_path, Some(Arc::new(()))).await });
tokio::task::spawn_blocking(move || entered_rx.recv_timeout(std::time::Duration::from_secs(30)))
.await
.expect("entry waiter")
.expect("the real blocking executor must enter");
operation.abort();
assert!(operation.await.expect_err("cancel the async waiter").is_cancelled());
assert!(
!validate_native_creation_witnesses(std::slice::from_ref(&witness)),
"queued publication is permanently Started"
);
assert!(
acquire_namespace_mutation_lease(&path).now_or_never().is_none(),
"the actual blocked executor must still own the namespace lease"
);
release_tx.send(()).expect("release the syscall");
let drained = tokio::time::timeout(std::time::Duration::from_secs(30), acquire_namespace_mutation_lease(&path))
.await
.expect("the lease must drain with the syscall");
drop(drained);
assert!(path.is_dir(), "the cancelled waiter cannot discard the already queued syscall");
std::fs::remove_dir_all(&path).expect("remove the published directory");
assert!(fixture.witness(&bucket).is_none(), "drain and removal cannot restore proof");
}
#[tokio::test]
async fn publication_detects_retargeted_physical_alias_without_registration() {
let source = NativeFixture::new();
let destination = NativeFixture::new();
let bucket = format!("native-retarget-{}", uuid::Uuid::new_v4());
let witness = destination.witness(&bucket).expect("destination witness");
let retired = source.root.with_extension("retired");
std::fs::rename(&source.root, &retired).expect("move the original physical root");
std::os::unix::fs::symlink(&destination.root, &source.root).expect("retarget the registered alias");
super::super::fs::mkdir(source.root.join(&bucket))
.await
.expect("publish through the changed alias");
assert!(
!validate_native_creation_witnesses(std::slice::from_ref(&witness)),
"publication-time alias drift must invalidate the actual destination root"
);
std::fs::remove_dir(destination.root.join(&bucket)).expect("remove the published bucket");
assert!(destination.witness(&bucket).is_none(), "removal must not undo alias uncertainty");
std::fs::remove_file(&source.root).expect("remove the symlink");
std::fs::rename(retired, &source.root).expect("restore the test root");
}
#[test]
fn blocked_physical_identity_lookup_does_not_block_other_publication() {
use std::sync::mpsc;
let stalled = NativeFixture::new();
let healthy = NativeFixture::new();
let witness = stalled.witness("unpublished").expect("stalled root witness");
let (entered_tx, entered_rx) = mpsc::channel();
let (release_tx, release_rx) = mpsc::channel();
let _hook = prepared_publication_test_hooks::install_at(
prepared_publication_test_hooks::Stage::NativeRootIdentity,
&stalled.root,
move || {
entered_tx.send(()).expect("signal physical lookup");
let _ = release_rx.recv();
},
);
let validator = std::thread::spawn(move || validate_native_creation_witnesses(&[witness]));
entered_rx
.recv_timeout(std::time::Duration::from_secs(30))
.expect("enter the actual identity lookup");
let (published_tx, published_rx) = mpsc::channel();
let path = healthy.root.join("published");
let publisher = std::thread::spawn(move || {
record_native_publication(&path);
published_tx.send(()).expect("signal independent publication");
});
let published = published_rx.recv_timeout(std::time::Duration::from_secs(5));
release_tx.send(()).expect("release the identity lookup");
validator.join().expect("identity lookup thread");
publisher.join().expect("publication thread");
assert!(
published.is_ok(),
"a slow physical lookup must not retain the process-wide publication mutex"
);
}
#[test]
fn native_consume_is_atomic_and_history_survives_reconnect() {
let fixture = NativeFixture::new();
let first = format!("native-first-{}", uuid::Uuid::new_v4());
let second = format!("native-second-{}", uuid::Uuid::new_v4());
let first_witness = fixture.witness(&first).expect("first witness");
let second_witness = fixture.witness(&second).expect("second witness");
record_native_publication(&fixture.root.join(&second));
assert!(!consume_native_creation_witnesses(&[first_witness.clone(), second_witness]));
assert!(
validate_native_creation_witnesses(std::slice::from_ref(&first_witness)),
"failed cohort consume must be atomic"
);
assert!(consume_native_creation_witnesses(std::slice::from_ref(&first_witness)));
assert!(!validate_native_creation_witnesses(std::slice::from_ref(&first_witness)));
let reconnected = register_native_publication_root(&fixture.root, &fixture.root).expect("reconnected registration");
let backend: Arc<dyn super::super::local::LocalIoBackend> =
Arc::new(super::super::local::StdBackend::new(fixture.root.clone()));
assert!(
reconnected.witness(&first, fixture.disk_id, backend).is_none(),
"reconnect must retain Started"
);
}
#[test]
fn physical_alias_reuse_is_unknown_and_backend_identity_is_real() {
let temp = tempfile::tempdir().expect("alias parent");
let root = temp.path().join("root");
std::fs::create_dir(&root).expect("first physical root");
let old = register_native_publication_root(&root, &root).expect("first registration");
let backend: Arc<dyn super::super::local::LocalIoBackend> = Arc::new(super::super::local::StdBackend::new(root.clone()));
let id = uuid::Uuid::new_v4();
let witness = old.witness("same-name", id, Arc::clone(&backend)).expect("old witness");
let other_backend: Arc<dyn super::super::local::LocalIoBackend> =
Arc::new(super::super::local::StdBackend::new(root.clone()));
let other = old.witness("same-name", id, other_backend).expect("new backend witness");
assert!(!witness.matches(&other), "a different actual backend cannot reuse a pending observation");
std::fs::rename(&root, temp.path().join("retired-root")).expect("retire physical root");
std::fs::create_dir(&root).expect("replacement root at the same alias");
let replacement = register_native_publication_root(&root, &root).expect("replacement registration");
assert!(replacement.witness("same-name", id, backend).is_none(), "alias reuse must remain Unknown");
assert!(
!validate_native_creation_witnesses(&[witness]),
"old physical identity must not validate at a new root"
);
}
}
+1 -1
View File
@@ -17,8 +17,8 @@
use crate::bucket::error::BucketMetadataError;
use crate::disk::error::DiskError;
use crate::storage_api_contracts::{error::StorageErrorCode, range::HTTPRangeError};
use rustfs_s3_types::S3ErrorCode;
use rustfs_utils::path::decode_dir_object;
use s3s::S3ErrorCode;
pub type Error = StorageError;
pub type Result<T> = core::result::Result<T, Error>;
@@ -29,7 +29,7 @@ use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions};
use crate::storage_api_contracts::range::HTTPRangeSpec;
use rustfs_filemeta::ObjectPartInfo;
use rustfs_rio::HashReader;
use rustfs_s3_types::S3ErrorCode;
use s3s::S3ErrorCode;
//#[derive(Clone)]
pub struct PutObjReader {
@@ -231,45 +231,6 @@ mod tests {
assert_eq!(result.2, 1024 * 1024);
}
#[test]
fn test_new_getobjectreader_rejects_a_range_past_the_object_as_invalid_range() {
let oi = multipart_object_info();
let past_the_end = Some(HTTPRangeSpec {
is_suffix_length: false,
start: oi.size,
end: oi.size + 10,
});
let err = new_getobjectreader(&past_the_end, &oi, &ObjectOptions::default(), &HeaderMap::new())
.err()
.expect("a range starting at the object size must be rejected");
assert_eq!(err.code, S3ErrorCode::InvalidRange);
assert_eq!(err.code.as_str(), "InvalidRange");
assert_eq!(err.key, None);
assert_eq!(err.bucket_name, None);
}
#[test]
fn test_new_getobjectreader_rejects_an_unknown_size_without_a_range_as_invalid_range() {
let oi = ObjectInfo {
bucket: "bucket".to_string(),
name: "object".to_string(),
size: -1,
..Default::default()
};
let err = new_getobjectreader(&None, &oi, &ObjectOptions::default(), &HeaderMap::new())
.err()
.expect("an object of unknown size cannot be read without a range");
assert_eq!(err.code, S3ErrorCode::InvalidRange);
assert_eq!(err.code.as_str(), "InvalidRange");
assert_eq!(err.message, "Invalid range");
assert_eq!(err.key.as_deref(), Some("object"));
assert_eq!(err.bucket_name.as_deref(), Some("bucket"));
}
#[test]
fn test_to_s3s_etag() {
// Test unquoted ETag - should become strong etag
+4
View File
@@ -1006,6 +1006,9 @@ pub struct ObjectOptions {
/// Proves an upper layer holds the bucket lifecycle sentinel. A separate
/// fence avoids recursively acquiring the read lock behind a queued writer.
pub bucket_lifecycle_lock_fence: Option<NamespaceLockFence>,
/// Retains the actual lifecycle guard for an owned multipart commit. A
/// lock-loss fence alone does not keep a cancelled caller's lock alive.
pub bucket_lifecycle_lock_owner: Option<Arc<rustfs_lock::NamespaceLockGuard>>,
pub replication_request: bool,
/// True when the inbound request carried the
/// `{x-rustfs-,x-minio-}source-proxy-request` header family with the
@@ -1165,6 +1168,7 @@ impl std::fmt::Debug for ObjectOptions {
.field("delete_mutation_owner", &self.delete_mutation_owner.is_some())
.field("namespace_lock_fence", &self.namespace_lock_fence.is_some())
.field("bucket_lifecycle_lock_fence", &self.bucket_lifecycle_lock_fence.is_some())
.field("bucket_lifecycle_lock_owner", &self.bucket_lifecycle_lock_owner.is_some())
.field("replication_request", &self.replication_request)
.field("proxy_request", &self.proxy_request)
.field("proxy_header_set", &self.proxy_header_set)
+290 -34
View File
@@ -18,8 +18,8 @@ use super::{
encode_rebalance_stop_propagation_record,
};
use crate::core::pools::{
PoolMeta, PoolRebalanceActivationFence, acquire_pool_activation_fleet_proof, acquire_pool_rebalance_activation_locks,
pool_meta_has_active_decommission,
PoolMeta, PoolRebalanceActivationFence, PoolRebalanceWorkerFence, acquire_pool_activation_fleet_proof,
acquire_pool_rebalance_activation_locks, acquire_pool_rebalance_worker_locks, pool_meta_has_active_decommission,
};
use crate::error::{Error, Result};
use crate::object_api::ObjectOptions;
@@ -77,10 +77,33 @@ fn ensure_rebalance_activation_pool_meta_allowed(meta: &PoolMeta) -> Result<()>
}
pub(super) enum RebalanceWorkerActivationFence {
Ready(Box<PoolRebalanceActivationFence>),
Ready(Box<PoolRebalanceWorkerFence>),
NotStartedTerminal,
}
async fn persisted_rebalance_requires_worker_activation<S>(pool: Arc<S>, expected_id: &str) -> Result<bool>
where
S: EcstoreObjectIO,
{
let mut persisted = RebalanceMeta::new();
persisted
.load_with_opts(
pool,
ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await?;
if persisted.id != expected_id {
return Err(Error::other(format!(
"rebalance metadata changed before worker activation: expected {expected_id}, found {}",
persisted.id
)));
}
Ok(crate::services::rebalance::rebalance_requires_worker_activation(&persisted))
}
pub(super) struct RebalanceRunGuard {
_local_guard: tokio::sync::OwnedRwLockReadGuard<()>,
persisted_guard: rustfs_lock::NamespaceLockGuard,
@@ -577,48 +600,58 @@ impl ECStore {
where
S: EcstoreObjectIO + StorageNamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
{
// Lock order: pool_meta_save_gate -> rebalance.bin -> pool.bin.
// Lock order: pool_meta_save_gate -> rebalance.bin -> pool.bin. These
// shared fences admit a peer without waiting for the run's own entries.
let mut pool_meta_guard = self.pool_meta_save_gate.lock().await;
pool_meta_guard.ensure_write_safe("rebalance worker activation")?;
// Classify the durable rebalance record while holding both namespace
// fences. A terminal record is a no-op and must not depend on the
// notification subsystem having published a fleet proof yet.
let mut activation_fence = acquire_pool_rebalance_activation_locks(pool.clone(), None).await?;
let mut worker_fence = acquire_pool_rebalance_worker_locks(pool.clone()).await?;
let pool_meta = self
.load_runtime_pool_meta_under_activation_fence(&mut pool_meta_guard, &activation_fence, "rebalance worker activation")
.load_runtime_pool_meta_under_worker_fence(&mut pool_meta_guard, &worker_fence, "rebalance worker activation")
.await?;
ensure_rebalance_activation_pool_meta_allowed(&pool_meta)?;
let mut persisted = RebalanceMeta::new();
persisted
.load_with_opts(
pool,
ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await?;
if persisted.id != expected_id {
return Err(Error::other(format!(
"rebalance metadata changed before worker activation: expected {expected_id}, found {}",
persisted.id
)));
}
activation_fence.ensure_held()?;
if !crate::services::rebalance::rebalance_requires_worker_activation(&persisted) {
let needs_activation = persisted_rebalance_requires_worker_activation(pool, expected_id).await?;
worker_fence.ensure_held()?;
if !needs_activation {
return Ok(RebalanceWorkerActivationFence::NotStartedTerminal);
}
// Active worker admission still requires the fail-closed fleet proof.
// Attach it immediately before the final fence validation so expiry or
// topology changes are checked again at every later commit boundary.
let fleet_proof = acquire_pool_activation_fleet_proof(&self.ctx).await?;
worker_fence.set_fleet_proof(fleet_proof);
worker_fence.ensure_held()?;
Ok(RebalanceWorkerActivationFence::Ready(Box::new(worker_fence)))
}
pub(super) async fn fence_rebalance_worker_metadata_update<S>(
&self,
pool: Arc<S>,
expected_id: &str,
) -> Result<Option<Box<PoolRebalanceActivationFence>>>
where
S: EcstoreObjectIO + StorageNamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
{
// The admission read fences must already be dropped. Reacquire in the
// original order and revalidate every durable boundary across that gap.
let mut pool_meta_guard = self.pool_meta_save_gate.lock().await;
pool_meta_guard.ensure_write_safe("rebalance worker metadata update")?;
let mut activation_fence = acquire_pool_rebalance_activation_locks(pool.clone(), None).await?;
let pool_meta = self
.load_runtime_pool_meta_under_activation_fence(
&mut pool_meta_guard,
&activation_fence,
"rebalance worker metadata update",
)
.await?;
ensure_rebalance_activation_pool_meta_allowed(&pool_meta)?;
let needs_activation = persisted_rebalance_requires_worker_activation(pool, expected_id).await?;
activation_fence.ensure_held()?;
if !needs_activation {
return Ok(None);
}
let fleet_proof = acquire_pool_activation_fleet_proof(&self.ctx).await?;
activation_fence.set_fleet_proof(fleet_proof);
activation_fence.ensure_held()?;
Ok(RebalanceWorkerActivationFence::Ready(Box::new(activation_fence)))
Ok(Some(Box::new(activation_fence)))
}
#[tracing::instrument(skip_all)]
@@ -1816,6 +1849,107 @@ mod tests {
assert_activation_locks_released(&store).await;
}
#[tokio::test]
#[serial_test::serial]
async fn rebalance_init_missing_fleet_proof_preserves_metadata_and_rechecks_competing_run() {
let previous_id = "completed-before-capability-refresh";
let terminal = RebalanceMeta {
id: previous_id.to_string(),
stopped_at: Some(OffsetDateTime::now_utc()),
pool_stats: vec![RebalanceStats::default(); 2],
..Default::default()
};
let (_temp_dirs, store, competing_store) = crate::services::rebalance::test_two_pool_stores(Some(terminal.clone())).await;
let disk_stats = vec![
DiskStat {
total_space: 100,
available_space: 0,
},
DiskStat {
total_space: 100,
available_space: 100,
},
];
let initial_epoch = store.ctx.data_movement_operation_epoch();
let proof_guard = crate::services::notification_sys::without_cross_pool_fence_fleet_proof_for_test();
set_rebalance_disk_stats_override_for_test(store.id, disk_stats.clone());
let err = store
.init_rebalance_start(vec!["waiting-request".to_string()])
.await
.expect_err("initialization must fail closed while the live proof is missing");
assert!(
err.to_string()
.contains("pool activation requires a live fleet capability proof")
);
assert!(
crate::core::pools::is_pool_activation_fleet_proof_error(&err),
"the admin handler retries init only on this error shape: {err}"
);
assert_eq!(store.ctx.data_movement_operation_epoch(), initial_epoch);
{
let local = store.rebalance_meta.read().await;
let local = local.as_ref().expect("failed initialization must preserve the previous run");
assert_eq!(local.id, previous_id);
assert_eq!(local.stopped_at, terminal.stopped_at);
assert!(local.cancel.is_none(), "failed initialization must not start a worker");
}
let mut persisted = RebalanceMeta::new();
persisted
.load(store.pools[0].clone())
.await
.expect("failed initialization must leave durable metadata readable");
assert_eq!(persisted.id, previous_id);
assert_eq!(persisted.stopped_at, terminal.stopped_at);
drop(proof_guard);
set_rebalance_disk_stats_override_for_test(competing_store.id, disk_stats.clone());
let winning_id = competing_store
.init_rebalance_start(vec!["winning-request".to_string()])
.await
.expect("a ready request must be able to initialize one run after the failed attempt");
assert_ne!(winning_id, previous_id);
// The waiting request still holds its old local snapshot. Its retry must
// recheck durable state under the activation locks before publishing.
set_rebalance_disk_stats_override_for_test(store.id, disk_stats);
let remote_conflict = store
.init_rebalance_start(vec!["waiting-request".to_string()])
.await
.expect_err("the waiting request must not replace a durable competing run");
assert!(
matches!(
crate::data_movement::data_movement_stage_source(&remote_conflict),
Some(Error::RebalanceAlreadyRunning)
),
"the save-stage error must retain the typed competing-run conflict: {remote_conflict:?}"
);
let local_conflict = competing_store
.init_rebalance_start(vec!["another-request".to_string()])
.await
.expect_err("a local retry must not replace its active run");
assert!(
matches!(local_conflict, Error::RebalanceAlreadyRunning),
"the local admission check must report the competing-run conflict: {local_conflict:?}"
);
persisted
.load(store.pools[0].clone())
.await
.expect("the competing run must remain durable after both rejected retries");
assert_eq!(persisted.id, winning_id);
assert_eq!(persisted.pool_stats[0].buckets, vec!["winning-request".to_string()]);
assert_eq!(
store
.rebalance_meta
.read()
.await
.as_ref()
.expect("old local run must survive")
.id,
previous_id
);
assert_eq!(store.ctx.data_movement_operation_epoch(), initial_epoch + 1);
}
#[tokio::test]
#[serial_test::serial]
async fn rebalance_worker_skips_terminal_metadata_without_fleet_proof() {
@@ -1874,6 +2008,128 @@ mod tests {
);
}
async fn assert_worker_write_upgrade_revalidates_durable_state(change: &str) {
let rebalance_id = "rebalance-worker-write-upgrade";
let active = RebalanceMeta {
id: rebalance_id.to_string(),
percent_free_goal: 0.5,
pool_stats: vec![
RebalanceStats {
participating: true,
init_capacity: 100,
info: RebalanceInfo {
status: RebalStatus::Started,
..Default::default()
},
..Default::default()
},
RebalanceStats::default(),
],
..Default::default()
};
let (_temp_dirs, store, peer) =
crate::services::rebalance::test_two_pool_stores_with_isolated_node_contexts(Some(active.clone())).await;
let arrived = Arc::new(tokio::sync::Notify::new());
let release = Arc::new(tokio::sync::Notify::new());
let worker = Arc::clone(&store);
let mut tasks = tokio::task::JoinSet::new();
tasks.spawn(
super::super::runtime::REBALANCE_WORKER_WRITE_UPGRADE_BARRIER
.scope((Arc::clone(&arrived), Arc::clone(&release)), async move {
worker.start_rebalance_for_id(rebalance_id).await
}),
);
tokio::time::timeout(std::time::Duration::from_secs(30), arrived.notified())
.await
.expect("empty-queue activation must release its read fences before requesting write authority");
let mut changed = active.clone();
match change {
"stopped" => {
changed.stopped_at = Some(OffsetDateTime::now_utc());
changed.pool_stats[0].info.status = RebalStatus::Stopped;
}
"replaced" => changed.id = "replacement-rebalance-run".to_string(),
"decommission" => {
let mut pool_meta = peer.pool_meta.read().await.clone();
pool_meta.pools[0].decommission = Some(crate::core::pools::PoolDecommissionInfo {
start_time: Some(OffsetDateTime::now_utc()),
..Default::default()
});
pool_meta
.save(peer.pools.clone())
.await
.expect("a competing pool metadata writer must acquire the released pool fence");
}
_ => panic!("unknown durable state transition"),
}
changed
.save_with_opts(
peer.pools[0].clone(),
ObjectOptions {
write_completion: crate::object_api::WriteCompletion::TailDrained,
..Default::default()
},
)
.await
.expect("a competing run writer must acquire the released rebalance fence");
release.notify_one();
let result = tokio::time::timeout(std::time::Duration::from_secs(30), tasks.join_next())
.await
.expect("activation must finish after the durable transition")
.expect("activation task must be tracked")
.expect("activation must not panic");
match change {
"stopped" => result.expect("a durably stopped run must become an activation no-op"),
"replaced" => assert!(
result
.expect_err("a replaced run must reject stale activation")
.to_string()
.contains("expected"),
),
"decommission" => assert!(matches!(result, Err(Error::DecommissionAlreadyRunning))),
_ => unreachable!(),
}
let mut persisted = RebalanceMeta::new();
persisted
.load(peer.pools[0].clone())
.await
.expect("the competing durable state must remain readable");
assert_eq!(persisted.id, changed.id, "activation must not overwrite the run identity");
assert_eq!(persisted.stopped_at, changed.stopped_at, "activation must not overwrite the stop marker");
assert_eq!(persisted.pool_stats[0].info.status, changed.pool_stats[0].info.status);
assert!(
store
.rebalance_meta
.read()
.await
.as_ref()
.expect("local metadata must remain present")
.cancel
.is_none(),
"rejected activation must not install a worker token",
);
assert_activation_locks_released(&store).await;
}
#[tokio::test]
#[serial_test::serial]
async fn rebalance_worker_write_upgrade_rechecks_stop() {
assert_worker_write_upgrade_revalidates_durable_state("stopped").await;
}
#[tokio::test]
#[serial_test::serial]
async fn rebalance_worker_write_upgrade_rechecks_run_identity() {
assert_worker_write_upgrade_revalidates_durable_state("replaced").await;
}
#[tokio::test]
#[serial_test::serial]
async fn rebalance_worker_write_upgrade_rechecks_decommission() {
assert_worker_write_upgrade_revalidates_durable_state("decommission").await;
}
#[tokio::test]
#[serial_test::serial]
async fn rebalance_activation_adopts_commit_after_post_save_fence_loss() {
@@ -1440,7 +1440,7 @@ mod tests {
#[tokio::test]
#[serial_test::serial]
async fn real_rebalance_entry_progresses_while_peer_activation_waits_for_run_fence() {
async fn real_rebalance_peer_activates_while_entry_retains_run_fence() {
const REBALANCE_ID: &str = "rebalance-peer-activation-lock-order";
let (_temp_dirs, store, peer) = crate::services::rebalance::test_two_pool_stores_with_isolated_node_contexts(Some(
active_rebalance_meta(REBALANCE_ID),
@@ -1456,7 +1456,7 @@ mod tests {
let bucket = crate::disk::RUSTFS_META_BUCKET;
let object = "rebalance-peer-activation-object";
let version_id = uuid::Uuid::new_v4();
let payload = b"entry must drain before peer activation takes the pool fence".repeat(1024);
let payload = b"peer activation must not wait for this entry to drain".repeat(1024);
let source_set = store.pools[0].get_disks_by_key(object);
let target_set = store.pools[1].get_disks_by_key(object);
let opts = ObjectOptions {
@@ -1516,7 +1516,12 @@ mod tests {
);
tokio::time::timeout(StdDuration::from_secs(30), attempted.notified())
.await
.expect("peer activation must attempt the persisted rebalance write fence");
.expect("peer activation must attempt the persisted rebalance fence");
tokio::time::timeout(StdDuration::from_secs(30), activation_result)
.await
.expect("peer activation must finish while the entry remains paused")
.expect("peer activation result should be sent")
.expect("joining the same run must not require its active entry to release the run fence");
release.notify_one();
tokio::time::timeout(StdDuration::from_secs(30), async {
@@ -1531,10 +1536,6 @@ mod tests {
})
.await
.expect("entry and peer activation must both make progress");
activation_result
.await
.expect("peer activation result should be sent")
.expect("peer activation must not time out behind the entry it blocks");
let mut reader = target_set
.get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
@@ -27,6 +27,11 @@ use tokio::time::{Duration, Instant};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
#[cfg(test)]
tokio::task_local! {
pub(super) static REBALANCE_WORKER_WRITE_UPGRADE_BARRIER: (Arc<tokio::sync::Notify>, Arc<tokio::sync::Notify>);
}
pub(super) fn should_fail_repeated_rebalance_bucket_defer(
deferred_buckets: &mut HashSet<String>,
bucket: &str,
@@ -213,6 +218,22 @@ impl ECStore {
}
if must_persist {
// Shared admission grants no write authority. Release it before
// taking the exclusive metadata fences, then recheck stop/run/pool
// state before publishing the locally staged terminal candidate.
drop(activation_fence);
#[cfg(test)]
if let Ok((arrived, release)) = REBALANCE_WORKER_WRITE_UPGRADE_BARRIER.try_with(Clone::clone) {
arrived.notify_one();
release.notified().await;
}
let Some(activation_fence) = self
.fence_rebalance_worker_metadata_update(pool.clone(), expected_id.as_ref())
.await?
else {
cancel_tx.cancel();
return Ok(false);
};
let save_result = resolve_rebalance_meta_save_result(
self.save_rebalance_meta_under_activation_fence(
pool,
@@ -242,13 +263,14 @@ impl ECStore {
cancel_tx.cancel();
return Err(err);
}
} else {
if let Err(err) = activation_fence.ensure_held() {
let mut rebalance_meta = self.rebalance_meta.write().await;
rollback_local_rebalance_worker_activation(rebalance_meta.as_mut(), expected_id.as_ref(), &rx);
return Err(err);
}
drop(activation_fence);
}
if !must_persist && let Err(err) = activation_fence.ensure_held() {
let mut rebalance_meta = self.rebalance_meta.write().await;
rollback_local_rebalance_worker_activation(rebalance_meta.as_mut(), expected_id.as_ref(), &rx);
return Err(err);
}
drop(activation_fence);
if activation_outcome != RebalanceLocalActivationOutcome::Started {
return Ok(must_persist);
+1 -2
View File
@@ -82,9 +82,8 @@ use crate::{
use rustfs_filemeta::FileInfo;
use rustfs_rio::HashReader;
use rustfs_s3_client::{admin_handler_utils::AdminError, provider_versions::ProviderVersionCapabilities};
use rustfs_s3_types::S3ErrorCode;
use rustfs_utils::path::{SLASH_SEPARATOR, path_join};
use s3s::dto::BucketLifecycleConfiguration;
use s3s::{S3ErrorCode, dto::BucketLifecycleConfiguration};
use super::{
tier_handlers::{ERR_TIER_BUCKET_NOT_FOUND, ERR_TIER_CONNECT_ERR, ERR_TIER_INVALID_CREDENTIALS, ERR_TIER_PERM_ERR},
+40 -4
View File
@@ -31,6 +31,7 @@ use crate::disk::DiskAPI;
pub(crate) enum BucketInfoQuorum {
Read,
Write,
HealCandidates,
}
impl SetDisks {
@@ -104,7 +105,9 @@ impl SetDisks {
// class. A namespace read must intersect that majority; object
// readers still enforce the persisted layout's data-shard quorum.
BucketInfoQuorum::Read => reduce_read_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, disk_count.div_ceil(2).max(1)),
BucketInfoQuorum::Write => reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, disk_count / 2 + 1),
BucketInfoQuorum::Write | BucketInfoQuorum::HealCandidates => {
reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, disk_count / 2 + 1)
}
};
if let Some(err) = error {
return Err(err.into());
@@ -126,6 +129,29 @@ impl SetDisks {
self.list_bucket_with_quorum(BucketInfoQuorum::Write).await
}
pub(crate) async fn bucket_needs_reconstruction(&self, bucket: &str) -> Result<bool> {
let results = join_all(self.disk_inventory().await.into_iter().map(|disk| async move {
match disk {
Some(disk) => disk.stat_volume(bucket).await,
None => Err(DiskError::DiskNotFound),
}
}))
.await;
let mut present = false;
let mut missing = false;
for result in results {
match result {
Ok(_) => present = true,
Err(DiskError::VolumeNotFound) => missing = true,
Err(error) => return Err(error.into()),
}
}
if !present {
return Err(Error::VolumeNotFound);
}
Ok(missing)
}
pub(crate) async fn list_bucket_with_quorum(&self, quorum: BucketInfoQuorum) -> Result<(Vec<BucketInfo>, bool)> {
let disks = self.disk_inventory().await;
let write_quorum = (disks.len() / 2) + 1;
@@ -134,6 +160,7 @@ impl SetDisks {
let required = match quorum {
BucketInfoQuorum::Read => disks.len().div_ceil(2).max(1),
BucketInfoQuorum::Write => write_quorum,
BucketInfoQuorum::HealCandidates => disks.len().max(1),
};
let mut futures = Vec::with_capacity(disks.len());
@@ -165,7 +192,9 @@ impl SetDisks {
let error = match quorum {
BucketInfoQuorum::Read => reduce_read_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, required),
BucketInfoQuorum::Write => reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, required),
BucketInfoQuorum::Write | BucketInfoQuorum::HealCandidates => {
reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, required)
}
};
if let Some(err) = error {
return Err(err.into());
@@ -190,10 +219,17 @@ impl SetDisks {
}
}
topology_complete &= counts.values().all(|(count, _)| *count >= write_quorum);
let minimum_presence = if matches!(quorum, BucketInfoQuorum::HealCandidates) {
// Candidate presence is not mutation authority. Every disk must answer,
// and the heal owner validates the committed generation before recreation.
1
} else {
topology_complete &= counts.values().all(|(count, _)| *count >= write_quorum);
required
};
let mut buckets = counts
.into_values()
.filter_map(|(count, bucket)| (count >= required).then_some(bucket))
.filter_map(|(count, bucket)| (count >= minimum_presence).then_some(bucket))
.collect::<Vec<_>>();
buckets.sort_by(|left, right| left.name.cmp(&right.name));
Ok((buckets, topology_complete))
+736 -122
View File
@@ -63,7 +63,6 @@ use crate::crash_inject::{self, CrashPoint};
use crate::disk::DiskAPI;
#[cfg(test)]
use crate::disk::DiskOption;
#[cfg(test)]
use crate::disk::STORAGE_FORMAT_FILE;
#[cfg(test)]
use crate::disk::new_disk;
@@ -231,6 +230,60 @@ pub(crate) struct StaleMultipartCleanupGuard {
lock_guard: ObjectLockDiagGuard,
}
/// Exact-key recovery is admitted only after every disk proves that the upload
/// metadata is absent. Keep the checked disk generation and upload lock through
/// deletion; a quorum miss is not proof that an upload is an orphan.
/// Final object data, completion replay identity, and old-data cleanup receipts
/// live under the destination bucket/object, never this staging path. This also
/// permits removing an exact orphan from a previous bucket incarnation without
/// claiming that its UploadID belongs to the current incarnation.
pub(crate) struct OrphanMultipartCleanupGuard {
set: Arc<SetDisks>,
disks: Vec<Option<DiskStore>>,
upload_path: String,
lock_guard: Arc<ObjectLockDiagGuard>,
exists: bool,
}
impl OrphanMultipartCleanupGuard {
pub(crate) async fn delete(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<bool> {
fence_commit_on_lock_loss(Some(&self.lock_guard), "orphan_multipart_cleanup", &self.upload_path)?;
ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
if !self.exists {
return Ok(false);
}
let results = join_all(self.disks.iter().map(|disk| async {
let Some(disk) = disk else {
return Err(DiskError::DiskNotFound);
};
match disk
.delete(
RUSTFS_META_MULTIPART_BUCKET,
&self.upload_path,
crate::disk::DeleteOptions {
recursive: true,
..Default::default()
},
)
.await
{
Err(DiskError::FileNotFound | DiskError::PathNotFound | DiskError::VolumeNotFound) => Ok(()),
result => result,
}
}))
.await;
self.set.record_capacity_scope_if_needed(None, &self.disks);
let errors = results.into_iter().map(|result| result.err()).collect::<Vec<_>>();
if let Some(err) = reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, self.disks.len()) {
return Err(to_object_err(err.into(), vec![bucket, object]));
}
fence_commit_on_lock_loss(Some(&self.lock_guard), "orphan_multipart_cleanup", &self.upload_path)?;
ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
Ok(true)
}
}
pub(crate) struct DataMovementMultipartAbortGuard {
upload_path: String,
write_quorum: Option<usize>,
@@ -557,6 +610,34 @@ fn multipart_bucket_incarnation_matches(metadata: &HashMap<String, String>, expe
matches!(multipart_bucket_incarnation_id(metadata), Ok(Some(actual)) if actual == expected)
}
fn orphan_multipart_upload_id_is_canonical(upload_id: &str) -> bool {
let Ok(decoded) = base64_simd::URL_SAFE_NO_PAD.decode_to_vec(upload_id.as_bytes()) else {
return false;
};
let Ok(decoded) = std::str::from_utf8(&decoded) else {
return false;
};
let Some((_, raw)) = decoded.split_once('.') else {
return false;
};
let Some((uuid, timestamp)) = raw.split_once('x') else {
return false;
};
let Ok(id) = Uuid::parse_str(uuid) else {
return false;
};
if id.is_nil() || id.to_string() != uuid {
return false;
}
let Ok(timestamp_ns) = timestamp.parse::<i128>() else {
return false;
};
if timestamp_ns.to_string() != timestamp {
return false;
}
OffsetDateTime::from_unix_timestamp_nanos(timestamp_ns).is_ok()
}
fn validate_multipart_bucket_incarnation(
metadata: &HashMap<String, String>,
bucket: &str,
@@ -1246,6 +1327,114 @@ impl SetDisks {
})
}
pub(crate) async fn lock_orphan_multipart_cleanup(
self: &Arc<Self>,
bucket: &str,
object: &str,
upload_id: &str,
opts: &ObjectOptions,
held_guards: &[OrphanMultipartCleanupGuard],
) -> Result<OrphanMultipartCleanupGuard> {
let invalid_upload = || StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned());
if opts.data_movement
|| opts.no_lock
|| self.local_lock_manager.is_disabled()
|| !orphan_multipart_upload_id_is_canonical(upload_id)
{
return Err(invalid_upload());
}
// This path is derived from the caller's exact bucket/key and a canonical
// UUID/timestamp upload identity, never from the shared MPU directory list.
let upload_path = Self::get_upload_id_dir(bucket, object, upload_id);
// Pool labels do not partition the raw ObjectKey sent to lock clients.
// Reuse the actual owner only for this exact path and proven backend;
// independent distributed domains still need their own guards.
let distributed = self.ctx.is_dist_erasure().await;
let mut lock_guard = None;
for held in held_guards {
if held.upload_path != upload_path {
continue;
}
let shared_domain = match (distributed, held.set.ctx.is_dist_erasure().await) {
(false, false) => self.shares_namespace_lock_domain(&held.set).await,
(true, true) => {
let same = super::super::same_distributed_lock_domain(&self.lockers, &held.set.lockers);
if !same
&& self
.lockers
.iter()
.any(|client| held.set.lockers.iter().any(|held_client| Arc::ptr_eq(client, held_client)))
{
// A guard for only part of another quorum is not proof
// of ownership there, and acquiring it again may wait
// on ourselves. Preserve all staging in this topology.
return Err(Error::other("orphan multipart cleanup cannot cover partially overlapping lock domains"));
}
same
}
_ => false,
};
if shared_domain {
lock_guard = Some(Arc::clone(&held.lock_guard));
break;
}
}
let lock_guard = match lock_guard {
Some(guard) => guard,
None => Arc::new(
self.acquire_write_lock_diag("orphan_multipart_cleanup", RUSTFS_META_MULTIPART_BUCKET, &upload_path)
.await?,
),
};
fence_commit_on_lock_loss(Some(&lock_guard), "orphan_multipart_cleanup", &upload_path)?;
ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
let disks = self.disks.read().await.clone();
if disks.len() != self.set_drive_count || disks.is_empty() {
return Err(Error::ErasureReadQuorum);
}
// Raw full fan-out cannot quorum-short-circuit or interpret an empty
// version list as absence of the physical metadata file.
let (_, errors) = Self::read_all_raw_file_info(&disks, RUSTFS_META_MULTIPART_BUCKET, &upload_path, false).await;
if errors.len() != disks.len() {
return Err(Error::ErasureReadQuorum);
}
for error in errors {
match error {
Some(DiskError::FileNotFound | DiskError::VolumeNotFound) => {}
// Even one valid metadata copy may still be recoverable. Never
// convert a partial, corrupt, or unreachable upload into debris.
None => return Err(invalid_upload()),
Some(error) => return Err(to_object_err(error.into(), vec![bucket, object, upload_id])),
}
}
let mut exists = false;
for disk in &disks {
let Some(disk) = disk else {
return Err(Error::ErasureReadQuorum);
};
match disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_path, -1).await {
Ok(entries) => {
// Raw metadata reads report a directory named xl.meta as
// FileNotFound. Its physical presence is unknown/corrupt
// metadata, not proof that this staging can be deleted.
if entries.iter().any(|entry| entry.trim_end_matches('/') == STORAGE_FORMAT_FILE) {
return Err(to_object_err(DiskError::FileCorrupt.into(), vec![bucket, object, upload_id]));
}
exists = true;
}
Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => {}
Err(error) => return Err(to_object_err(error.into(), vec![bucket, object, upload_id])),
}
}
Ok(OrphanMultipartCleanupGuard {
set: self.clone(),
disks,
upload_path,
lock_guard,
exists,
})
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn list_multipart_uploads_for_incarnation(
&self,
@@ -1678,6 +1867,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
drop(admission_guard);
}
let mut commit_owns_cleanup = false;
let result: Result<PartInfo> = async {
let erasure =
Arc::new(coding::Erasure::try_new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size)
@@ -1881,120 +2071,171 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
.await?;
(Some(upload_guard), Some(part_guard))
};
let (commit_fi, _) = self
.check_upload_id_exists_with_opts(bucket, object, upload_id, multipart_size_limit.is_some(), opts)
.await?;
let commit_size_limit = multipart_size_limit_from_metadata(&commit_fi.metadata)?;
if commit_size_limit != multipart_size_limit {
return Err(Error::InvalidArgument(
"multipart upload".to_string(),
SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
"size limit metadata changed or is missing".to_string(),
));
}
ensure_data_movement_upload_access(&commit_fi, bucket, object, upload_id, opts)?;
ensure_multipart_bucket_incarnation(
&self.ctx,
&commit_fi,
bucket,
object,
upload_id,
opts.expected_bucket_incarnation_id,
)
.await?;
#[cfg(test)]
pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost).await;
fence_commit_on_lock_loss(_upload_commit_guard.as_ref(), "put_object_part_commit", &upload_id_path)?;
fence_commit_on_lock_loss(_part_commit_guard.as_ref(), "put_object_part_commit", &part_lock_path)?;
if opts
.namespace_lock_fence
.as_ref()
.is_some_and(NamespaceLockFence::is_lock_lost)
{
return Err(StorageError::NamespaceLockQuorumUnavailable {
mode: "put_object_part_outer_lock",
bucket: bucket.to_string(),
object: object.to_string(),
required: 1,
achieved: 0,
});
}
ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
if let Some(limit) = commit_size_limit {
let current_size = self
.current_multipart_logical_size(bucket, object, upload_id, &upload_id_path, &commit_fi, part_id)
.await?;
let candidate_size = u64::try_from(actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
admitted_multipart_size(current_size, candidate_size, limit)?;
}
let decommission_capacity_guard = if let Some(store) = opts.decommission_capacity_admission.as_ref() {
Some(
store
.acquire_external_decommission_capacity_fence(
&[self.pool_index],
DecommissionCapacityAdmission::ExistingMultipart,
)
.await?,
)
} else {
None
};
#[cfg(test)]
pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterCapacityAdmission).await;
if decommission_capacity_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
return Err(StorageError::NamespaceLockQuorumUnavailable {
mode: "put_object_part_decommission_capacity",
bucket: bucket.to_string(),
object: object.to_string(),
required: 1,
achieved: 0,
});
}
let rename_result = self
.rename_part(
&shuffle_disks,
RUSTFS_META_TMP_BUCKET,
&tmp_part_path,
RUSTFS_META_MULTIPART_BUCKET,
&part_path,
part_info_buff.into(),
write_quorum,
Some(MultipartWriteQuorumContext {
stage: MULTIPART_WRITE_QUORUM_RENAME_PART,
// Only guards owned by this operation may outlive its caller. Internal
// data-movement and borrowed-lock callers retain their synchronous
// lifetime; a fence by itself cannot transfer the outer lock lease.
let detach_commit_owner = _upload_commit_guard.is_some()
&& opts.namespace_lock_fence.is_none()
&& (opts.bucket_lifecycle_lock_fence.is_none() || opts.bucket_lifecycle_lock_owner.is_some());
let commit_set = self.clone();
let commit_bucket = bucket.to_owned();
let commit_object = object.to_owned();
let commit_upload_id = upload_id.to_owned();
let commit_upload_id_path = upload_id_path.clone();
let commit_tmp_part = tmp_part.clone();
let commit_opts = opts.clone();
commit_owns_cleanup = true;
// The body is already staged. Keep every commit guard through the
// prepare, publish, settlement/rollback and temporary cleanup I/O,
// even when the HTTP request stops awaiting this task.
let commit = async move {
let _upload_commit_guard = _upload_commit_guard;
let _part_commit_guard = _part_commit_guard;
let _capped_staging_guard = _capped_staging_guard;
let bucket = commit_bucket.as_str();
let object = commit_object.as_str();
let upload_id = commit_upload_id.as_str();
let upload_id_path = commit_upload_id_path;
let opts = &commit_opts;
let result: Result<PartInfo> = async {
let (commit_fi, _) = commit_set
.check_upload_id_exists_with_opts(bucket, object, upload_id, multipart_size_limit.is_some(), opts)
.await?;
let commit_size_limit = multipart_size_limit_from_metadata(&commit_fi.metadata)?;
if commit_size_limit != multipart_size_limit {
return Err(Error::InvalidArgument(
"multipart upload".to_string(),
SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
"size limit metadata changed or is missing".to_string(),
));
}
ensure_data_movement_upload_access(&commit_fi, bucket, object, upload_id, opts)?;
ensure_multipart_bucket_incarnation(
&commit_set.ctx,
&commit_fi,
bucket,
object,
upload_id,
part_number: Some(part_id),
}),
)
opts.expected_bucket_incarnation_id,
)
.await?;
#[cfg(test)]
pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost).await;
fence_commit_on_lock_loss(_upload_commit_guard.as_ref(), "put_object_part_commit", &upload_id_path)?;
fence_commit_on_lock_loss(_part_commit_guard.as_ref(), "put_object_part_commit", &part_lock_path)?;
if opts
.namespace_lock_fence
.as_ref()
.is_some_and(NamespaceLockFence::is_lock_lost)
{
return Err(StorageError::NamespaceLockQuorumUnavailable {
mode: "put_object_part_outer_lock",
bucket: bucket.to_string(),
object: object.to_string(),
required: 1,
achieved: 0,
});
}
ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
if let Some(limit) = commit_size_limit {
let current_size = commit_set
.current_multipart_logical_size(bucket, object, upload_id, &upload_id_path, &commit_fi, part_id)
.await?;
let candidate_size = u64::try_from(actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
admitted_multipart_size(current_size, candidate_size, limit)?;
}
let decommission_capacity_guard = if let Some(store) = opts.decommission_capacity_admission.as_ref() {
Some(
store
.acquire_external_decommission_capacity_fence(
&[commit_set.pool_index],
DecommissionCapacityAdmission::ExistingMultipart,
)
.await?,
)
} else {
None
};
#[cfg(test)]
pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterCapacityAdmission).await;
if decommission_capacity_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
return Err(StorageError::NamespaceLockQuorumUnavailable {
mode: "put_object_part_decommission_capacity",
bucket: bucket.to_string(),
object: object.to_string(),
required: 1,
achieved: 0,
});
}
let rename_result = commit_set
.rename_part(
&shuffle_disks,
RUSTFS_META_TMP_BUCKET,
&tmp_part_path,
RUSTFS_META_MULTIPART_BUCKET,
&part_path,
part_info_buff.into(),
write_quorum,
Some(MultipartWriteQuorumContext {
stage: MULTIPART_WRITE_QUORUM_RENAME_PART,
bucket,
object,
upload_id,
part_number: Some(part_id),
}),
)
.await;
drop(decommission_capacity_guard);
let _ = rename_result?;
#[cfg(test)]
observe_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost);
#[cfg(test)]
pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterRename).await;
let ret: PartInfo = PartInfo {
etag: Some(etag.clone()),
part_num: part_id,
last_mod: Some(OffsetDateTime::now_utc()),
size: w_size,
actual_size,
};
// error!("put_object_part ret {:?}", &ret);
Ok(ret)
}
.await;
drop(decommission_capacity_guard);
let _ = rename_result?;
#[cfg(test)]
observe_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost);
#[cfg(test)]
pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterRename).await;
drop(_part_commit_guard);
drop(_upload_commit_guard);
let ret: PartInfo = PartInfo {
etag: Some(etag.clone()),
part_num: part_id,
last_mod: Some(OffsetDateTime::now_utc()),
size: w_size,
actual_size,
if result.is_err()
&& let Err(err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_part).await
{
warn!(
event = EVENT_SET_DISK_MULTIPART,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
result = "cleanup_failed",
bucket,
object,
error = ?err,
"failed to clean multipart commit staging"
);
}
result
};
// error!("put_object_part ret {:?}", &ret);
Ok(ret)
if detach_commit_owner {
tokio::spawn(commit)
.await
.map_err(|err| Error::other_with_context("put_object_part commit task failed", err))?
} else {
commit.await
}
}
.await;
if result.is_err()
if !commit_owns_cleanup
&& result.is_err()
&& let Err(err) = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_part).await
{
warn!(tmp_part = %tmp_part, error = ?err, "failed to cleanup multipart temporary data");
@@ -4233,6 +4474,300 @@ mod tests {
}
}
#[test]
fn orphan_multipart_identity_requires_canonical_upload_id() {
let created = OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(10);
let id = Uuid::new_v4();
let timestamp = (created + time::Duration::seconds(1)).unix_timestamp_nanos();
let encode = |raw: &str| runtime_sources::deployment_upload_id(raw);
let upload_id = encode(&format!("{id}x{timestamp}"));
assert!(orphan_multipart_upload_id_is_canonical(&upload_id));
for raw in [
format!("{}x{timestamp}", Uuid::nil()),
format!("{}x{timestamp}", id.simple()),
format!("{id}x0{timestamp}"),
format!("{id}x+{timestamp}"),
format!("{id}x{timestamp}/../another-upload"),
format!("../{id}x{timestamp}"),
format!("{id}x{}", i128::MAX),
] {
assert!(
!orphan_multipart_upload_id_is_canonical(&encode(&raw)),
"noncanonical upload identity must be rejected: {raw}"
);
}
assert!(!orphan_multipart_upload_id_is_canonical("not-base64"));
}
#[tokio::test]
async fn orphan_multipart_cleanup_requires_every_disk_to_prove_metadata_absence() {
let (_temp_dirs, disks, set) = hermetic_set_disks(4).await;
let bucket = "orphan-multipart-proof";
let object = "object";
make_bucket_on_all(&disks, bucket).await;
let upload = set
.new_multipart_upload(bucket, object, &ObjectOptions::default())
.await
.expect("create upload");
let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload.upload_id);
let metadata_path = format!("{upload_path}/{STORAGE_FORMAT_FILE}");
let leftover_path = format!("{upload_path}/part.late");
for disk in &disks {
disk.write_all(RUSTFS_META_MULTIPART_BUCKET, &leftover_path, Bytes::from_static(b"late part"))
.await
.expect("write late part fixture");
}
for disk in &disks[..3] {
disk.delete(RUSTFS_META_MULTIPART_BUCKET, &metadata_path, Default::default())
.await
.expect("remove majority metadata");
}
let opts = ObjectOptions::default();
assert!(
set.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
.await
.is_err(),
"one surviving metadata copy must block orphan recovery"
);
disks[3]
.write_all(RUSTFS_META_MULTIPART_BUCKET, &metadata_path, Bytes::from_static(b"corrupt metadata"))
.await
.expect("corrupt final metadata copy");
assert!(
set.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
.await
.is_err(),
"corrupt metadata must not be treated as absence"
);
disks[3]
.delete(RUSTFS_META_MULTIPART_BUCKET, &metadata_path, Default::default())
.await
.expect("remove final metadata copy");
set.disks.write().await[3] = None;
assert!(
set.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
.await
.is_err(),
"an unreachable disk must not be hidden by absence quorum"
);
set.disks.write().await[3] = Some(disks[3].clone());
for disk in &disks {
assert_eq!(
disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &leftover_path)
.await
.expect("failed preflight must preserve every part"),
Bytes::from_static(b"late part")
);
}
let guard = set
.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
.await
.expect("all disks now prove metadata absence");
assert!(guard.delete(bucket, object, &opts).await.expect("recover exact staging"));
drop(guard);
for disk in &disks {
assert!(matches!(
disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_path, -1).await,
Err(DiskError::FileNotFound | DiskError::VolumeNotFound)
));
}
let guard = set
.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &opts, &[])
.await
.expect("already absent staging remains provably absent");
assert!(
!guard
.delete(bucket, object, &opts)
.await
.expect("absence does not report recovery")
);
}
#[tokio::test]
async fn orphan_multipart_cleanup_rejects_metadata_directory_on_one_disk() {
let (temp_dirs, disks, set) = hermetic_set_disks(4).await;
let bucket = "orphan-multipart-directory";
let object = "object";
make_bucket_on_all(&disks, bucket).await;
let upload = set
.new_multipart_upload(bucket, object, &ObjectOptions::default())
.await
.expect("create upload");
let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload.upload_id);
let metadata_path = format!("{upload_path}/{STORAGE_FORMAT_FILE}");
let leftover_path = format!("{upload_path}/part.late");
for disk in &disks {
disk.delete(RUSTFS_META_MULTIPART_BUCKET, &metadata_path, Default::default())
.await
.expect("remove upload metadata");
disk.write_all(RUSTFS_META_MULTIPART_BUCKET, &leftover_path, Bytes::from_static(b"late part"))
.await
.expect("write leftover part");
}
let metadata_directory = temp_dirs[3].path().join(RUSTFS_META_MULTIPART_BUCKET).join(&metadata_path);
tokio::fs::create_dir_all(&metadata_directory)
.await
.expect("replace one missing metadata file with a directory");
let (_, errors) =
SetDisks::read_all_raw_file_info(&set.disks.read().await, RUSTFS_META_MULTIPART_BUCKET, &upload_path, false).await;
assert!(
errors.iter().all(|error| matches!(error, Some(DiskError::FileNotFound))),
"raw metadata reads alone must reproduce the directory-as-absence case: {errors:?}"
);
assert!(
set.lock_orphan_multipart_cleanup(bucket, object, &upload.upload_id, &ObjectOptions::default(), &[])
.await
.is_err(),
"one physical metadata directory must block cleanup on every disk"
);
assert!(metadata_directory.is_dir(), "failed preflight must preserve the metadata directory");
for disk in &disks {
assert_eq!(
disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &leftover_path)
.await
.expect("failed preflight must preserve every leftover part"),
Bytes::from_static(b"late part")
);
}
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum OrphanMultipartLockDomainCase {
Shared,
Independent,
PartiallyOverlapping,
}
async fn check_orphan_multipart_distributed_lock_owners(case: OrphanMultipartLockDomainCase) {
let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
ctx.update_erasure_type(SetupType::DistErasure).await;
let first_client: Arc<dyn LockClient> =
Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
let second_client: Arc<dyn LockClient> =
Arc::new(LocalClient::with_manager(Arc::new(rustfs_lock::GlobalLockManager::new())));
let second_clients = match case {
OrphanMultipartLockDomainCase::Shared => vec![first_client.clone()],
OrphanMultipartLockDomainCase::Independent => vec![second_client],
OrphanMultipartLockDomainCase::PartiallyOverlapping => vec![first_client.clone(), second_client],
};
let (first_dirs, first_disks, first_set) =
hermetic_set_disks_with_lockers_and_ctx(4, 0, 2, vec![first_client], ctx.clone()).await;
let (second_dirs, second_disks, second_set) = hermetic_set_disks_with_lockers_and_ctx(4, 1, 2, second_clients, ctx).await;
let bucket = "orphan-multipart-dist-locks";
let object = "object";
make_bucket_on_all(&first_disks, bucket).await;
make_bucket_on_all(&second_disks, bucket).await;
let upload_id = runtime_sources::deployment_upload_id(&format!(
"{}x{}",
Uuid::new_v4(),
OffsetDateTime::now_utc().unix_timestamp_nanos()
));
let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
for directory in first_dirs.iter().chain(second_dirs.iter()) {
let staging = directory.path().join(RUSTFS_META_MULTIPART_BUCKET).join(&upload_path);
tokio::fs::create_dir_all(&staging)
.await
.expect("create exact orphan staging");
tokio::fs::write(staging.join("part.late"), b"late part")
.await
.expect("write orphan part");
}
let opts = ObjectOptions::default();
let first_guard = first_set
.lock_orphan_multipart_cleanup(bucket, object, &upload_id, &opts, &[])
.await
.expect("first domain has an orphan");
let second_result = tokio::time::timeout(
Duration::from_secs(10),
second_set.lock_orphan_multipart_cleanup(bucket, object, &upload_id, &opts, std::slice::from_ref(&first_guard)),
)
.await
.expect("lock-domain admission must not wait on a guard it already owns");
if case == OrphanMultipartLockDomainCase::PartiallyOverlapping {
let error = match second_result {
Ok(_) => panic!("one existing quorum cannot authorize a partially overlapping domain"),
Err(error) => error,
};
assert!(error.to_string().contains("partially overlapping lock domains"));
for directory in first_dirs.iter().chain(second_dirs.iter()) {
assert!(
directory
.path()
.join(RUSTFS_META_MULTIPART_BUCKET)
.join(&upload_path)
.join("part.late")
.exists(),
"uncertain lock coverage must preserve every physical part"
);
}
return;
}
let second_guard = second_result.expect("the second pool has a proven lock domain");
assert_eq!(
Arc::ptr_eq(&first_guard.lock_guard, &second_guard.lock_guard),
case == OrphanMultipartLockDomainCase::Shared,
"only the same exact backend may share a real upload guard across pool labels"
);
let second_namespace = second_set
.new_ns_lock(RUSTFS_META_MULTIPART_BUCKET, &upload_path)
.await
.unwrap();
let mut second_writer = Box::pin(second_namespace.get_write_lock(Duration::from_secs(10)));
assert!(futures::poll!(&mut second_writer).is_pending());
if case == OrphanMultipartLockDomainCase::Shared {
drop(first_guard);
assert!(
futures::poll!(&mut second_writer).is_pending(),
"the second pool must retain the shared real owner after the first guard drops"
);
drop(second_guard);
drop(
tokio::time::timeout(Duration::from_secs(10), second_writer)
.await
.expect("the shared backend should release after the last owner")
.expect("shared-domain writer resumes"),
);
return;
}
let first_namespace = first_set
.new_ns_lock(RUSTFS_META_MULTIPART_BUCKET, &upload_path)
.await
.unwrap();
let mut first_writer = Box::pin(first_namespace.get_write_lock(Duration::from_secs(10)));
assert!(futures::poll!(&mut first_writer).is_pending());
drop(first_guard);
drop(
tokio::time::timeout(Duration::from_secs(10), first_writer)
.await
.expect("first domain should release independently")
.expect("first writer resumes"),
);
assert!(futures::poll!(&mut second_writer).is_pending(), "the second real owner must remain held");
drop(second_guard);
drop(
tokio::time::timeout(Duration::from_secs(10), second_writer)
.await
.expect("second domain should release with its owner")
.expect("second writer resumes"),
);
}
#[tokio::test]
async fn orphan_multipart_cleanup_keeps_separate_distributed_lock_owners() {
check_orphan_multipart_distributed_lock_owners(OrphanMultipartLockDomainCase::Independent).await;
}
#[tokio::test]
async fn orphan_multipart_cleanup_reuses_shared_distributed_owner_across_pools() {
check_orphan_multipart_distributed_lock_owners(OrphanMultipartLockDomainCase::Shared).await;
}
#[tokio::test]
async fn orphan_multipart_cleanup_rejects_partially_overlapping_distributed_domains() {
check_orphan_multipart_distributed_lock_owners(OrphanMultipartLockDomainCase::PartiallyOverlapping).await;
}
async fn stage_upload_with_create_opts(
set_disks: &Arc<SetDisks>,
bucket: &str,
@@ -6370,6 +6905,76 @@ mod tests {
.expect("abort should delete the upload after UploadPart releases the lock");
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn cancelled_put_object_part_keeps_upload_lock_through_commit() {
for pause in [
MultipartCommitPause::PutPartAfterCapacityAdmission,
MultipartCommitPause::PutPartAfterRename,
] {
let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
let bucket = "multipart-cancelled-part-commit-bucket";
let object = "object";
make_bucket_on_all(&disk_stores, bucket).await;
let upload = set_disks
.new_multipart_upload(bucket, object, &ObjectOptions::default())
.await
.expect("multipart upload should be created");
let upload_id = upload.upload_id;
let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_path.clone()));
let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
let barrier = MultipartCommitBarrier::install(bucket, object, pause);
let put_store = set_disks.clone();
let put_upload_id = upload_id.clone();
let put = tokio::spawn(async move {
let mut reader = PutObjReader::from_vec(vec![0x45; 4096]);
put_store
.put_object_part(bucket, object, &put_upload_id, 1, &mut reader, &ObjectOptions::default())
.await
});
barrier.wait_until_paused().await;
put.abort();
assert!(
put.await
.expect_err("the request waiter should remain cancellable")
.is_cancelled()
);
let abort_store = set_disks.clone();
let abort_upload_id = upload_id.clone();
let mut abort = tokio::spawn(async move {
abort_store
.abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default())
.await
});
signaling.wait_for_attempts(2).await;
assert!(
tokio::time::timeout(Duration::from_millis(100), &mut abort).await.is_err(),
"Abort must not acquire the upload lock while a cancelled request's commit is paused at {pause:?}"
);
barrier.release();
tokio::time::timeout(Duration::from_secs(10), abort)
.await
.expect("Abort should finish after the commit drains")
.expect("Abort task should not panic")
.expect("Abort should delete the committed part staging");
for disk in &disk_stores {
assert!(
matches!(
disk.list_dir(bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_path, -1).await,
Err(DiskError::FileNotFound)
),
"the upload directory must be absent after the commit and Abort finish"
);
}
}
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn put_object_part_different_part_numbers_commit_concurrently() {
@@ -9799,6 +10404,18 @@ mod tests {
"the post-commit crash must leave the upload listable for reclamation"
);
let cleanup = set_disks
.lock_orphan_multipart_cleanup(bucket, object, &u_new, &ObjectOptions::default(), &[])
.await
.expect("the committed upload has no staging metadata on any disk");
assert!(
cleanup
.delete(bucket, object, &ObjectOptions::default())
.await
.expect("recover completed upload staging")
);
drop(cleanup);
// The commit rename already consumed the upload's xl.meta, so the
// retry cannot read staging. It must still return the committed
// object: AWS CompleteMultipartUpload is idempotent for the same
@@ -9817,21 +10434,6 @@ mod tests {
!upload_is_listed(&set_disks, bucket, object, &u_new).await,
"the idempotent retry must reclaim staging left behind by the crash"
);
// Reclaim the leftover exactly as the production tail does: delete_all
// on the upload path (abort_multipart_upload cannot — the upload's
// metadata is gone, so it too answers InvalidUploadID).
let upload_dir = SetDisks::get_upload_id_dir(bucket, object, &u_new);
set_disks
.delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_dir)
.await
.expect("sweeping the leftover upload staging should succeed");
assert!(
!upload_is_listed(&set_disks, bucket, object, &u_new).await,
"reclaiming the upload must drop it from the listing"
);
let (body_after, _) = read_object(&set_disks, bucket, object).await;
assert_eq!(body_after, new, "reclaiming the leftover upload must not disturb the committed object");
}
#[tokio::test]
@@ -9878,6 +10480,18 @@ mod tests {
.expect("the failed post-commit completion should release its object guard"),
);
let cleanup = set_disks
.lock_orphan_multipart_cleanup(bucket, object, &u_new, &ObjectOptions::default(), &[])
.await
.expect("committed staging has no upload metadata");
assert!(
cleanup
.delete(bucket, object, &ObjectOptions::default())
.await
.expect("orphan recovery must only remove upload staging")
);
drop(cleanup);
let (body, _) = read_object(&set_disks, bucket, object).await;
assert_eq!(body, new, "the committed replacement must remain readable after the crash");
let mut receipts = 0;
File diff suppressed because it is too large Load Diff
+183 -1
View File
@@ -404,7 +404,17 @@ impl ECStore {
expected: uuid::Uuid,
opts: &HealOpts,
) -> Result<HealResultItem> {
self.run_bucket_heal_at_incarnation(bucket, expected, opts, |store, bucket, opts| async move {
self.run_bucket_heal_at_incarnation(bucket, expected, opts, move |store, bucket, mut opts| async move {
if !opts.recreate && !opts.remove {
let mut missing = false;
for (pool, set) in store.heal_erasure_set_scopes(&opts).await? {
missing |= store.pools[pool].get_disks(set).bucket_needs_reconstruction(&bucket).await?;
}
if missing {
crate::bucket::metadata_sys::validate_bucket_reconstruction_in(&store.ctx, &bucket, expected).await?;
opts.recreate = true;
}
}
store.heal_bucket(&bucket, &opts).await
})
.await
@@ -1451,6 +1461,178 @@ mod tests {
.expect("minimal pool should build")
}
#[tokio::test]
#[serial_test::serial]
async fn admin_bucket_recovery_recreates_missing_directories_at_current_incarnation() {
let (root, store, shutdown) = multi_pool_heal_store().await;
let bucket = "admin-bucket-recovery";
store
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("create a committed bucket generation");
let incarnation = store
.bucket_incarnation_id_from_disk(bucket)
.await
.expect("persisted generation");
for disk in 1..4 {
tokio::fs::remove_dir(root.path().join(format!("pool0-disk{disk}/{bucket}")))
.await
.expect("remove only an empty fixture bucket directory");
}
store
.heal_bucket_at_incarnation(bucket, incarnation, &HealOpts::default())
.await
.expect("default administrator recovery should use committed generation authority");
for disk in 0..4 {
assert!(root.path().join(format!("pool0-disk{disk}/{bucket}")).is_dir());
}
assert_eq!(
store
.bucket_incarnation_id_from_disk(bucket)
.await
.expect("recovered generation"),
incarnation
);
shutdown.cancel();
}
#[tokio::test]
#[serial_test::serial]
async fn admin_bucket_recovery_rejects_missing_corrupt_uncommitted_and_retired_authority() {
let (root, store, shutdown) = multi_pool_heal_store().await;
for state in ["missing", "corrupt", "uncommitted", "mismatched", "retired"] {
let bucket = format!("recovery-authority-{state}");
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create fixture generation");
let incarnation = store
.bucket_incarnation_id_from_disk(&bucket)
.await
.expect("fixture identity");
let (mut metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&store.ctx, &bucket)
.await
.expect("committed metadata");
assert!(persisted && metadata.bucket_creation_committed);
let path = metadata.save_file_path();
match state {
"missing" => delete_config(store.clone(), &path).await.expect("remove metadata authority"),
"corrupt" => save_config(store.clone(), &path, b"corrupt bucket metadata".to_vec())
.await
.expect("corrupt authority"),
"uncommitted" | "mismatched" => {
if state == "uncommitted" {
metadata.bucket_creation_committed = false;
} else {
metadata.bucket_incarnation_id = Uuid::new_v4();
}
metadata
.save_with_store_committed(store.clone())
.await
.expect("inject inconsistent metadata authority");
}
"retired" => {
// A returning disk may retain a directory after retirement was committed.
crate::bucket::retirement::commit_retirement(store.clone(), &bucket, incarnation, &ObjectOptions::default())
.await
.expect("persist retirement evidence");
}
_ => unreachable!("fixed fixture states"),
}
for disk in 1..4 {
tokio::fs::remove_dir(root.path().join(format!("pool0-disk{disk}/{bucket}")))
.await
.expect("lose fixture directories");
}
assert!(
store
.heal_bucket_at_incarnation(&bucket, incarnation, &HealOpts::default())
.await
.is_err(),
"must reject {state}"
);
for disk in 1..4 {
assert!(
!root.path().join(format!("pool0-disk{disk}/{bucket}")).exists(),
"{state} cannot authorize reconstruction"
);
}
}
shutdown.cancel();
}
#[tokio::test]
#[serial_test::serial]
async fn admin_bucket_recovery_rejects_offline_disks_and_fully_missing_sets() {
for offline in [false, true] {
let (root, store, shutdown) = multi_pool_heal_store().await;
let bucket = "recovery-unproven-set";
store
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("fixture bucket");
let incarnation = store.bucket_incarnation_id_from_disk(bucket).await.expect("fixture identity");
let first_missing = usize::from(offline);
for disk in first_missing..4 {
tokio::fs::remove_dir(root.path().join(format!("pool0-disk{disk}/{bucket}")))
.await
.expect("remove fixture directories");
}
if offline {
let disk = store.pools[0].disk_set[0].disks.read().await[0]
.clone()
.expect("online fixture disk");
disk.close().await.expect("stop disk monitoring");
disk.force_offline_for_test();
}
assert!(
store
.heal_bucket_at_incarnation(bucket, incarnation, &HealOpts::default())
.await
.is_err()
);
for disk in first_missing..4 {
assert!(!root.path().join(format!("pool0-disk{disk}/{bucket}")).exists());
}
shutdown.cancel();
}
}
#[tokio::test]
#[serial_test::serial]
async fn admin_bucket_recovery_dry_run_preserves_missing_directories() {
let (root, store, shutdown) = multi_pool_heal_store().await;
let bucket = "recovery-dry-run";
store
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("fixture bucket");
let incarnation = store.bucket_incarnation_id_from_disk(bucket).await.expect("fixture identity");
for disk in 1..4 {
tokio::fs::remove_dir(root.path().join(format!("pool0-disk{disk}/{bucket}")))
.await
.expect("remove fixture directories");
}
store
.heal_bucket_at_incarnation(
bucket,
incarnation,
&HealOpts {
dry_run: true,
..Default::default()
},
)
.await
.expect("dry run can inspect eligible recovery");
for disk in 1..4 {
assert!(
!root.path().join(format!("pool0-disk{disk}/{bucket}")).exists(),
"dry run must not reconstruct directories"
);
}
shutdown.cancel();
}
#[tokio::test]
#[serial_test::serial]
async fn bucket_incarnation_heal_preserves_successor_shards_and_allows_fresh_repair() {
File diff suppressed because it is too large Load Diff
+128 -11
View File
@@ -32,6 +32,42 @@ fn data_movement_multipart_discovery_counts() -> &'static std::sync::Mutex<std::
DATA_MOVEMENT_MULTIPART_DISCOVERY_COUNTS.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()))
}
#[cfg(test)]
type OrphanMultipartCleanupPause = (tokio::sync::oneshot::Sender<()>, tokio::sync::oneshot::Receiver<()>);
#[cfg(test)]
static ORPHAN_MULTIPART_CLEANUP_PAUSES: std::sync::OnceLock<
std::sync::Mutex<std::collections::HashMap<String, OrphanMultipartCleanupPause>>,
> = std::sync::OnceLock::new();
#[cfg(test)]
pub(crate) fn pause_orphan_multipart_cleanup_for_test(
bucket: &str,
) -> (tokio::sync::oneshot::Receiver<()>, tokio::sync::oneshot::Sender<()>) {
let (arrived_tx, arrived_rx) = tokio::sync::oneshot::channel();
let (release_tx, release_rx) = tokio::sync::oneshot::channel();
let previous = ORPHAN_MULTIPART_CLEANUP_PAUSES
.get_or_init(Default::default)
.lock()
.expect("orphan cleanup pause mutex should not poison")
.insert(bucket.to_owned(), (arrived_tx, release_rx));
assert!(previous.is_none(), "each bucket has at most one orphan cleanup pause");
(arrived_rx, release_tx)
}
#[cfg(test)]
async fn pause_orphan_multipart_cleanup(bucket: &str) {
let pause = ORPHAN_MULTIPART_CLEANUP_PAUSES
.get_or_init(Default::default)
.lock()
.expect("orphan cleanup pause mutex should not poison")
.remove(bucket);
if let Some((arrived, release)) = pause {
let _ = arrived.send(());
let _ = release.await;
}
}
fn decommission_multipart_target_clear_pending(opts: &ObjectOptions, target: Option<&ObjectInfo>) -> Result<bool> {
let expected_mod_time = opts.mod_time.ok_or_else(|| Error::DecommissionCapacityBlocked {
message: "multipart cleanup cannot prove exact target absence without a modification time".to_string(),
@@ -654,7 +690,10 @@ impl ECStore {
opts: &ObjectOptions,
) -> Result<PartInfo> {
check_put_object_part_args(bucket, object, upload_id)?;
let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
let (mut opts, bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
if let Some(guard) = bucket_lifecycle_guard {
opts.bucket_lifecycle_lock_owner = Some(Arc::new(guard));
}
if !self.single_pool() {
opts.decommission_capacity_admission = crate::bucket::metadata_sys::object_store_if_initialized_in(&self.ctx).await;
}
@@ -750,19 +789,12 @@ impl ECStore {
opts: &ObjectOptions,
) -> Result<()> {
check_abort_multipart_args(bucket, object, upload_id)?;
let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
let opts = &opts;
// TODO(backlog): defer DeleteUploadID to background for faster abort response
if self.single_pool() {
return self.pools[0].abort_multipart_upload(bucket, object, upload_id, opts).await;
}
let (opts, bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
for pool_idx in self.existing_multipart_pool_order().await {
let pool = &self.pools[pool_idx];
let err = match pool.abort_multipart_upload(bucket, object, upload_id, opts).await {
let err = match pool.abort_multipart_upload(bucket, object, upload_id, &opts).await {
Ok(_) => return Ok(()),
Err(err) => {
//
@@ -775,7 +807,92 @@ impl ECStore {
}
}
Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned()))
self.recover_orphan_multipart_upload(bucket, object, upload_id, opts, bucket_lifecycle_guard)
.await
}
async fn recover_orphan_multipart_upload(
&self,
bucket: &str,
object: &str,
upload_id: &str,
opts: ObjectOptions,
bucket_lifecycle_guard: Option<rustfs_lock::NamespaceLockGuard>,
) -> Result<()> {
let invalid_upload = || StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned());
if self.ctx.lock_manager().is_disabled() {
return Err(invalid_upload());
}
let Some(bucket_lifecycle_guard) = bucket_lifecycle_guard else {
// A signal borrowed from an enclosing operation cannot keep its lock
// alive when the request waiter disappears during physical cleanup.
return Err(invalid_upload());
};
let Some(expected_incarnation) = opts.expected_bucket_incarnation_id.filter(|id| !id.is_nil()) else {
return Err(invalid_upload());
};
// Recovery is a side path of an idempotent Abort. Until the first
// deletion is issued, anything that stops staging absence from being
// proven (an offline disk, lock contention, partially overlapping lock
// domains, unreadable bucket metadata) keeps the upload reported as
// gone, exactly as before recovery existed; only failures inside the
// deletion task are request failures, because a retry can finish them.
let (metadata, persisted) =
match crate::bucket::metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await {
Ok(loaded) => loaded,
Err(error) => {
debug!(bucket, object, upload_id, %error, "orphan multipart cleanup skipped: bucket metadata unavailable");
return Err(invalid_upload());
}
};
if !persisted
|| !metadata.bucket_creation_committed
|| !metadata.bucket_incarnation_sidecar
|| metadata.bucket_incarnation_id != expected_incarnation
{
return Err(invalid_upload());
}
// Lock order: bucket lifecycle, then upload locks in pool-index order.
// Prove metadata absence across every pool before deleting any staging.
let mut cleanup_guards = Vec::with_capacity(self.pools.len());
for pool in &self.pools {
let guard = match pool
.get_disks_by_key(object)
.lock_orphan_multipart_cleanup(bucket, object, upload_id, &opts, &cleanup_guards)
.await
{
Ok(guard) => guard,
Err(error) => {
if !is_err_invalid_upload_id(&error) {
debug!(bucket, object, upload_id, %error, "orphan multipart cleanup skipped: staging absence not proven");
}
return Err(invalid_upload());
}
};
cleanup_guards.push(guard);
}
let bucket = bucket.to_owned();
let object = object.to_owned();
let upload_id = upload_id.to_owned();
tokio::spawn(async move {
// Own both the real bucket guard and all upload guards until every
// disk deletion settles, including after cancellation of the waiter.
let _bucket_lifecycle_guard = bucket_lifecycle_guard;
#[cfg(test)]
pause_orphan_multipart_cleanup(&bucket).await;
let mut recovered = false;
for guard in &cleanup_guards {
recovered |= guard.delete(&bucket, &object, &opts).await?;
}
if recovered {
Ok(())
} else {
Err(StorageError::InvalidUploadID(bucket, object, upload_id))
}
})
.await
.map_err(|error| Error::other_with_context("orphan multipart cleanup task failed", error))?
}
pub(crate) async fn abort_multipart_upload_for_data_movement(
+4 -2
View File
@@ -1697,7 +1697,9 @@ mod tests {
#[serial_test::serial]
async fn target_ordinary_timeout_keeps_its_physical_namespace_owner() {
use crate::disk::os::prepared_publication_test_hooks as hooks;
temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("1"))], async {
// Allow real fixture I/O to finish under load. Expire the operation
// with virtual time only after its physical publication is paused.
temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("60"))], async {
let ctx = Arc::new(InstanceContext::new());
let store = super::super::tests::build_store_with_ctx(ctx.clone());
let root = tempfile::tempdir().expect("root");
@@ -1731,7 +1733,7 @@ mod tests {
.await
.expect("bounded entry");
tokio::time::pause();
tokio::time::advance(std::time::Duration::from_secs(2)).await;
tokio::time::advance(std::time::Duration::from_secs(61)).await;
tokio::time::resume();
let result = tokio::time::timeout(std::time::Duration::from_secs(5), &mut rename)
.await
@@ -1,497 +0,0 @@
// Copyright 2026 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Byte-level goldens for the 13 persisted bucket-configuration XML families
//! (rustfs/backlog#2744).
//!
//! Responsible for: pinning the exact bytes that `deserialize` followed by
//! `serialize` produces for every family in its three source shapes (AWS
//! documentation shape, MinIO-written shape, RustFS-written shape), and the
//! exact bytes RustFS writes for a typed value. Only bytes are compared; a
//! structural comparison would let a codec replacement change the persisted
//! encoding without failing here.
//!
//! Not responsible for: validating configurations, touching the bucket
//! metadata store, or choosing a canonical shape.
//!
//! Upstream: the bucket-config `deserialize`/`serialize` pair, reached through the
//! integration-test facade `tests/storage_api.rs`, and the s3s DTOs they are generic over. Downstream: the DTO/codec
//! replacement (T1.4) and the persisted-bytes comparison (T3.7), which must
//! keep every golden green without regenerating it. Regeneration is reserved
//! to `scripts/gen_bucket_config_goldens.sh`, which sets
//! `BUCKET_CONFIG_GOLDENS_WRITE`: in that mode every test writes its files
//! and then fails on purpose, so a write run never prints a green line.
mod storage_api;
use s3s::dto::{
AbortIncompleteMultipartUpload, AccelerateConfiguration, BucketAccelerateStatus, BucketLifecycleConfiguration,
BucketLoggingStatus, BucketVersioningStatus, CORSConfiguration, CORSRule, Condition, DefaultRetention, DelMarkerExpiration,
DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination,
ErrorDocument, Event, ExcludedPrefix, ExistingObjectReplication, ExistingObjectReplicationStatus, ExpirationStatus,
FilterRule, FilterRuleName, IndexDocument, LifecycleExpiration, LifecycleRule, LifecycleRuleAndOperator, LifecycleRuleFilter,
LoggingEnabled, NoncurrentVersionExpiration, NotificationConfiguration, NotificationConfigurationFilter,
ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRetentionMode, ObjectLockRule, Payer, PublicAccessBlockConfiguration,
QueueConfiguration, Redirect, ReplicaModifications, ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule,
ReplicationRuleAndOperator, ReplicationRuleFilter, ReplicationRuleStatus, RequestPaymentConfiguration, RoutingRule,
S3KeyFilter, ServerSideEncryption, ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration,
ServerSideEncryptionRule, SourceSelectionCriteria, StorageClass, Tag, Tagging, Transition, TransitionStorageClass,
VersioningConfiguration, WebsiteConfiguration,
};
use s3s::xml;
use std::path::{Path, PathBuf};
use storage_api::bucket_config_codec::{deserialize, serialize};
/// Set only by `scripts/gen_bucket_config_goldens.sh`. Every pair test then
/// writes its `.out.xml` (and, for the `rustfs` shape, its `.in.xml`) and
/// fails on purpose, so a write run can never be mistaken for a verification.
const WRITE_ENV: &str = "BUCKET_CONFIG_GOLDENS_WRITE";
fn goldens_dir() -> PathBuf {
Path::new(env!("CARGO_MANIFEST_DIR"))
.join("tests")
.join("fixtures")
.join("bucket-config-goldens")
}
fn pair_path(family: &str, shape: &str, side: &str) -> PathBuf {
goldens_dir().join(family).join(format!("{shape}.{side}.xml"))
}
fn read(path: &Path) -> Vec<u8> {
std::fs::read(path).unwrap_or_else(|err| panic!("read {}: {err}", path.display()))
}
fn write(path: &Path, bytes: &[u8]) {
std::fs::write(path, bytes).unwrap_or_else(|err| panic!("write {}: {err}", path.display()));
}
fn contains(haystack: &[u8], needle: &[u8]) -> bool {
haystack.windows(needle.len()).any(|window| window == needle)
}
fn round_trip<T>(input: &[u8], label: &str) -> Vec<u8>
where
T: for<'xml> xml::Deserialize<'xml> + xml::Serialize,
{
let parsed: T = deserialize(input).unwrap_or_else(|err| panic!("{label}: deserialize failed: {err}"));
serialize(&parsed).unwrap_or_else(|err| panic!("{label}: serialize failed: {err}"))
}
/// Byte comparison with a readable report. Parsed values are never compared.
fn assert_bytes_eq(actual: &[u8], expected: &[u8], context: &str) {
if actual != expected {
panic!(
"{context}\n--- expected ({} bytes) ---\n{}\n--- actual ({} bytes) ---\n{}\n",
expected.len(),
String::from_utf8_lossy(expected),
actual.len(),
String::from_utf8_lossy(actual),
);
}
}
/// Checks one `<family>/<shape>` pair. `typed` is the value RustFS writes for
/// the `rustfs` shape; the `aws` and `minio` shapes pass `None`.
fn check_pair<T>(family: &str, shape: &str, typed: Option<&T>)
where
T: for<'xml> xml::Deserialize<'xml> + xml::Serialize,
{
let in_path = pair_path(family, shape, "in");
let out_path = pair_path(family, shape, "out");
let label = format!("{family}/{shape}");
if std::env::var_os(WRITE_ENV).is_some() {
if let Some(value) = typed {
let written = serialize(value).unwrap_or_else(|err| panic!("{label}: typed value failed to serialize: {err}"));
write(&in_path, &written);
}
let input = read(&in_path);
write(&out_path, &round_trip::<T>(&input, &label));
panic!(
"{WRITE_ENV} is set: wrote {} without verifying anything; rerun without {WRITE_ENV}",
out_path.display()
);
}
let input = read(&in_path);
let expected = read(&out_path);
if let Some(value) = typed {
let written = serialize(value).unwrap_or_else(|err| panic!("{label}: typed value failed to serialize: {err}"));
assert_bytes_eq(
&written,
&input,
&format!(
"{label}: RustFS no longer writes the pinned bytes for the typed value ({})",
in_path.display()
),
);
}
let actual = round_trip::<T>(&input, &label);
assert_bytes_eq(
&actual,
&expected,
&format!("{label}: deserialize->serialize bytes drifted from {}", out_path.display()),
);
let again = round_trip::<T>(&expected, &label);
assert_bytes_eq(
&again,
&expected,
&format!("{label}: the golden is not a fixed point of deserialize->serialize"),
);
}
/// The typed values behind every `rustfs.in.xml`: what RustFS writes today
/// through `serialize` when it persists a configuration it built itself.
/// MinIO-only members (`DelMarkerExpiration`, `ExcludeFolders`,
/// `ExcludedPrefixes`, `DeleteReplication`) are included where RustFS accepts
/// them, so the `rustfs` shape is the one that pins their encoding.
mod rustfs_values {
use super::*;
fn tag(key: &str, value: &str) -> Tag {
Tag {
key: Some(key.to_owned()),
value: Some(value.to_owned()),
}
}
pub(super) fn notification() -> NotificationConfiguration {
NotificationConfiguration {
queue_configurations: Some(vec![QueueConfiguration {
events: vec![
Event::from("s3:ObjectCreated:*".to_owned()),
Event::from("s3:ObjectRemoved:*".to_owned()),
],
filter: Some(NotificationConfigurationFilter {
key: Some(S3KeyFilter {
filter_rules: Some(vec![FilterRule {
name: Some(FilterRuleName::from_static(FilterRuleName::PREFIX)),
value: Some("uploads/".to_owned()),
}]),
}),
}),
id: Some("rustfs-webhook".to_owned()),
queue_arn: "arn:rustfs:sqs::primary:webhook".to_owned(),
}]),
..Default::default()
}
}
pub(super) fn lifecycle() -> BucketLifecycleConfiguration {
BucketLifecycleConfiguration {
rules: vec![
LifecycleRule {
abort_incomplete_multipart_upload: Some(AbortIncompleteMultipartUpload {
days_after_initiation: Some(3),
}),
del_marker_expiration: Some(DelMarkerExpiration { days: Some(7) }),
expiration: Some(LifecycleExpiration {
days: Some(30),
..Default::default()
}),
filter: Some(LifecycleRuleFilter {
prefix: Some("tmp/".to_owned()),
..Default::default()
}),
id: Some("expire-tmp".to_owned()),
noncurrent_version_expiration: Some(NoncurrentVersionExpiration {
newer_noncurrent_versions: Some(2),
noncurrent_days: Some(7),
}),
noncurrent_version_transitions: None,
prefix: None,
status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
transitions: None,
},
LifecycleRule {
abort_incomplete_multipart_upload: None,
del_marker_expiration: None,
expiration: None,
filter: Some(LifecycleRuleFilter {
and: Some(LifecycleRuleAndOperator {
prefix: Some("archive/".to_owned()),
tags: Some(vec![tag("tier", "cold")]),
..Default::default()
}),
..Default::default()
}),
id: Some("tier-cold".to_owned()),
noncurrent_version_expiration: None,
noncurrent_version_transitions: None,
prefix: None,
status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
transitions: Some(vec![Transition {
days: Some(90),
storage_class: Some(TransitionStorageClass::from_static("COLDTIER")),
..Default::default()
}]),
},
],
..Default::default()
}
}
pub(super) fn object_lock() -> ObjectLockConfiguration {
ObjectLockConfiguration {
object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)),
rule: Some(ObjectLockRule {
default_retention: Some(DefaultRetention {
days: Some(30),
mode: Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::GOVERNANCE)),
years: None,
}),
}),
}
}
pub(super) fn versioning() -> VersioningConfiguration {
VersioningConfiguration {
exclude_folders: Some(true),
excluded_prefixes: Some(vec![
ExcludedPrefix {
prefix: Some("tmp/".to_owned()),
},
ExcludedPrefix {
prefix: Some("cache/".to_owned()),
},
]),
mfa_delete: None,
status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
}
}
pub(super) fn encryption() -> ServerSideEncryptionConfiguration {
ServerSideEncryptionConfiguration {
rules: vec![ServerSideEncryptionRule {
apply_server_side_encryption_by_default: Some(ServerSideEncryptionByDefault {
kms_master_key_id: Some("rustfs-bucket-key".to_owned()),
sse_algorithm: ServerSideEncryption::from_static(ServerSideEncryption::AWS_KMS),
}),
blocked_encryption_types: None,
bucket_key_enabled: Some(false),
}],
}
}
pub(super) fn tagging() -> Tagging {
Tagging {
tag_set: vec![tag("team", "storage"), tag("env", "staging")],
}
}
pub(super) fn replication() -> ReplicationConfiguration {
ReplicationConfiguration {
role: String::new(),
rules: vec![ReplicationRule {
delete_marker_replication: Some(DeleteMarkerReplication {
status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
}),
delete_replication: Some(DeleteReplication {
status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
}),
destination: Destination {
bucket: "arn:rustfs:replication::0b6f3c2a-5e6a-4d2b-9c1e-7a8f9d0e1b2c:interop-dr".to_owned(),
storage_class: Some(StorageClass::from_static(StorageClass::STANDARD)),
..Default::default()
},
existing_object_replication: Some(ExistingObjectReplication {
status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED),
}),
filter: Some(ReplicationRuleFilter {
and: Some(ReplicationRuleAndOperator {
prefix: Some("docs/".to_owned()),
tags: Some(vec![tag("replicate", "yes")]),
}),
..Default::default()
}),
id: Some("rustfs-dr".to_owned()),
prefix: None,
priority: Some(1),
source_selection_criteria: Some(SourceSelectionCriteria {
replica_modifications: Some(ReplicaModifications {
status: ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED),
}),
sse_kms_encrypted_objects: None,
}),
status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
}],
}
}
pub(super) fn cors() -> CORSConfiguration {
CORSConfiguration {
cors_rules: vec![CORSRule {
allowed_headers: Some(vec!["Authorization".to_owned(), "Content-Type".to_owned()]),
allowed_methods: vec!["GET".to_owned(), "PUT".to_owned()],
allowed_origins: vec!["https://app.example.internal".to_owned()],
expose_headers: Some(vec!["ETag".to_owned(), "x-amz-request-id".to_owned()]),
id: Some("rustfs-app".to_owned()),
max_age_seconds: Some(3600),
}],
}
}
pub(super) fn logging() -> BucketLoggingStatus {
BucketLoggingStatus {
logging_enabled: Some(LoggingEnabled {
target_bucket: "rustfs-access-logs".to_owned(),
target_grants: None,
target_object_key_format: None,
target_prefix: "buckets/interop/".to_owned(),
}),
}
}
pub(super) fn website() -> WebsiteConfiguration {
WebsiteConfiguration {
error_document: Some(ErrorDocument {
key: "404.html".to_owned(),
}),
index_document: Some(IndexDocument {
suffix: "index.html".to_owned(),
}),
redirect_all_requests_to: None,
routing_rules: Some(vec![RoutingRule {
condition: Some(Condition {
http_error_code_returned_equals: None,
key_prefix_equals: Some("legacy/".to_owned()),
}),
redirect: Redirect {
host_name: None,
http_redirect_code: Some("301".to_owned()),
protocol: None,
replace_key_prefix_with: Some("current/".to_owned()),
replace_key_with: None,
},
}]),
}
}
pub(super) fn accelerate() -> AccelerateConfiguration {
AccelerateConfiguration {
status: Some(BucketAccelerateStatus::from_static(BucketAccelerateStatus::SUSPENDED)),
}
}
pub(super) fn request_payment() -> RequestPaymentConfiguration {
RequestPaymentConfiguration {
payer: Payer::from_static(Payer::BUCKET_OWNER),
}
}
pub(super) fn public_access_block() -> PublicAccessBlockConfiguration {
PublicAccessBlockConfiguration {
block_public_acls: Some(true),
block_public_policy: Some(false),
ignore_public_acls: Some(true),
restrict_public_buckets: Some(false),
}
}
}
/// One test per `<family>/<shape>` pair: 13 families x 3 shapes = 39.
mod bucket_config_goldens {
use super::*;
macro_rules! golden {
($name:ident, $family:literal, $shape:literal, $ty:ty) => {
#[test]
fn $name() {
check_pair::<$ty>($family, $shape, None);
}
};
($name:ident, $family:literal, rustfs = $value:expr) => {
#[test]
fn $name() {
check_pair($family, "rustfs", Some(&$value));
}
};
}
golden!(notification_aws, "notification", "aws", NotificationConfiguration);
golden!(notification_minio, "notification", "minio", NotificationConfiguration);
golden!(notification_rustfs, "notification", rustfs = rustfs_values::notification());
golden!(lifecycle_aws, "lifecycle", "aws", BucketLifecycleConfiguration);
golden!(lifecycle_minio, "lifecycle", "minio", BucketLifecycleConfiguration);
golden!(lifecycle_rustfs, "lifecycle", rustfs = rustfs_values::lifecycle());
golden!(object_lock_aws, "object_lock", "aws", ObjectLockConfiguration);
golden!(object_lock_minio, "object_lock", "minio", ObjectLockConfiguration);
golden!(object_lock_rustfs, "object_lock", rustfs = rustfs_values::object_lock());
golden!(versioning_aws, "versioning", "aws", VersioningConfiguration);
golden!(versioning_minio, "versioning", "minio", VersioningConfiguration);
golden!(versioning_rustfs, "versioning", rustfs = rustfs_values::versioning());
golden!(encryption_aws, "encryption", "aws", ServerSideEncryptionConfiguration);
golden!(encryption_minio, "encryption", "minio", ServerSideEncryptionConfiguration);
golden!(encryption_rustfs, "encryption", rustfs = rustfs_values::encryption());
golden!(tagging_aws, "tagging", "aws", Tagging);
golden!(tagging_minio, "tagging", "minio", Tagging);
golden!(tagging_rustfs, "tagging", rustfs = rustfs_values::tagging());
/// `replication/aws.in.xml` is the unknown-element sample from
/// `crates/replication/src/config.rs`
/// (`s3_xml_parser_discards_unknown_replication_elements_before_validation`).
/// Beyond the byte golden, the pair proves the root-level `FutureTopLevel`
/// element is skipped on read and never re-emitted on write.
#[test]
fn replication_aws() {
check_pair::<ReplicationConfiguration>("replication", "aws", None);
let input = read(&pair_path("replication", "aws", "in"));
let output = read(&pair_path("replication", "aws", "out"));
assert!(
contains(&input, b"<FutureTopLevel>future</FutureTopLevel>"),
"the sample must still carry the unknown element"
);
assert!(
!contains(&output, b"FutureTopLevel"),
"an unknown element must not be re-emitted: {}",
String::from_utf8_lossy(&output)
);
}
golden!(replication_minio, "replication", "minio", ReplicationConfiguration);
golden!(replication_rustfs, "replication", rustfs = rustfs_values::replication());
golden!(cors_aws, "cors", "aws", CORSConfiguration);
golden!(cors_minio, "cors", "minio", CORSConfiguration);
golden!(cors_rustfs, "cors", rustfs = rustfs_values::cors());
golden!(logging_aws, "logging", "aws", BucketLoggingStatus);
golden!(logging_minio, "logging", "minio", BucketLoggingStatus);
golden!(logging_rustfs, "logging", rustfs = rustfs_values::logging());
golden!(website_aws, "website", "aws", WebsiteConfiguration);
golden!(website_minio, "website", "minio", WebsiteConfiguration);
golden!(website_rustfs, "website", rustfs = rustfs_values::website());
golden!(accelerate_aws, "accelerate", "aws", AccelerateConfiguration);
golden!(accelerate_minio, "accelerate", "minio", AccelerateConfiguration);
golden!(accelerate_rustfs, "accelerate", rustfs = rustfs_values::accelerate());
golden!(request_payment_aws, "request_payment", "aws", RequestPaymentConfiguration);
golden!(request_payment_minio, "request_payment", "minio", RequestPaymentConfiguration);
golden!(request_payment_rustfs, "request_payment", rustfs = rustfs_values::request_payment());
golden!(public_access_block_aws, "public_access_block", "aws", PublicAccessBlockConfiguration);
golden!(public_access_block_minio, "public_access_block", "minio", PublicAccessBlockConfiguration);
golden!(
public_access_block_rustfs,
"public_access_block",
rustfs = rustfs_values::public_access_block()
);
}
@@ -1,58 +0,0 @@
# Persisted bucket-configuration XML goldens (rustfs/backlog#2744)
Byte-level goldens for the 13 XML families that `BucketMetadata` persists as
`*_config_xml`. They pin what `rustfs_ecstore::api::bucket::utils::deserialize`
followed by `serialize` emits today, so that the DTO/codec replacement (T1.4)
and the persisted-bytes comparison (T3.7) can be proven byte-identical: an
older binary must keep reading what a newer one writes, and the reverse.
Generated at commit: `6b1554003ebf8f2037ffb7da9c9b906527e758da` (2026-10-07T17:40:56Z)
Checked by `crates/ecstore/tests/bucket_config_goldens.rs`
(`cargo nextest run -p rustfs-ecstore bucket_config_goldens`, 39 tests).
Generated once by `scripts/gen_bucket_config_goldens.sh`, which records the
commit above and refuses to run again unless `--force` is passed. Do not
regenerate after a DTO or codec replacement: at that point a diff here is the
finding, not noise.
## Layout
`<family>/<shape>.in.xml` is an input; `<family>/<shape>.out.xml` is the exact
bytes of `serialize(deserialize(in))` on the generating commit. The test
asserts `bytes == out` (never structural equality), that `out` is a fixed point
of the same round trip, and for the `rustfs` shape that `serialize(typed
value)` still equals `rustfs.in.xml`.
| Shape | Source |
| --- | --- |
| `aws` | Written for this corpus from the public S3 XML schema in the shape of the AWS documentation examples: XML declaration, `xmlns` on the root, indented. No AWS prose is copied. |
| `minio` | The bytes a MinIO server persists. Seven families are carved verbatim out of `../minio/bucket_metadata.blob.hex` (MinIO `RELEASE.2025-07-23T15-54-02Z`); the rest are hand-shaped as compact Go `encoding/xml` output with the S3 namespace on the root, because MinIO does not persist those families. |
| `rustfs` | The bytes RustFS writes today through `serialize` for a typed value (`rustfs_values` in the test). This shape carries the MinIO-only members RustFS accepts (`DelMarkerExpiration`, `ExcludeFolders`, `ExcludedPrefixes`, `DeleteReplication`). |
| Family | Root element | `minio.in.xml` provenance |
| --- | --- | --- |
| `notification` | `NotificationConfiguration` | MinIO fixture |
| `lifecycle` | `LifecycleConfiguration` | MinIO fixture (carries the MinIO `ExpiryUpdatedAt` extension) |
| `object_lock` | `ObjectLockConfiguration` | MinIO fixture |
| `versioning` | `VersioningConfiguration` | MinIO fixture |
| `encryption` | `ServerSideEncryptionConfiguration` | MinIO fixture |
| `tagging` | `Tagging` | MinIO fixture |
| `replication` | `ReplicationConfiguration` | MinIO fixture |
| `cors` | `CORSConfiguration` | hand-shaped |
| `logging` | `BucketLoggingStatus` | hand-shaped |
| `website` | `WebsiteConfiguration` | hand-shaped |
| `accelerate` | `AccelerateConfiguration` | hand-shaped |
| `request_payment` | `RequestPaymentConfiguration` | hand-shaped |
| `public_access_block` | `PublicAccessBlockConfiguration` | hand-shaped |
## Unknown elements
`replication/aws.in.xml` is the unknown-element sample from
`crates/replication/src/config.rs`
(`s3_xml_parser_discards_unknown_replication_elements_before_validation`),
re-indented to two spaces. Its golden proves the root-level `FutureTopLevel`
element is skipped on read and never re-emitted on write, which is the
persisted-configuration exception recorded in the root `AGENTS.md`. On the
generating commit every one of the 13 root elements skips unknown children;
nested members (`Rule`, `Filter`, `Destination`, ...) reject them. A
replacement codec must keep that boundary where it is.
@@ -1,4 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<AccelerateConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<Status>Enabled</Status>
</AccelerateConfiguration>
@@ -1 +0,0 @@
<AccelerateConfiguration><Status>Enabled</Status></AccelerateConfiguration>
@@ -1 +0,0 @@
<AccelerateConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>Suspended</Status></AccelerateConfiguration>
@@ -1 +0,0 @@
<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>
@@ -1 +0,0 @@
<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>
@@ -1 +0,0 @@
<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>
@@ -1,17 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<CORSConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<CORSRule>
<ID>AllowWebApp</ID>
<AllowedOrigin>https://www.example.com</AllowedOrigin>
<AllowedMethod>PUT</AllowedMethod>
<AllowedMethod>POST</AllowedMethod>
<AllowedMethod>DELETE</AllowedMethod>
<AllowedHeader>*</AllowedHeader>
<ExposeHeader>x-amz-server-side-encryption</ExposeHeader>
<MaxAgeSeconds>3000</MaxAgeSeconds>
</CORSRule>
<CORSRule>
<AllowedOrigin>*</AllowedOrigin>
<AllowedMethod>GET</AllowedMethod>
</CORSRule>
</CORSConfiguration>
@@ -1 +0,0 @@
<CORSConfiguration><CORSRule><AllowedHeader>*</AllowedHeader><AllowedMethod>PUT</AllowedMethod><AllowedMethod>POST</AllowedMethod><AllowedMethod>DELETE</AllowedMethod><AllowedOrigin>https://www.example.com</AllowedOrigin><ExposeHeader>x-amz-server-side-encryption</ExposeHeader><ID>AllowWebApp</ID><MaxAgeSeconds>3000</MaxAgeSeconds></CORSRule><CORSRule><AllowedMethod>GET</AllowedMethod><AllowedOrigin>*</AllowedOrigin></CORSRule></CORSConfiguration>
@@ -1 +0,0 @@
<CORSConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><CORSRule><ID>console</ID><AllowedOrigin>https://console.example.internal</AllowedOrigin><AllowedMethod>GET</AllowedMethod><AllowedMethod>HEAD</AllowedMethod><AllowedHeader>Authorization</AllowedHeader><AllowedHeader>Content-Type</AllowedHeader><ExposeHeader>ETag</ExposeHeader><MaxAgeSeconds>600</MaxAgeSeconds></CORSRule></CORSConfiguration>
@@ -1 +0,0 @@
<CORSConfiguration><CORSRule><AllowedHeader>Authorization</AllowedHeader><AllowedHeader>Content-Type</AllowedHeader><AllowedMethod>GET</AllowedMethod><AllowedMethod>HEAD</AllowedMethod><AllowedOrigin>https://console.example.internal</AllowedOrigin><ExposeHeader>ETag</ExposeHeader><ID>console</ID><MaxAgeSeconds>600</MaxAgeSeconds></CORSRule></CORSConfiguration>
@@ -1 +0,0 @@
<CORSConfiguration><CORSRule><AllowedHeader>Authorization</AllowedHeader><AllowedHeader>Content-Type</AllowedHeader><AllowedMethod>GET</AllowedMethod><AllowedMethod>PUT</AllowedMethod><AllowedOrigin>https://app.example.internal</AllowedOrigin><ExposeHeader>ETag</ExposeHeader><ExposeHeader>x-amz-request-id</ExposeHeader><ID>rustfs-app</ID><MaxAgeSeconds>3600</MaxAgeSeconds></CORSRule></CORSConfiguration>
@@ -1 +0,0 @@
<CORSConfiguration><CORSRule><AllowedHeader>Authorization</AllowedHeader><AllowedHeader>Content-Type</AllowedHeader><AllowedMethod>GET</AllowedMethod><AllowedMethod>PUT</AllowedMethod><AllowedOrigin>https://app.example.internal</AllowedOrigin><ExposeHeader>ETag</ExposeHeader><ExposeHeader>x-amz-request-id</ExposeHeader><ID>rustfs-app</ID><MaxAgeSeconds>3600</MaxAgeSeconds></CORSRule></CORSConfiguration>
@@ -1,10 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<ServerSideEncryptionConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<Rule>
<ApplyServerSideEncryptionByDefault>
<SSEAlgorithm>aws:kms</SSEAlgorithm>
<KMSMasterKeyID>arn:aws:kms:us-east-1:123456789012:key/1234abcd-12ab-34cd-56ef-1234567890ab</KMSMasterKeyID>
</ApplyServerSideEncryptionByDefault>
<BucketKeyEnabled>true</BucketKeyEnabled>
</Rule>
</ServerSideEncryptionConfiguration>
@@ -1 +0,0 @@
<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><KMSMasterKeyID>arn:aws:kms:us-east-1:123456789012:key/1234abcd-12ab-34cd-56ef-1234567890ab</KMSMasterKeyID><SSEAlgorithm>aws:kms</SSEAlgorithm></ApplyServerSideEncryptionByDefault><BucketKeyEnabled>true</BucketKeyEnabled></Rule></ServerSideEncryptionConfiguration>
@@ -1 +0,0 @@
<ServerSideEncryptionConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm></ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>
@@ -1 +0,0 @@
<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm></ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>
@@ -1 +0,0 @@
<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><KMSMasterKeyID>rustfs-bucket-key</KMSMasterKeyID><SSEAlgorithm>aws:kms</SSEAlgorithm></ApplyServerSideEncryptionByDefault><BucketKeyEnabled>false</BucketKeyEnabled></Rule></ServerSideEncryptionConfiguration>
@@ -1 +0,0 @@
<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><KMSMasterKeyID>rustfs-bucket-key</KMSMasterKeyID><SSEAlgorithm>aws:kms</SSEAlgorithm></ApplyServerSideEncryptionByDefault><BucketKeyEnabled>false</BucketKeyEnabled></Rule></ServerSideEncryptionConfiguration>
@@ -1,49 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<LifecycleConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<Rule>
<ID>Archive and then delete rule</ID>
<Filter>
<And>
<Prefix>projectdocs/</Prefix>
<Tag>
<Key>department</Key>
<Value>finance</Value>
</Tag>
<ObjectSizeGreaterThan>1024</ObjectSizeGreaterThan>
</And>
</Filter>
<Status>Enabled</Status>
<Transition>
<Days>30</Days>
<StorageClass>STANDARD_IA</StorageClass>
</Transition>
<Transition>
<Days>365</Days>
<StorageClass>GLACIER</StorageClass>
</Transition>
<Expiration>
<Days>3650</Days>
</Expiration>
<NoncurrentVersionTransition>
<NoncurrentDays>90</NoncurrentDays>
<StorageClass>GLACIER</StorageClass>
</NoncurrentVersionTransition>
<NoncurrentVersionExpiration>
<NoncurrentDays>400</NoncurrentDays>
<NewerNoncurrentVersions>3</NewerNoncurrentVersions>
</NoncurrentVersionExpiration>
<AbortIncompleteMultipartUpload>
<DaysAfterInitiation>7</DaysAfterInitiation>
</AbortIncompleteMultipartUpload>
</Rule>
<Rule>
<ID>Expire delete markers</ID>
<Filter>
<Prefix>logs/</Prefix>
</Filter>
<Status>Disabled</Status>
<Expiration>
<ExpiredObjectDeleteMarker>true</ExpiredObjectDeleteMarker>
</Expiration>
</Rule>
</LifecycleConfiguration>
@@ -1 +0,0 @@
<LifecycleConfiguration><Rule><AbortIncompleteMultipartUpload><DaysAfterInitiation>7</DaysAfterInitiation></AbortIncompleteMultipartUpload><Expiration><Days>3650</Days></Expiration><Filter><And><ObjectSizeGreaterThan>1024</ObjectSizeGreaterThan><Prefix>projectdocs/</Prefix><Tag><Key>department</Key><Value>finance</Value></Tag></And></Filter><ID>Archive and then delete rule</ID><NoncurrentVersionExpiration><NewerNoncurrentVersions>3</NewerNoncurrentVersions><NoncurrentDays>400</NoncurrentDays></NoncurrentVersionExpiration><NoncurrentVersionTransition><NoncurrentDays>90</NoncurrentDays><StorageClass>GLACIER</StorageClass></NoncurrentVersionTransition><Status>Enabled</Status><Transition><Days>30</Days><StorageClass>STANDARD_IA</StorageClass></Transition><Transition><Days>365</Days><StorageClass>GLACIER</StorageClass></Transition></Rule><Rule><Expiration><ExpiredObjectDeleteMarker>true</ExpiredObjectDeleteMarker></Expiration><Filter><Prefix>logs/</Prefix></Filter><ID>Expire delete markers</ID><Status>Disabled</Status></Rule></LifecycleConfiguration>
@@ -1 +0,0 @@
<LifecycleConfiguration><Rule><ID>d96i4g89k8h26a95st60</ID><Status>Enabled</Status><Filter><Prefix></Prefix></Filter><Expiration><Days>30</Days></Expiration><NoncurrentVersionExpiration><NoncurrentDays>7</NoncurrentDays></NoncurrentVersionExpiration></Rule><ExpiryUpdatedAt>2026-07-07T15:58:57.337315Z</ExpiryUpdatedAt></LifecycleConfiguration>
@@ -1 +0,0 @@
<LifecycleConfiguration><ExpiryUpdatedAt>2026-07-07T15:58:57.337Z</ExpiryUpdatedAt><Rule><Expiration><Days>30</Days></Expiration><Filter><Prefix></Prefix></Filter><ID>d96i4g89k8h26a95st60</ID><NoncurrentVersionExpiration><NoncurrentDays>7</NoncurrentDays></NoncurrentVersionExpiration><Status>Enabled</Status></Rule></LifecycleConfiguration>
@@ -1 +0,0 @@
<LifecycleConfiguration><Rule><AbortIncompleteMultipartUpload><DaysAfterInitiation>3</DaysAfterInitiation></AbortIncompleteMultipartUpload><DelMarkerExpiration><Days>7</Days></DelMarkerExpiration><Expiration><Days>30</Days></Expiration><Filter><Prefix>tmp/</Prefix></Filter><ID>expire-tmp</ID><NoncurrentVersionExpiration><NewerNoncurrentVersions>2</NewerNoncurrentVersions><NoncurrentDays>7</NoncurrentDays></NoncurrentVersionExpiration><Status>Enabled</Status></Rule><Rule><Filter><And><Prefix>archive/</Prefix><Tag><Key>tier</Key><Value>cold</Value></Tag></And></Filter><ID>tier-cold</ID><Status>Enabled</Status><Transition><Days>90</Days><StorageClass>COLDTIER</StorageClass></Transition></Rule></LifecycleConfiguration>
@@ -1 +0,0 @@
<LifecycleConfiguration><Rule><AbortIncompleteMultipartUpload><DaysAfterInitiation>3</DaysAfterInitiation></AbortIncompleteMultipartUpload><DelMarkerExpiration><Days>7</Days></DelMarkerExpiration><Expiration><Days>30</Days></Expiration><Filter><Prefix>tmp/</Prefix></Filter><ID>expire-tmp</ID><NoncurrentVersionExpiration><NewerNoncurrentVersions>2</NewerNoncurrentVersions><NoncurrentDays>7</NoncurrentDays></NoncurrentVersionExpiration><Status>Enabled</Status></Rule><Rule><Filter><And><Prefix>archive/</Prefix><Tag><Key>tier</Key><Value>cold</Value></Tag></And></Filter><ID>tier-cold</ID><Status>Enabled</Status><Transition><Days>90</Days><StorageClass>COLDTIER</StorageClass></Transition></Rule></LifecycleConfiguration>
@@ -1,20 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<BucketLoggingStatus xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<LoggingEnabled>
<TargetBucket>amzn-s3-demo-logging-bucket</TargetBucket>
<TargetPrefix>logs/</TargetPrefix>
<TargetGrants>
<Grant>
<Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="AmazonCustomerByEmail">
<EmailAddress>user@example.com</EmailAddress>
</Grantee>
<Permission>READ</Permission>
</Grant>
</TargetGrants>
<TargetObjectKeyFormat>
<PartitionedPrefix>
<PartitionDateSource>EventTime</PartitionDateSource>
</PartitionedPrefix>
</TargetObjectKeyFormat>
</LoggingEnabled>
</BucketLoggingStatus>
@@ -1 +0,0 @@
<BucketLoggingStatus><LoggingEnabled><TargetBucket>amzn-s3-demo-logging-bucket</TargetBucket><TargetGrants><Grant><Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="AmazonCustomerByEmail"><EmailAddress>user@example.com</EmailAddress></Grantee><Permission>READ</Permission></Grant></TargetGrants><TargetObjectKeyFormat><PartitionedPrefix><PartitionDateSource>EventTime</PartitionDateSource></PartitionedPrefix></TargetObjectKeyFormat><TargetPrefix>logs/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -1 +0,0 @@
<BucketLoggingStatus xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><LoggingEnabled><TargetBucket>interop-logs</TargetBucket><TargetPrefix>interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -1 +0,0 @@
<BucketLoggingStatus><LoggingEnabled><TargetBucket>interop-logs</TargetBucket><TargetPrefix>interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -1 +0,0 @@
<BucketLoggingStatus><LoggingEnabled><TargetBucket>rustfs-access-logs</TargetBucket><TargetPrefix>buckets/interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -1 +0,0 @@
<BucketLoggingStatus><LoggingEnabled><TargetBucket>rustfs-access-logs</TargetBucket><TargetPrefix>buckets/interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -1,30 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<NotificationConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
<TopicConfiguration>
<Id>ImageUploads</Id>
<Topic>arn:aws:sns:us-east-1:123456789012:s3-events</Topic>
<Event>s3:ObjectCreated:*</Event>
<Filter>
<S3Key>
<FilterRule>
<Name>prefix</Name>
<Value>images/</Value>
</FilterRule>
<FilterRule>
<Name>suffix</Name>
<Value>.jpg</Value>
</FilterRule>
</S3Key>
</Filter>
</TopicConfiguration>
<QueueConfiguration>
<Id>Deletions</Id>
<Queue>arn:aws:sqs:us-east-1:123456789012:s3-deletes</Queue>
<Event>s3:ObjectRemoved:*</Event>
</QueueConfiguration>
<CloudFunctionConfiguration>
<Id>Thumbnails</Id>
<CloudFunction>arn:aws:lambda:us-east-1:123456789012:function:thumbnail</CloudFunction>
<Event>s3:ObjectCreated:Put</Event>
</CloudFunctionConfiguration>
</NotificationConfiguration>
@@ -1 +0,0 @@
<NotificationConfiguration><CloudFunctionConfiguration><Event>s3:ObjectCreated:Put</Event><Id>Thumbnails</Id><CloudFunction>arn:aws:lambda:us-east-1:123456789012:function:thumbnail</CloudFunction></CloudFunctionConfiguration><QueueConfiguration><Event>s3:ObjectRemoved:*</Event><Id>Deletions</Id><Queue>arn:aws:sqs:us-east-1:123456789012:s3-deletes</Queue></QueueConfiguration><TopicConfiguration><Event>s3:ObjectCreated:*</Event><Filter><S3Key><FilterRule><Name>prefix</Name><Value>images/</Value></FilterRule><FilterRule><Name>suffix</Name><Value>.jpg</Value></FilterRule></S3Key></Filter><Id>ImageUploads</Id><Topic>arn:aws:sns:us-east-1:123456789012:s3-events</Topic></TopicConfiguration></NotificationConfiguration>
@@ -1 +0,0 @@
<NotificationConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><QueueConfiguration><Id></Id><Event>s3:ObjectCreated:*</Event><Queue>arn:minio:sqs::primary:webhook</Queue></QueueConfiguration></NotificationConfiguration>
@@ -1 +0,0 @@
<NotificationConfiguration><QueueConfiguration><Event>s3:ObjectCreated:*</Event><Id></Id><Queue>arn:minio:sqs::primary:webhook</Queue></QueueConfiguration></NotificationConfiguration>
@@ -1 +0,0 @@
<NotificationConfiguration><QueueConfiguration><Event>s3:ObjectCreated:*</Event><Event>s3:ObjectRemoved:*</Event><Filter><S3Key><FilterRule><Name>prefix</Name><Value>uploads/</Value></FilterRule></S3Key></Filter><Id>rustfs-webhook</Id><Queue>arn:rustfs:sqs::primary:webhook</Queue></QueueConfiguration></NotificationConfiguration>
@@ -1 +0,0 @@
<NotificationConfiguration><QueueConfiguration><Event>s3:ObjectCreated:*</Event><Event>s3:ObjectRemoved:*</Event><Filter><S3Key><FilterRule><Name>prefix</Name><Value>uploads/</Value></FilterRule></S3Key></Filter><Id>rustfs-webhook</Id><Queue>arn:rustfs:sqs::primary:webhook</Queue></QueueConfiguration></NotificationConfiguration>

Some files were not shown because too many files have changed in this diff Show More