Merge pinned main into gateway integration

Merge main acd26afd8b into
db0828aa1c. Keep the gateway runtime and
recovery assertions, restore the original OIDC assertions in the IAM and
admin tests, and refresh both guardrail baselines from measured output.

The three former OIDC service-only states are no longer representable
after the paired service/query change. Their removal is not counted as
equivalent execution coverage.

All 20 original local source-gate commands completed with child exit 0:
- shellcheck --version && actionlint (exit 0)
- cargo fmt --all --check (exit 0)
- ./scripts/check_unsafe_code_allowances.sh (exit 0)
- ./scripts/check_layer_dependencies.sh (exit 0)
- ./scripts/check_architecture_migration_rules.sh (exit 0)
- ./scripts/check_logging_guardrails.sh (exit 0)
- ./scripts/check_error_other_format_ratchet.sh (exit 0)
- ./scripts/check_no_tokio_io_uring.sh (exit 0)
- ./scripts/check_extension_schema_boundaries.sh (exit 0)
- ./scripts/check_body_cache_whitelist.sh (exit 0)
- ./scripts/check_s3s_footprint.sh (exit 0)
- ./scripts/check_test_count_ratchet.sh (exit 0)
- ./scripts/check_fips_wording.sh (exit 0)
- ./scripts/check_embedded_secrets.sh (exit 0)
- make script-tests (exit 0)
- python3 ./scripts/test_object_generation_protocol_model.py (exit 0)
- python3 ./scripts/check_test_wiring.py (exit 0)
- ./scripts/check_no_planning_docs.sh (exit 0)
- ./scripts/check_uring_lane_lib_only.sh (exit 0)
- python3 ./scripts/ci_gate.py --check-workflow (exit 0)

Earlier runs stopped for a formatting mismatch, the measured S3S footprint
baseline mismatch, and capacity limits, including an interrupted script-test
run. Those original results remain recorded; this was not a first-pass run.
Full remote branch CI has not run yet.

.agents/skills/security-advisory-lessons/references/advisory-patterns.md

@@ -45,11 +45,13 @@ Update this file only when an advisory adds or changes a reusable lesson, affect
 - `GHSA-6r96-hmgc-726c`: request headers collided with lowercase server-derived condition keys such as `userid`, `groups`, `versionid`, and JWT/LDAP claims. Lesson: never let caller-controlled headers append to or replace server-derived policy context; reserve trusted condition keys and keep intentional request-header keys separate.
 - `GHSA-v9cp-qfw9-9pfp`: quantified negated string conditions applied negation after aggregation, transposing `ForAllValues` and `ForAnyValue` semantics. Lesson: push negation into the per-value predicate for quantified operators and test partially overlapping multi-value sets.
 - `GHSA-5w8r-p896-6vq2`: OPA policy mode skipped `ExistingObjectTag/*` loading, so tagged objects looked untagged to external policies. Lesson: external authorization plugins need the same object-tag and request context as built-in policy evaluation before they decide.
+- `GHSA-9wgh-5r68-49p7`: substituted policy-variable values could recreate `${...}` and loop inside a single resolution pass, never reaching the outer iteration limit. Lesson: bound substitutions within each pass and test cycles formed across otherwise literal resolver values; confirm whether attacker-controlled claims can reach those values before assigning remote impact.
 
 ### S3 object actions, copy, multipart, and upload policy validation
 
 - `GHSA-g8w9-qw9q-fghr` and `GHSA-xm99-m3gq-83g8`: a valid presigned `PutObject` accepted extra `x-amz-*` headers omitted from `SignedHeaders`; `x-amz-copy-source` could turn that upload capability into a cross-bucket read performed as the signer. Lesson: a presigned URL is a bounded capability; inspect all received security-sensitive headers and reject unsigned ones before selecting the S3 operation or reaching storage.
 - `GHSA-3ppv-fx5m-m749`: explicit `versionId` reads and copy sources authorized `s3:GetObject` instead of `s3:GetObjectVersion`. Lesson: version-specific object access must select version-specific actions for direct reads, `CopyObject`, and `UploadPartCopy`, with tests proving the backend is not reached on denial.
+- `GHSA-cg33-pxfp-qcj3` and `GHSA-65wq-gjw8-xr9w`: `HeadObject` and object-tagging paths authorized current-object actions while operating on an explicit historical `versionId`. Lesson: select the corresponding version action for metadata and tag reads and writes too; audit every operation that accepts a version ID.
 - `GHSA-gjph-xcgw-4p7w`: `PutObject` and browser POST accepted a caller-supplied `versionId` and replaced an unlocked historical version with only `s3:PutObject`. Lesson: only replication-authorized writes may choose an existing version ID; ordinary uploads must not rewrite version history.
 - `GHSA-x298-9x87-fvjq`: anonymous `ListObjectVersions` fell back to `ListBucket` and returned before public-access-block gates. Lesson: compatibility fallbacks must converge on the same post-authorization checks as direct grants, especially `RestrictPublicBuckets` and anonymous data-plane denies.
 - `GHSA-mx42-j6wv-px98`: `UploadPartCopy` missed source authorization and allowed cross-bucket object exfiltration. Lesson: multipart copy must enforce the same source and destination contract as `CopyObject`.
@@ -110,6 +112,7 @@ Update this file only when an advisory adds or changes a reusable lesson, affect
 
 - `GHSA-j548-9grx-fh4f`: Object Lock enforcement treated unreadable, fabricated, or unparsable bucket metadata as absent configuration and allowed retained objects to be deleted or expired. Lesson: retention must fail closed unless Object Lock absence is authoritative, and every delete, lifecycle, scanner, force-delete, and default-retention path needs the same state distinction.
 - `GHSA-28w4-rp2p-636w` and `GHSA-c7hx-vr97-45rg`: an authorized `DeleteBucket` with `x-rustfs-force-delete: true` removed active COMPLIANCE-retained versions even though ordinary version deletion was denied. Lesson: bucket force-delete must enforce retention across every version, including against root/owner callers.
+- `GHSA-q23w-fp7p-472v`: `PutObjectRetention` policy conditions read `x-amz-object-lock-*` headers while the handler applied mode and date from the XML body. Lesson: authorize the parsed retention that will be stored, or reject mismatched header/body values; compare retention dates as instants so offsets cannot change the applied bound.
 
 ### Serde deserialization and input validation
 
@@ -143,13 +146,13 @@ rg -n "deny_unknown_fields|serde.default|as u32|as usize|as i32" rustfs crates
 - IAM fixes: include import/update/list service-account cases with attacker-controlled parent, claims, access key, secret key, and policy.
 - Copy/upload fixes: include cross-bucket, cross-user, source-denied, destination-denied, copy-source-condition, and multipart completion cases.
 - Presigned upload fixes: include a valid presign with extra unsigned tagging, redirect, storage-class, and cross-bucket copy-source headers; require rejection before source or destination storage access, and verify explicitly signed equivalents still work.
-- Version-action fixes: include historical UUID, explicit current version, `null`, range, partNumber, presigned, STS/session, service-account, anonymous bucket-policy, copy source, and multipart-copy source cases.
+- Version-action fixes: include historical UUID, explicit current version, `null`, range, partNumber, presigned, STS/session, service-account, anonymous bucket-policy, copy source, multipart-copy source, `HeadObject`, and tag read/write cases.
 - Version-write fixes: a `PutObject`-only principal must not replace an existing version through `PutObject` or browser POST `?versionId=`; verify replication-authorized writes and versioned, suspended, and unversioned buckets separately.
-- Policy-condition fixes: include reserved-key header collisions, missing keys, partially overlapping multi-value sets, plugin mode, and built-in policy mode.
+- Policy-condition fixes: include reserved-key header collisions, missing keys, partially overlapping multi-value sets, cyclic variable substitutions, plugin mode, and built-in policy mode.
 - Path fixes: include encoded traversal, absolute path, nested traversal, archive entries with `..`, valid object keys that resemble traversal text but should be rejected, and canonical bucket/prefix boundary checks; test per-key `DeleteObjects` errors and prefix-scoped denials on versioned buckets.
 - Logging fixes: assert redacted output for structs and response bodies that may contain credentials.
 - IAM export fixes: assert exported archives omit plaintext user and service-account secrets unless the format deliberately encrypts or seals them.
 - RPC auth fixes: include captured metadata replay across two concrete methods, stale timestamps, wrong path, wrong method surrogate, wrong secret, and valid same-method calls.
 - Browser/CORS fixes: assert no credentials on reflected/default origins, correct behavior for explicit allowlists, and no same-origin script execution for previewed object content.
 - SSE fixes: inspect stored bytes and verify API metadata, read-back behavior, and on-disk ciphertext together; for ETags, upload equal plaintext under different keys and assert list-only callers cannot derive or correlate the plaintext digest, including multipart objects.
-- Object Lock fixes: include unreadable metadata, fabricated metadata defaults, unparsable config, confirmed absent config, COMPLIANCE/GOVERNANCE retention, lifecycle expiry, scanner sweeps, and force-delete paths; an authorized force-delete must preserve active COMPLIANCE versions.
+- Object Lock fixes: include unreadable metadata, fabricated metadata defaults, unparsable config, confirmed absent config, COMPLIANCE/GOVERNANCE retention, lifecycle expiry, scanner sweeps, and force-delete paths; an authorized force-delete must preserve active COMPLIANCE versions. For `PutObjectRetention`, test conflicting header/body values and offset dates against policy bounds and read-back retention.

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=98dff0961c6eb21d6128bfd2d4a6100a624ddf1fb7db05dba360a5d3d0a310c2
-sha256-linux=890f5925b87e1bf3710c356253f3b2276d901e156df9ccc04ec9315374de6d22
+sha256-darwin=b9438fd1b82558f1fcd7e5a19dc0ceb0b34b5357d4ce99b7033de3ef019333ab
+sha256-linux=d4622517c13283247a022f49962c2cdb69e5ed0e13874fb55e0ab34fb923c044

.config/e2e-nightly-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=efe68675f4676a0d2901242c5815b1f7af5cdec8231f7cccd408f6dacf5c75db
-sha256-linux=003ed6480935a0a95680cc061be143722fbd37bd477b8ff98607b1a064f3aef1
+sha256-darwin=5dbb9154b8feb94cedc8fb4aa6895d4b3259cc31b7df1f5c3b9928ef07f04535
+sha256-linux=d3a210be9eccf593d3723ae669b66680e9751dd01a6949f03df20614f913b707

.config/e2e-smoke-selection.txt

@@ -1 +1 @@
-sha256=ba8f392b617cf9f2338142233a417e43927b28096635e628233ce8a60cf49d10
+sha256=58df0d240cd74b916ce8f094ed1d8ad3e136f2f9bac157a155192a8786dae65c

.config/make/lint-fmt.mak

@@ -70,6 +70,11 @@ 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..."

.config/make/pre-commit.mak

@@ -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 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 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
 	@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 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 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
 	@echo "✅ All pre-PR checks passed!"
 
 .PHONY: dev-check

.config/make/tests.mak

@@ -50,8 +50,11 @@ 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_upgrade_workflow.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_migration_gate_evidence.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/ci_gate.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_security_coverage.py --self-test

.config/nextest.toml

@@ -589,6 +589,14 @@ 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) & (
@@ -596,6 +604,7 @@ 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)$/)
     )

.config/s3s-edge-allowlist.txt

@@ -0,0 +1,37 @@
+# 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/**

.config/s3s-footprint-baseline.txt

@@ -0,0 +1,17 @@
+# 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=1566
+ecstore_files=36

.config/test-count-baseline.txt

@@ -0,0 +1,63 @@
+# 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	892	18
+rustfs	5536	62
+rustfs-audit	70	0
+rustfs-checksums	26	0
+rustfs-common	23	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	6322	1
+rustfs-extension-schema	23	0
+rustfs-filemeta	306	0
+rustfs-heal	839	3
+rustfs-heal-contracts	6	0
+rustfs-iam	377	0
+rustfs-io-core	49	0
+rustfs-io-metrics	125	0
+rustfs-keystone	29	0
+rustfs-kms	715	25
+rustfs-license	2	0
+rustfs-lifecycle	152	0
+rustfs-lock	135	0
+rustfs-log-analyzer	88	0
+rustfs-madmin	209	0
+rustfs-notify	136	0
+rustfs-object-capacity	92	0
+rustfs-object-data-cache	140	0
+rustfs-obs	366	0
+rustfs-policy	680	0
+rustfs-protocols	736	1
+rustfs-protos	52	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	991	20
+rustfs-scanner-metrics	76	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	20	0
+rustfs-trusted-proxies	68	0
+rustfs-utils	217	0
+rustfs-zip	30	0

.github/actions/quick-checks/action.yml

@@ -86,6 +86,10 @@ 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

.github/workflows/cache-warm.yml

@@ -99,7 +99,7 @@ env:
   RUST_BACKTRACE: 1
 
 jobs:
-  # Readers: test-and-lint, test-ilm-integration-serial, build-rustfs-debug-binary,
+  # Readers: test-and-lint (workspace and ILM), build-rustfs-debug-binary,
   # e2e-tests, e2e-full.
   warm-ci-dev:
     name: Warm ci-dev

.github/workflows/ci.yml

@@ -16,7 +16,9 @@ name: Continuous Integration
 
 on:
   push:
-    branches: [ main ]
+    # 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 ]
     paths-ignore:
       - "**.md"
       - "docs/**"
@@ -168,13 +170,10 @@ jobs:
             echo "started_at=$(date --utc --iso-8601=seconds)"
           } > artifacts/test-and-lint/run-metadata.txt
 
-      # Clippy runs before the test pass: lint failures are the most common
-      # CI-only breakage and should surface in minutes, not after 20+ minutes
-      # of tests.
-      # Sampled too: clippy is the natural control arm for any CARGO_BUILD_JOBS
-      # experiment, since --all-targets is check-only for workspace members and
-      # never links the ~100 test binaries the limit exists to throttle.
+      # Keep full Clippy validation on main, merge groups, and scheduled/manual
+      # runs. PRs retain the complete default test suite and compile-free checks.
       - name: Run clippy lints
+        if: github.event_name != 'pull_request'
         run: |
           ./scripts/ci/resource_sampler.sh start clippy
           trap './scripts/ci/resource_sampler.sh stop' EXIT
@@ -283,6 +282,49 @@ jobs:
       - name: Verify rebalance/decommission migration proofs
         run: ./scripts/check_migration_gate_count.sh evidence artifacts/test-and-lint/core-test-listing.json target/nextest/ci/junit.xml
 
+      # The workspace already compiled these ignored tests. Keep the same Cargo
+      # package selection to reuse their binaries, then run them serially in
+      # separate processes because they share fixed ports and global state.
+      # test_transition_and_restore_flows was re-enabled by rustfs/backlog#1303:
+      # its "missing xl.meta on disk2" was a test-util bug (open_disk hardcoded
+      # disk_index 0), not an EC metadata-distribution issue.
+      # restore_object_usecase_reports_ongoing_conflict_and_completion was
+      # re-enabled by backlog#1304 (restore accepts serialize on a short CAS
+      # guard; the copy-back no longer holds the #4877 whole-copy-back lock,
+      # so the mid-restore ongoing read and fast 409 rejection it asserts are
+      # the implemented contract).
+      - name: Run ignored ILM integration tests serially
+        if: needs.classify-changes.outputs.mode == 'full'
+        env:
+          # Match the preceding workspace build and reuse its feature graph.
+          CARGO_BUILD_JOBS: ${{ (github.event_name == 'push' || github.event_name == 'workflow_dispatch') && '3' || '2' }}
+        run: |
+          mkdir -p artifacts/ilm-integration
+          set +e
+          NEXTEST_HIDE_PROGRESS_BAR=1 timeout --verbose --signal=TERM --kill-after=30s 80m \
+            cargo nextest run --profile default -j1 --run-ignored ignored-only \
+            --all --exclude e2e_test \
+            -E 'binary(lifecycle_integration_test) or (package(rustfs) and test(lifecycle_transition_api_test))' \
+            --status-level all --final-status-level all \
+            2>&1 | tee artifacts/ilm-integration/nextest.log
+          status=${PIPESTATUS[0]}
+          {
+            echo "exit_status=${status}"
+            echo "finished_at=$(date --utc --iso-8601=seconds)"
+            echo
+            echo "Remaining test-related processes:"
+            pgrep -af 'cargo|nextest|target/.*/deps/' || true
+          } > artifacts/ilm-integration/diagnostics.txt
+          exit "${status}"
+
+      - name: Upload ILM test diagnostics
+        if: always() && needs.classify-changes.outputs.mode == 'full'
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: ilm-integration-${{ github.run_number }}-${{ github.run_attempt }}
+          path: artifacts/ilm-integration
+          retention-days: 3
+
   # This gate builds into a fresh target directory to isolate the E2E root.
   # Give its cold build a separate budget from workspace tests and migration proofs.
   offline-enrollment-e2e:
@@ -314,83 +356,11 @@ jobs:
       - name: Check offline enrollment E2E root boundary
         run: ./scripts/check_offline_enrollment_e2e.sh
 
-  # Dedicated serial lane for the ILM / lifecycle integration tests. These tests
-  # drive the object layer through process-global singletons (the GLOBAL_ENV
-  # ECStore, the global tier-config manager, background-expiry workers) and bind
-  # fixed ports (9002 for the scanner suite, 9003 for the app suite), so they are
-  # marked #[ignore] to keep them out of the default parallel `cargo nextest run
-  # --all` pass. Run them here explicitly with `--run-ignored ignored-only` and
-  # `-j1` so no two of them share global state or race for a port at once.
-  # serial_test's #[serial] does NOT serialize across nextest's process-per-test
-  # boundary (see .config/nextest.toml); `-j1` is what actually serializes them.
-  # See rustfs/backlog#1148 (ilm-1) and #1155.
-  test-ilm-integration-serial:
-    name: ILM Integration (serial)
-    if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
-    needs: [ quick-checks, classify-changes ]
-    runs-on: sm-standard-4
-    timeout-minutes: 90
-    env:
-      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
-    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'
-
-      # test_transition_and_restore_flows was re-enabled by rustfs/backlog#1303:
-      # its "missing xl.meta on disk2" was a test-util bug (open_disk hardcoded
-      # disk_index 0), not an EC metadata-distribution issue.
-      # restore_object_usecase_reports_ongoing_conflict_and_completion was
-      # re-enabled by backlog#1304 (restore accepts serialize on a short CAS
-      # guard; the copy-back no longer holds the #4877 whole-copy-back lock,
-      # so the mid-restore ongoing read and fast 409 rejection it asserts are
-      # the implemented contract).
-      - name: Run ignored ILM integration tests serially
-        env:
-          # Match the measured Test and Lint link budget. The default exposed
-          # all 14 pod CPUs and a cold cache spent the full 80m compiling
-          # without starting one ILM test (main run 32982910990).
-          CARGO_BUILD_JOBS: ${{ (github.event_name == 'push' || github.event_name == 'workflow_dispatch') && '3' || '2' }}
-        run: |
-          mkdir -p artifacts/ilm-integration
-          set +e
-          NEXTEST_HIDE_PROGRESS_BAR=1 timeout --verbose --signal=TERM --kill-after=30s 80m \
-            cargo nextest run -j1 --run-ignored ignored-only \
-            -p rustfs-scanner -p rustfs \
-            -E 'binary(lifecycle_integration_test) or (package(rustfs) and test(lifecycle_transition_api_test))' \
-            --status-level all --final-status-level all \
-            2>&1 | tee artifacts/ilm-integration/nextest.log
-          status=${PIPESTATUS[0]}
-          {
-            echo "exit_status=${status}"
-            echo "finished_at=$(date --utc --iso-8601=seconds)"
-            echo
-            echo "Remaining test-related processes:"
-            pgrep -af 'cargo|nextest|target/.*/deps/' || true
-          } > artifacts/ilm-integration/diagnostics.txt
-          exit "${status}"
-
-      - name: Upload ILM test diagnostics
-        if: always()
-        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
-        with:
-          name: ilm-integration-${{ github.run_number }}-${{ github.run_attempt }}
-          path: |
-            artifacts/ilm-integration
-            target/nextest/ci/junit.xml
-
   test-and-lint-rio-v2:
     name: Test and Lint (rio-v2)
-    if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
+    # Alternate feature configurations run on main, merge groups, and scheduled
+    # or manual validation; the default workspace remains the PR gate.
+    if: needs.classify-changes.outputs.mode == 'full' && github.event_name != 'pull_request'
     needs: [ quick-checks, classify-changes ]
     runs-on: sm-standard-4
     timeout-minutes: 90
@@ -506,7 +476,7 @@ jobs:
 
   test-and-lint-protocols:
     name: "Test and Lint (${{ matrix.features.name }})"
-    if: needs.classify-changes.outputs.mode == 'full' && (github.event_name != 'pull_request' || github.event.action != 'closed')
+    if: needs.classify-changes.outputs.mode == 'full' && github.event_name != 'pull_request'
     needs: [ quick-checks, classify-changes ]
     runs-on: sm-standard-4
     timeout-minutes: 90
@@ -780,6 +750,8 @@ 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
@@ -898,18 +870,105 @@ 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 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 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).
     if: >-
       github.event_name == 'workflow_dispatch' ||
       github.event_name == 'merge_group' ||
-      (github.event_name == 'push' && github.ref == 'refs/heads/main')
+      (github.event_name == 'push' && (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/gateway/integration'))
     needs: [ build-rustfs-debug-binary ]
     runs-on: sm-standard-2
     timeout-minutes: 55
@@ -1264,7 +1323,6 @@ jobs:
       - quick-checks
       - test-and-lint
       - offline-enrollment-e2e
-      - test-ilm-integration-serial
       - test-and-lint-rio-v2
       - connect-short-credential-boundary
       - test-and-lint-protocols
@@ -1298,7 +1356,6 @@ jobs:
       - quick-checks
       - test-and-lint
       - offline-enrollment-e2e
-      - test-ilm-integration-serial
       - test-and-lint-rio-v2
       - test-and-lint-protocols
       - build-rustfs-debug-binary

.github/workflows/e2e-upgrade.yml

@@ -96,38 +96,8 @@ jobs:
           overwrite: true
 
   upgrade:
-    name: ${{ matrix.name }}
+    name: Upgrade compatibility
     needs: build
-    strategy:
-      fail-fast: false
-      matrix:
-        include:
-          # The two `_from_rc2_` tests keep their names: they assert
-          # release-independent object contracts and pass unchanged against the
-          # newer pinned source, so renaming them would only churn history and
-          # the CI required-check names. UPGRADE_SOURCE_VERSION above is the
-          # single source of truth for which release they actually run against.
-          - name: Direct upgrade from the previous release
-            test: direct_upgrade_from_rc2_preserves_object_contracts
-            artifact: direct-upgrade
-          - name: Mixed-version rolling upgrade from the previous release
-            test: rolling_upgrade_from_rc2_preserves_mixed_version_contracts
-            artifact: mixed-version-upgrade
-          - name: Bucket configuration survives the upgrade
-            test: direct_upgrade_from_previous_release_preserves_bucket_configuration
-            artifact: bucket-config-upgrade
-          - name: Rollback reads current bucket metadata
-            test: rollback_to_previous_release_reads_current_bucket_metadata
-            artifact: bucket-config-rollback
-          - name: ODM configuration recovery after rc.5 rollback
-            test: rc5_rollback_requires_restoring_odm_configuration
-            artifact: odm-config-rollback
-          - name: Multipart layouts survive the rc.5 upgrade
-            test: direct_upgrade_from_rc5_preserves_multipart_layouts
-            artifact: multipart-layout-upgrade
-          - name: rc.5 multipart replication baseline
-            test: rc5_baseline_replicates_multipart_layouts
-            artifact: multipart-layout-baseline
     runs-on: sm-standard-2
     timeout-minutes: 60
     env:
@@ -142,8 +112,7 @@ jobs:
         uses: ./.github/actions/setup
         with:
           cache-shared-key: e2e-upgrade-tests
-          # One main-branch consumer warms the shared test-harness dependencies.
-          cache-save-if: ${{ github.ref == 'refs/heads/main' && matrix.artifact == 'direct-upgrade' }}
+          cache-save-if: ${{ github.ref == 'refs/heads/main' }}
           install-build-packaging-tools: "false"
           install-test-tools: "false"
 
@@ -170,22 +139,49 @@ jobs:
           echo "RUSTFS_UPGRADE_SOURCE_BINARY=$SOURCE_DIR/rustfs" >> "$GITHUB_ENV"
           echo "RUSTFS_E2E_LOG_DIR=$RUNNER_TEMP/rustfs-upgrade-logs" >> "$GITHUB_ENV"
 
-      - name: Run upgrade compatibility test
+      # Reuse one compiled harness, while keeping each case in its own process.
+      # Continue after a failure to retain the former matrix's complete verdict.
+      - name: Run upgrade compatibility tests
+        shell: bash
         env:
-          RUSTFS_SCANNER_HEAL_G09_EVIDENCE_DIR: ${{ runner.temp }}/rustfs-upgrade-g09-evidence/${{ matrix.artifact }}
           NO_PROXY: 127.0.0.1,localhost,::1
         run: |
           export no_proxy="$NO_PROXY"
-          python3 scripts/e2e_binary.py run -- cargo test --locked -p e2e_test \
-            "upgrade_compatibility_test::${{ matrix.test }}" \
-            -- --ignored --exact --nocapture
+          cases=(
+            direct-upgrade:direct_upgrade_from_rc2_preserves_object_contracts
+            mixed-version-upgrade:rolling_upgrade_from_rc2_preserves_mixed_version_contracts
+            bucket-config-upgrade:direct_upgrade_from_previous_release_preserves_bucket_configuration
+            bucket-config-rollback:rollback_to_previous_release_reads_current_bucket_metadata
+            odm-config-rollback:rc5_rollback_requires_restoring_odm_configuration
+            multipart-layout-upgrade:direct_upgrade_from_rc5_preserves_multipart_layouts
+            multipart-layout-baseline:rc5_baseline_replicates_multipart_layouts
+          )
+          status=0
+          for entry in "${cases[@]}"; do
+            artifact="${entry%%:*}"
+            test="${entry#*:}"
+            export RUSTFS_SCANNER_HEAL_G09_EVIDENCE_DIR="$RUNNER_TEMP/rustfs-upgrade-g09-evidence/$artifact"
+            export RUSTFS_E2E_LOG_DIR="$RUNNER_TEMP/rustfs-upgrade-logs/$artifact"
+            mkdir -p "$RUSTFS_E2E_LOG_DIR"
+            echo "::group::$test"
+            if python3 scripts/e2e_binary.py run -- cargo test --locked -p e2e_test \
+              "upgrade_compatibility_test::$test" -- --ignored --exact --nocapture \
+              2>&1 | tee "$RUSTFS_E2E_LOG_DIR/test.log"; then
+              echo "PASS: $test"
+            else
+              echo "::error::Upgrade compatibility failed: $test"
+              status=1
+            fi
+            echo "::endgroup::"
+          done
+          exit "$status"
 
       - name: Upload scanner/heal G09 evidence
         if: always()
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
         with:
-          name: ${{ matrix.artifact }}-g09-evidence-${{ github.run_number }}
-          path: ${{ runner.temp }}/rustfs-upgrade-g09-evidence/${{ matrix.artifact }}
+          name: upgrade-g09-evidence-${{ github.run_number }}
+          path: ${{ runner.temp }}/rustfs-upgrade-g09-evidence
           if-no-files-found: ignore
           retention-days: 14
 
@@ -193,7 +189,7 @@ jobs:
         if: always()
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
         with:
-          name: ${{ matrix.artifact }}-server-logs-${{ github.run_number }}
+          name: upgrade-server-logs-${{ github.run_number }}
           path: ${{ runner.temp }}/rustfs-upgrade-logs
           if-no-files-found: warn
           retention-days: 14

.github/workflows/oidc-keycloak.yml

@@ -20,7 +20,7 @@ on:
       - ".github/workflows/oidc-keycloak.yml"
       - "crates/config/src/constants/oidc.rs"
       - "crates/iam/src/federation/**"
-      - "crates/iam/src/oidc.rs"
+      - "crates/iam/src/oidc/**"
       - "rustfs/src/admin/handlers/oidc.rs"
       - "rustfs/src/admin/handlers/sts.rs"
       - "scripts/test/oidc_keycloak_live.sh"
@@ -31,7 +31,7 @@ on:
       - ".github/workflows/oidc-keycloak.yml"
       - "crates/config/src/constants/oidc.rs"
       - "crates/iam/src/federation/**"
-      - "crates/iam/src/oidc.rs"
+      - "crates/iam/src/oidc/**"
       - "rustfs/src/admin/handlers/oidc.rs"
       - "rustfs/src/admin/handlers/sts.rs"
       - "scripts/test/oidc_keycloak_live.sh"

.vscode/launch.json

@@ -7,7 +7,9 @@
             "name": "Debug(only) executable 'rustfs'",
             "env": {
               "RUST_LOG": "rustfs=info,ecstore=info,s3s=info,iam=info",
-              "RUSTFS_SKIP_BACKGROUND_TASK": "on"
+              "RUSTFS_SKIP_BACKGROUND_TASK": "on",
+              // Local test volumes share one physical disk.
+              "RUSTFS_UNSAFE_BYPASS_DISK_CHECK": "true"
               //"RUSTFS_OBS_LOG_DIRECTORY": "./deploy/logs",
               // "RUSTFS_POLICY_PLUGIN_URL":"http://localhost:8181/v1/data/rustfs/authz/allow",
               // "RUSTFS_POLICY_PLUGIN_AUTH_TOKEN":"your-opa-token"
@@ -44,6 +46,7 @@
             "env": {
                 "RUST_LOG": "rustfs=debug,ecstore=info,s3s=debug,iam=debug",
                 "RUSTFS_SKIP_BACKGROUND_TASK": "on",
+                "RUSTFS_UNSAFE_BYPASS_DISK_CHECK": "true",
                 //"RUSTFS_OBS_LOG_DIRECTORY": "./deploy/logs",
                 // "RUSTFS_POLICY_PLUGIN_URL":"http://localhost:8181/v1/data/rustfs/authz/allow",
                 // "RUSTFS_POLICY_PLUGIN_AUTH_TOKEN":"your-opa-token" 
@@ -117,7 +120,8 @@
             "env": {
                 "RUSTFS_ACCESS_KEY": "rustfsadmin",
                 "RUSTFS_SECRET_KEY": "rustfsadmin",
-                //"RUSTFS_VOLUMES": "./target/volume/test{1...4}",
+                "RUSTFS_VOLUMES": "./target/volume/test{1...4}",
+                "RUSTFS_UNSAFE_BYPASS_DISK_CHECK": "true",
                 "RUSTFS_ADDRESS": ":9000",
                 "RUSTFS_CONSOLE_ENABLE": "true",
                 // "RUSTFS_OBS_TRACE_ENDPOINT": "http://127.0.0.1:4318/v1/traces", // jeager otlp http endpoint
@@ -135,41 +139,30 @@
         {
             "type": "lldb",
             "request": "launch",
-            "name": "Debug test_lifecycle_transition_basic",
+            "name": "Debug test_transition_and_restore_flows",
             "cargo": {
                 "args": [
                     "test",
+                    "--no-run",
                     "-p",
                     "rustfs-scanner",
                     "--test",
                     "lifecycle_integration_test",
-                    "serial_tests::test_lifecycle_transition_basic",
                     "-j",
                     "1"
-                ]
+                ],
+                "filter": {
+                    "name": "lifecycle_integration_test",
+                    "kind": "test"
+                }
             },
-            "args": [],
-            "cwd": "${workspaceFolder}"
-        },
-        {
-            "name": "Debug executable target/debug/test",
-            "type":  "lldb",
-            "request": "launch",
-            "program": "${workspaceFolder}/target/debug/deps/lifecycle_integration_test-5915cbfcab491b3b",
             "args": [
-              "--skip",
-              "test_lifecycle_expiry_basic",
-              "--skip",
-              "test_lifecycle_expiry_deletemarker",
-              //"--skip",
-              //"test_lifecycle_transition_basic",
-            ],
-            "cwd": "${workspaceFolder}",
-            //"stopAtEntry": false,
-            //"preLaunchTask": "cargo build",
-            "sourceLanguages": [
-                "rust"
+                "serial_tests::test_transition_and_restore_flows",
+                "--exact",
+                "--ignored",
+                "--test-threads=1"
             ],
+            "cwd": "${workspaceFolder}"
         },
         {
             "name": "Debug executable target/debug/rustfs with sse kms",
@@ -285,7 +278,7 @@
                 "RUSTFS_COMPRESSION_ENABLED": "true",
                 "RUSTFS_KMS_ENABLE": "true",
                 "RUSTFS_KMS_BACKEND": "local",
-                "RUSTFS_KMS_KEY_DIR": "./target/rio-v2/kms-key-dir",
+                "RUSTFS_KMS_KEY_DIR": "${workspaceFolder}/target/rio-v2/kms-key-dir",
                 "RUSTFS_KMS_LOCAL_MASTER_KEY": "my-secret-key", // Some Password
                 "RUSTFS_KMS_DEFAULT_KEY_ID": "rustfs-master-key",
             },
@@ -307,6 +300,7 @@
                 "RUSTFS_ACCESS_KEY": "rustfsadmin",
                 "RUSTFS_SECRET_KEY": "rustfsadmin",
                 "RUSTFS_VOLUMES": "./target/e2e-test/test{1...4}",
+                "RUSTFS_UNSAFE_BYPASS_DISK_CHECK": "true",
                 "RUSTFS_REGION": "us-east-1",
                 "RUSTFS_ADDRESS": ":9000",
                 "RUSTFS_CONSOLE_ENABLE": "true",
@@ -315,7 +309,7 @@
 
                 "RUSTFS_KMS_ENABLE": "true",
                 "RUSTFS_KMS_BACKEND": "local",
-                "RUSTFS_KMS_KEY_DIR": "./target/e2e-key-dir",
+                "RUSTFS_KMS_KEY_DIR": "${workspaceFolder}/target/e2e-key-dir",
                 "RUSTFS_KMS_LOCAL_MASTER_KEY": "my-secret-key", // Some Password
                 "RUSTFS_KMS_DEFAULT_KEY_ID": "rustfs-master-key",
             },

ARCHITECTURE.md

@@ -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`
-     (`S3S_ECSTORE_FILES_BASELINE`; the `object_lock` module was converted to
+     (the `ecstore_files` line of `.config/s3s-footprint-baseline.txt`; 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`.
 

Cargo.lock

@@ -10045,6 +10045,7 @@ dependencies = [
  "tokio",
  "tokio-util",
  "tracing",
+ "tracing-subscriber",
  "url",
 ]
 
@@ -10284,7 +10285,6 @@ dependencies = [
  "reqwest",
  "rmp-serde",
  "rustfs-signer",
- "s3s",
  "serde",
  "serde_json",
  "sysinfo",
@@ -10509,6 +10509,7 @@ dependencies = [
  "rustfs-keystone",
  "rustfs-policy",
  "rustfs-rio",
+ "rustfs-s3-types",
  "rustfs-storage-api",
  "rustfs-test-utils",
  "rustfs-tls-runtime",
@@ -10523,6 +10524,7 @@ dependencies = [
  "sha2 0.11.0",
  "socket2",
  "subtle",
+ "sync_wrapper",
  "tempfile",
  "thiserror 2.0.21",
  "time",
@@ -10670,6 +10672,7 @@ dependencies = [
  "rustfs-checksums",
  "rustfs-config",
  "rustfs-rio",
+ "rustfs-s3-types",
  "rustfs-signer",
  "rustfs-storage-api",
  "rustfs-tls-runtime",
@@ -10703,9 +10706,19 @@ dependencies = [
 name = "rustfs-s3-types"
 version = "1.0.1"
 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]]
@@ -10729,6 +10742,7 @@ dependencies = [
  "parking_lot",
  "rustfs-common",
  "rustfs-ecstore",
+ "rustfs-s3-types",
  "rustfs-storage-api",
  "rustfs-test-utils",
  "s3s",
@@ -10842,9 +10856,10 @@ dependencies = [
  "bytes",
  "hotpath",
  "http 1.5.0",
+ "http-body",
  "hyper",
+ "rustfs-s3-types",
  "rustfs-utils",
- "s3s",
  "serde_urlencoded",
  "thiserror 2.0.21",
  "time",
@@ -10902,7 +10917,6 @@ dependencies = [
  "rustfs-utils",
  "rustls",
  "rustls-native-certs",
- "s3s",
  "serde",
  "serde_json",
  "sha2 0.11.0",

Cargo.toml

@@ -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 type definitions
+    "crates/s3-types", # S3 event and error type definitions
     "crates/s3-ops", # S3 operation definitions and mapping
     "crates/s3select-api", # S3 Select API interface
     "crates/s3select-query", # S3 Select query engine
@@ -323,6 +323,10 @@ rust-embed = { version = "8.12.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"
@@ -331,6 +335,7 @@ 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"

crates/e2e_test/src/common.rs

@@ -1542,6 +1542,15 @@ impl RustFSTestClusterEnvironment {
         self.start_node_from_binary(node_idx, &binary_path).await
     }
 
+    /// Start a process and wait only for TCP, allowing tests to bring up a
+    /// quorum one node at a time before the S3 service can become ready.
+    pub async fn start_node_process(&mut self, node_idx: usize) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        let binary_path = rustfs_binary_path();
+        let volumes_arg = self.build_volumes_arg();
+        self.spawn_node(node_idx, &binary_path, &volumes_arg)?;
+        self.wait_for_node_ready(&self.nodes[node_idx].address, node_idx).await
+    }
+
     /// Start one stopped cluster node with a specific RustFS binary while
     /// preserving the cluster's volume layout and that node's data directory.
     pub async fn start_node_from_binary(

crates/e2e_test/src/dual_stack_differential/compare.rs

@@ -0,0 +1,512 @@
+// 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",
+            }])
+        );
+    }
+}

crates/e2e_test/src/dual_stack_differential/mod.rs

@@ -0,0 +1,197 @@
+// 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)
+        );
+    }
+}

crates/e2e_test/src/lib.rs

@@ -31,6 +31,12 @@ 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.

crates/e2e_test/src/namespace_lock_quorum_test.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::common::RustFSTestClusterEnvironment;
+use crate::common::{ClusterTopology, RustFSTestClusterEnvironment};
 use aws_sdk_s3::Client;
 use aws_sdk_s3::error::SdkError;
 use bytes::Bytes;
@@ -185,7 +185,11 @@ async fn wait_for_quorum_read_admission(clients: &[Client], bucket: &str) -> Tes
             return Ok(());
         }
         if tokio::time::Instant::now() >= deadline {
-            return Err(format!("read quorum did not become available after lease convergence for {bucket}").into());
+            let diagnostics = clients[0].get_object().bucket(bucket).key("warm-small").send().await;
+            return Err(format!(
+                "read quorum did not become available after lease convergence for {bucket}: GET={diagnostics:?}"
+            )
+            .into());
         }
         tokio::time::sleep(std::time::Duration::from_millis(100)).await;
     }
@@ -195,11 +199,14 @@ async fn wait_for_quorum_read_admission(clients: &[Client], bucket: &str) -> Tes
 async fn test_degraded_cluster_read_quorum_follows_erasure_layout() -> TestResult {
     crate::common::init_logging();
 
-    for (node_count, parity) in [(4, 2), (6, 3), (6, 2)] {
+    for (node_count, parity) in [(3, 1), (4, 2), (6, 3), (6, 2), (4, 0)] {
         let read_quorum = node_count - parity;
         let write_quorum = read_quorum + usize::from(read_quorum == parity);
         let mut cluster = RustFSTestClusterEnvironment::new(node_count).await?;
         cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", format!("EC:{parity}"));
+        if parity == 0 {
+            cluster.set_env("RUSTFS_STORAGE_CLASS_RRS", "EC:0");
+        }
         // Wait for every seed fanout before removing any physical shard.
         cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false");
         cluster.set_env("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "false");
@@ -338,6 +345,277 @@ async fn test_degraded_cluster_read_quorum_follows_erasure_layout() -> TestResul
     Ok(())
 }
 
+#[tokio::test]
+async fn test_cold_start_read_quorum_follows_erasure_layout() -> TestResult {
+    crate::common::init_logging();
+    for (node_count, parity) in [(3, 1), (4, 2), (6, 3), (6, 2), (4, 0)] {
+        let read_quorum = node_count - parity;
+        let write_quorum = read_quorum + usize::from(read_quorum == parity);
+        let mut cluster = RustFSTestClusterEnvironment::new(node_count).await?;
+        cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", format!("EC:{parity}"));
+        if parity == 0 {
+            cluster.set_env("RUSTFS_STORAGE_CLASS_RRS", "EC:0");
+        }
+        cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false");
+        cluster.set_env("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "false");
+        cluster.set_env("RUST_LOG", "warn");
+        cluster.start().await?;
+        let clients = cluster
+            .create_all_clients()?
+            .into_iter()
+            .map(|client| {
+                Client::from_conf(
+                    client
+                        .config()
+                        .to_builder()
+                        .retry_config(aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(1))
+                        .build(),
+                )
+            })
+            .collect::<Vec<_>>();
+        let bucket = format!("cold-read-quorum-{node_count}-{parity}");
+        clients[0].create_bucket().bucket(&bucket).send().await?;
+        let small = b"cold startup follows the stored EC layout".to_vec();
+        let large = vec![73; 256 * 1024];
+        for (key, body) in [
+            ("warm-small", &small),
+            ("warm-large", &large),
+            ("cold-large", &large),
+            ("below-quorum", &large),
+        ] {
+            clients[0]
+                .put_object()
+                .bucket(&bucket)
+                .key(key)
+                .body(Bytes::copy_from_slice(body).into())
+                .send()
+                .await?;
+            for node in &cluster.nodes {
+                let census =
+                    crate::chaos::census_object_version_on_disk(std::path::Path::new(&node.data_dir), &bucket, key, None)?;
+                assert!(census.is_complete(), "seed shard must exist before cold startup: {census:?}");
+                assert_eq!((census.data_blocks, census.parity_blocks), (Some(read_quorum), Some(parity)));
+            }
+        }
+        for node in 0..node_count {
+            cluster.stop_node(node)?;
+        }
+        // Rebuild every process's in-memory state below read quorum, then
+        // cross the EC boundary. A format bootstrap must not admit S3 reads
+        // merely because half the peers are online (e.g. six disks, EC:2).
+        for node in 0..read_quorum - 1 {
+            cluster.start_node_process(node).await?;
+        }
+        let error = clients[0]
+            .get_object()
+            .bucket(&bucket)
+            .key("below-quorum")
+            .send()
+            .await
+            .expect_err("cold startup below the configured EC read quorum must stay unavailable");
+        assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(503));
+        cluster.start_node_process(read_quorum - 1).await?;
+        wait_for_quorum_read_admission(&clients[..read_quorum], &bucket).await?;
+        for client in clients.iter().take(read_quorum) {
+            assert_quorum_object_body(client, &bucket, "warm-small", &small).await?;
+            assert_quorum_object_body(client, &bucket, "cold-large", &large).await?;
+            let write = client
+                .put_object()
+                .bucket(&bucket)
+                .key("cold-quorum-write")
+                .body(Bytes::copy_from_slice(&small).into())
+                .send()
+                .await;
+            if read_quorum >= write_quorum {
+                write?;
+                assert_quorum_object_body(client, &bucket, "cold-quorum-write", &small).await?;
+            } else {
+                let error = write.expect_err("cold startup at read quorum must preserve the EC write quorum");
+                assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(503));
+            }
+        }
+        for node in read_quorum..node_count {
+            cluster.start_node(node).await?;
+        }
+        for client in &clients {
+            assert_quorum_object_body(client, &bucket, "cold-large", &large).await?;
+        }
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_four_by_six_cold_start_at_read_quorum() -> TestResult {
+    crate::common::init_logging();
+    let mut cluster = RustFSTestClusterEnvironment::with_topology(ClusterTopology::single_pool_multidrive(4, 6)).await?;
+    // Explicit URLs are consumed in order. Place one drive from each node in
+    // every set, matching a four-host, six-drive-per-host deployment.
+    let volumes = (0..6)
+        .flat_map(|drive| {
+            cluster
+                .nodes
+                .iter()
+                .map(move |node| format!("http://{}{}", node.address, node.data_dirs[drive]))
+        })
+        .collect::<Vec<_>>()
+        .join(" ");
+    cluster.set_env("RUSTFS_VOLUMES", volumes);
+    cluster.set_env("RUSTFS_ERASURE_SET_DRIVE_COUNT", "4");
+    cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", "EC:2");
+    cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false");
+    cluster.set_env("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "false");
+    cluster.set_env("RUST_LOG", "error");
+    cluster.start().await?;
+    let clients = cluster
+        .create_all_clients()?
+        .into_iter()
+        .map(|client| {
+            Client::from_conf(
+                client
+                    .config()
+                    .to_builder()
+                    .retry_config(aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(1))
+                    .build(),
+            )
+        })
+        .collect::<Vec<_>>();
+    let bucket = "cold-start-read-quorum";
+    clients[0].create_bucket().bucket(bucket).send().await?;
+    let mut fixtures = Vec::new();
+    let mut covered = [[false; 2]; 6];
+    for index in 0..128 {
+        let key = match index {
+            0 => "warm-small".to_string(),
+            1 => "warm-large".to_string(),
+            _ => format!("cold-{index}"),
+        };
+        let size_class = index % 2;
+        let size = if size_class == 0 { 71 } else { 256 * 1024 };
+        let body = (0..size)
+            .map(|offset| u8::try_from(offset % 251).expect("bounded fixture byte"))
+            .collect::<Vec<_>>();
+        clients[0]
+            .put_object()
+            .bucket(bucket)
+            .key(&key)
+            .body(Bytes::copy_from_slice(&body).into())
+            .send()
+            .await?;
+        for node in &cluster.nodes {
+            let mut complete_shards = 0;
+            for (drive, path) in node.data_dirs.iter().enumerate() {
+                let census = crate::chaos::census_object_version_on_disk(std::path::Path::new(path), bucket, &key, None)?;
+                if census.has_xl_meta {
+                    assert!(census.is_complete(), "seed shard must be complete before shutdown: {census:?}");
+                    assert_eq!((census.data_blocks, census.parity_blocks), (Some(2), Some(2)));
+                    covered[drive][size_class] = true;
+                    complete_shards += 1;
+                }
+            }
+            assert_eq!(complete_shards, 1, "each node must own one shard of every seed object");
+        }
+        fixtures.push((key, body));
+        if covered.iter().flatten().all(|covered| *covered) {
+            break;
+        }
+    }
+    assert!(
+        covered.iter().flatten().all(|covered| *covered),
+        "fixtures must cover all six sets with inline and large objects"
+    );
+
+    for first in 0..4 {
+        for second in first + 1..4 {
+            for node in 0..4 {
+                cluster.stop_node(node)?;
+            }
+            cluster.start_node_process(first).await?;
+            let error = clients[first]
+                .get_object()
+                .bucket(bucket)
+                .key("warm-large")
+                .send()
+                .await
+                .expect_err("one node cannot satisfy read quorum after a cold start");
+            assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(503));
+            cluster.start_node_process(second).await?;
+            wait_for_quorum_read_admission(&[clients[first].clone(), clients[second].clone()], bucket).await?;
+            for node in [first, second] {
+                for (key, body) in &fixtures {
+                    assert_quorum_object_body(&clients[node], bucket, key, body).await?;
+                }
+                let range = clients[node]
+                    .get_object()
+                    .bucket(bucket)
+                    .key("warm-large")
+                    .range("bytes=17-1023")
+                    .send()
+                    .await?;
+                assert_eq!(range.body.collect().await?.into_bytes().as_ref(), &fixtures[1].1[17..1024]);
+                let listing = clients[node].list_objects_v2().bucket(bucket).send().await?;
+                for (key, _) in &fixtures {
+                    assert!(
+                        listing.contents().iter().any(|object| object.key() == Some(key.as_str())),
+                        "listing omitted {key}"
+                    );
+                }
+                let error = clients[node]
+                    .put_object()
+                    .bucket(bucket)
+                    .key("recovery-write")
+                    .body(Bytes::from_static(b"after recovery").into())
+                    .send()
+                    .await
+                    .expect_err("two nodes must not authorize a write");
+                assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(503));
+            }
+            let error = clients[first]
+                .delete_object()
+                .bucket(bucket)
+                .key("warm-large")
+                .send()
+                .await
+                .expect_err("two nodes must not authorize a delete");
+            assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(503));
+            assert_quorum_object_body(&clients[first], bucket, "warm-large", &fixtures[1].1).await?;
+
+            let remaining = (0..4).filter(|node| *node != first && *node != second).collect::<Vec<_>>();
+            cluster.start_node(remaining[0]).await?;
+            let deadline = tokio::time::Instant::now() + Duration::from_secs(90);
+            loop {
+                match clients[first]
+                    .put_object()
+                    .bucket(bucket)
+                    .key("recovery-write")
+                    .body(Bytes::from_static(b"after recovery").into())
+                    .send()
+                    .await
+                {
+                    Ok(_) => break,
+                    Err(error)
+                        if error.raw_response().is_some_and(|response| response.status().as_u16() == 503)
+                            && tokio::time::Instant::now() < deadline =>
+                    {
+                        tokio::time::sleep(Duration::from_millis(200)).await
+                    }
+                    Err(error) => return Err(error.into()),
+                }
+            }
+            for node in [first, second, remaining[0]] {
+                assert_quorum_object_body(&clients[node], bucket, "recovery-write", b"after recovery").await?;
+            }
+            cluster.start_node(remaining[1]).await?;
+            for client in &clients {
+                for (key, body) in &fixtures {
+                    assert_quorum_object_body(client, bucket, key, body).await?;
+                }
+                assert_quorum_object_body(client, bucket, "recovery-write", b"after recovery").await?;
+            }
+        }
+    }
+    Ok(())
+}
+
 async fn put_object(client: Client, payload: Vec<u8>, writer_id: usize) -> Result<(), String> {
     client
         .put_object()

crates/e2e_test/src/sts_query_compat_test.rs

@@ -740,7 +740,7 @@ async fn test_bucket_tags_opa_contract() -> TestResult {
         .version_id(version)
         .send()
         .await?;
-    opa.expect_bucket_tags("s3:GetObject", source, Some("finance")).await?;
+    opa.expect_bucket_tags("s3:GetObjectVersion", source, Some("finance")).await?;
     client.list_objects_v2().bucket(source).send().await?;
     opa.expect_bucket_tags("s3:ListBucket", source, Some("finance")).await?;
     client.list_object_versions().bucket(source).send().await?;

crates/ecstore/src/api/mod.rs

@@ -210,7 +210,8 @@ pub mod bucket {
 
     pub mod migration {
         pub use crate::bucket::migration::{
-            LegacyBlobDecryptFn, migration_startup_error, try_migrate_bucket_metadata, try_migrate_iam_config,
+            LegacyBlobDecryptFn, migration_startup_error, try_migrate_bucket_metadata, try_migrate_bucket_metadata_for_buckets,
+            try_migrate_iam_config,
         };
     }
 

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -14214,7 +14214,11 @@ mod tests {
     #[tokio::test]
     #[serial]
     async fn tier_free_version_recovery_production_entrypoint_enqueues_seeded_item() {
-        let (disk_paths, ecstore) = setup_test_env().await;
+        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 runtime_state = install_unconsumed_runtime_expiry_worker(&ecstore, 1).await;
         let recovery_rx = {
             let state = runtime_state.read().await;

crates/ecstore/src/bucket/metadata.rs

@@ -439,6 +439,8 @@ 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>,
@@ -514,6 +516,7 @@ 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(),
@@ -752,7 +755,10 @@ 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)?,
+                "BucketCreationCommitted" => {
+                    self.bucket_creation_committed = read_msgp_bool(rd)?;
+                    self.bucket_creation_commit_record_present = true;
+                }
                 other => {
                     tracing::debug!(field = %other, "BucketMetadata decode_from: skipping unknown field");
                     skip_msgp_value(rd)?;
@@ -765,8 +771,10 @@ impl BucketMetadata {
 
     /// Encode to msgp bytes. Field order follows MinIO BucketMetadata for compatibility.
     pub fn encode_to<W: Write>(&self, wr: &mut W) -> Result<()> {
-        // Map size: MinIO fields (25) + RustFS extensions (22)
-        let map_len: u32 = 47;
+        // 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);
         rmp::encode::write_map_len(wr, map_len)?;
 
         // MinIO field order (same as Go struct)
@@ -847,8 +855,10 @@ 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)?;
-        rmp::encode::write_str(wr, "BucketCreationCommitted")?;
-        rmp::encode::write_bool(wr, self.bucket_creation_committed)?;
+        if commit_record_present {
+            rmp::encode::write_str(wr, "BucketCreationCommitted")?;
+            rmp::encode::write_bool(wr, self.bucket_creation_committed)?;
+        }
 
         Ok(())
     }
@@ -1691,6 +1701,13 @@ 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());
     }
@@ -1708,6 +1725,7 @@ 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);
     }

crates/ecstore/src/bucket/metadata_sys.rs

@@ -1573,9 +1573,27 @@ 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>,
@@ -1619,6 +1637,7 @@ 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()),
             }),
@@ -1649,6 +1668,81 @@ 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()
     }
@@ -1839,6 +1933,7 @@ 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,
@@ -1922,6 +2017,7 @@ 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")
@@ -1983,6 +2079,7 @@ 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;
@@ -2004,6 +2101,7 @@ 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;
@@ -2021,6 +2119,7 @@ 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);
@@ -2159,6 +2258,7 @@ 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;
@@ -2167,6 +2267,7 @@ 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?;
@@ -2176,7 +2277,9 @@ 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;
@@ -2186,6 +2289,7 @@ 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
@@ -2234,6 +2338,7 @@ 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")
@@ -2290,6 +2395,7 @@ 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)]
@@ -2598,6 +2704,7 @@ 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))
@@ -2675,6 +2782,7 @@ 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}")));
@@ -3046,6 +3154,7 @@ 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;
@@ -3100,6 +3209,180 @@ 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;

crates/ecstore/src/bucket/migration.rs

@@ -273,6 +273,30 @@ where
 
     let buckets: Vec<String> = buckets_list.into_iter().map(|b| b.name).collect();
 
+    try_migrate_bucket_metadata_for_buckets(store, &buckets).await
+}
+
+/// Migrate the already discovered startup candidates without repeating a
+/// majority-only namespace listing. Every migration write retains its quorum.
+pub async fn try_migrate_bucket_metadata_for_buckets<S>(store: Arc<S>, buckets: &[String]) -> Result<()>
+where
+    S: ObjectIO<
+            Error = Error,
+            RangeSpec = HTTPRangeSpec,
+            HeaderMap = HeaderMap,
+            ObjectOptions = ObjectOptions,
+            ObjectInfo = ObjectInfo,
+            GetObjectReader = GetObjectReader,
+            PutObjectReader = PutObjReader,
+        > + ObjectOperations<
+            Error = Error,
+            ObjectInfo = ObjectInfo,
+            ObjectOptions = ObjectOptions,
+            FileInfo = FileInfo,
+            ObjectToDelete = ObjectToDelete,
+            DeletedObject = DeletedObject,
+        >,
+{
     if buckets.is_empty() {
         debug!("No migrating bucket metadata found");
         return Ok(());

crates/ecstore/src/core/sets.rs

@@ -72,6 +72,38 @@ 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)
 }
@@ -273,6 +305,14 @@ 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)),

crates/ecstore/src/disk/disk_store.rs

@@ -2320,6 +2320,15 @@ 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() {

crates/ecstore/src/disk/error_reduce.rs

@@ -131,6 +131,15 @@ pub fn build_write_quorum_failure_summary(
 
 fn dominant_error_label(errors: &[Option<Error>], ignored_errs: &[Error], dominant_error: Option<&Error>) -> &'static str {
     let Some(dominant_error) = dominant_error else {
+        // The reducer also returns None when every failure is ignored for
+        // quorum voting. That case contains no successful nil entry.
+        if !errors.is_empty() && errors.iter().all(Option::is_some) {
+            return if errors.iter().all(|error| error.as_ref() == Some(&Error::DiskNotFound)) {
+                "disk_not_found"
+            } else {
+                "all_ignored"
+            };
+        }
         return "nil_dominated";
     };
 
@@ -387,9 +396,9 @@ mod tests {
     }
 
     #[test]
-    fn test_build_write_quorum_failure_summary_all_ignored_is_nil_dominated() {
-        // All disks offline (all DiskNotFound). The dominant error must be None and
-        // the label must be "nil_dominated" rather than falling back to "other_error".
+    fn test_build_write_quorum_failure_summary_all_missing_is_disk_not_found() {
+        // Ignoring unavailable disks for quorum reduction must not imply a
+        // successful nil entry in the diagnostic label.
         let errors = vec![
             Some(Error::DiskNotFound),
             Some(Error::DiskNotFound),
@@ -397,7 +406,21 @@ mod tests {
         ];
         let summary = build_write_quorum_failure_summary(&errors, OBJECT_OP_IGNORED_ERRS, 2);
         assert_eq!(summary.dominant_error, None);
-        assert_eq!(summary.dominant_error_label, "nil_dominated");
+        assert_eq!(summary.achieved, 0);
+        assert_eq!(summary.dominant_error_label, "disk_not_found");
+    }
+
+    #[test]
+    fn test_build_write_quorum_failure_summary_all_ignored_has_no_success() {
+        let errors = [Some(Error::FaultyDisk), Some(Error::DiskNotFound)];
+        let summary = build_write_quorum_failure_summary(&errors, OBJECT_OP_IGNORED_ERRS, 2);
+        assert_eq!(summary.dominant_error, None);
+        assert_eq!(summary.achieved, 0);
+        assert_eq!(summary.dominant_error_label, "all_ignored");
+        assert_eq!(
+            reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, 2),
+            Some(Error::ErasureWriteQuorum)
+        );
     }
 
     #[test]

crates/ecstore/src/disk/fs.rs

@@ -117,6 +117,9 @@ 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(),
@@ -160,6 +163,7 @@ 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
 }
 
@@ -211,15 +215,18 @@ 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)
 }
 

crates/ecstore/src/disk/local.rs

@@ -262,6 +262,7 @@ 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)?;
@@ -287,6 +288,7 @@ 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();
@@ -2148,6 +2150,9 @@ 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)
@@ -3920,6 +3925,7 @@ 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}")))??;
@@ -5167,6 +5173,7 @@ 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
@@ -5535,6 +5542,10 @@ 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))
@@ -5670,6 +5681,7 @@ impl LocalDisk {
         let mut disk = Self {
             root: root.clone(),
             publication_root,
+            publication_history,
             io_root: io_root.clone(),
             #[cfg(target_os = "linux")]
             mount_lease,
@@ -7439,6 +7451,7 @@ 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 {
@@ -7961,6 +7974,7 @@ 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
                 {
@@ -7968,6 +7982,7 @@ impl LocalDisk {
                 }
 
                 tokio::task::spawn_blocking(move || {
+                    os::record_native_publication(&path);
                     #[cfg(test)]
                     run_owned_file_write_before_open(&path);
 
@@ -9581,6 +9596,9 @@ 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
@@ -9738,6 +9756,7 @@ 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.
@@ -10126,6 +10145,7 @@ 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() {
@@ -10197,6 +10217,7 @@ 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() => {}
@@ -10807,12 +10828,27 @@ 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.
@@ -11552,6 +11588,190 @@ 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() {

crates/ecstore/src/disk/mod.rs

@@ -327,6 +327,13 @@ 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 {
@@ -1121,6 +1128,11 @@ 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>>;

crates/ecstore/src/disk/os.rs

@@ -21,7 +21,7 @@ use futures::TryStreamExt;
 use parking_lot::Mutex;
 use rustfs_utils::path::SLASH_SEPARATOR;
 use std::{
-    collections::{HashMap, VecDeque},
+    collections::{HashMap, HashSet, VecDeque},
     io,
     path::{Component, Path, PathBuf},
     sync::{Arc, LazyLock, Weak},
@@ -276,6 +276,14 @@ 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>;
@@ -1381,6 +1389,226 @@ 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>> =
@@ -2083,6 +2311,7 @@ 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;
     }
@@ -2104,12 +2333,19 @@ 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 || std::fs::create_dir_all(path)).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
 }
 
 #[tracing::instrument(name = "rename_all", level = "debug", skip_all)]
@@ -2120,6 +2356,7 @@ 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
 }
@@ -2131,6 +2368,7 @@ 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(),
@@ -2155,6 +2393,7 @@ 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)?;
@@ -2399,6 +2638,7 @@ 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(
@@ -2453,6 +2693,7 @@ 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();
@@ -2476,6 +2717,7 @@ 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();
@@ -2878,6 +3120,7 @@ fn rename_into_existing_parent(
     dst_file_path: &Path,
     parent_guard: Option<&ExistingBaseDirectoryGuard>,
 ) -> io::Result<()> {
+    record_native_publication(dst_file_path);
     use rustix::fs::{Mode, OFlags, open, renameat};
 
     let Some(parent_guard) = parent_guard else {
@@ -2926,6 +3169,7 @@ 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},
@@ -4214,6 +4458,7 @@ 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 = dir_path
         .strip_prefix(base_dir)
         .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "rename destination must remain below its base directory"))?;
@@ -5718,6 +5963,77 @@ 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() {
@@ -7737,3 +8053,272 @@ mod tests {
         assert_eq!(err.kind(), io::ErrorKind::NotFound);
     }
 }
+
+#[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"
+        );
+    }
+}

crates/ecstore/src/error/mod.rs

@@ -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>;

crates/ecstore/src/object_api/object_api_utils.rs

@@ -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 s3s::S3ErrorCode;
+use rustfs_s3_types::S3ErrorCode;
 
 //#[derive(Clone)]
 pub struct PutObjReader {
@@ -231,6 +231,45 @@ 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

crates/ecstore/src/services/tier/tier.rs

@@ -82,8 +82,9 @@ 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::{S3ErrorCode, dto::BucketLifecycleConfiguration};
+use s3s::dto::BucketLifecycleConfiguration;
 
 use super::{
     tier_handlers::{ERR_TIER_BUCKET_NOT_FOUND, ERR_TIER_CONNECT_ERR, ERR_TIER_INVALID_CREDENTIALS, ERR_TIER_PERM_ERR},

crates/ecstore/src/set_disk/ops/bucket.rs

@@ -123,8 +123,18 @@ impl SetDisks {
     }
 
     pub(crate) async fn list_bucket_for_scanner(&self, _opts: &BucketOptions) -> Result<(Vec<BucketInfo>, bool)> {
+        self.list_bucket_with_quorum(BucketInfoQuorum::Write).await
+    }
+
+    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;
+        // Discovery uses the same namespace read boundary as stat_bucket.
+        // It grants no authority to mutate buckets or accept object reads.
+        let required = match quorum {
+            BucketInfoQuorum::Read => disks.len().div_ceil(2).max(1),
+            BucketInfoQuorum::Write => write_quorum,
+        };
 
         let mut futures = Vec::with_capacity(disks.len());
         for disk in disks {
@@ -153,7 +163,11 @@ impl SetDisks {
             }
         }
 
-        if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) {
+        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),
+        };
+        if let Some(err) = error {
             return Err(err.into());
         }
 
@@ -179,7 +193,7 @@ impl SetDisks {
         topology_complete &= counts.values().all(|(count, _)| *count >= write_quorum);
         let mut buckets = counts
             .into_values()
-            .filter_map(|(count, bucket)| (count >= write_quorum).then_some(bucket))
+            .filter_map(|(count, bucket)| (count >= required).then_some(bucket))
             .collect::<Vec<_>>();
         buckets.sort_by(|left, right| left.name.cmp(&right.name));
         Ok((buckets, topology_complete))

crates/ecstore/src/set_disk/ops/heal.rs

@@ -99,6 +99,7 @@ static HEAL_RENAME_FAILURES: std::sync::Mutex<Vec<(String, String, usize)>> = st
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum ReadRepairPausePhase {
     Data,
+    PartStaging(usize),
     Commit,
 }
 
@@ -250,7 +251,16 @@ fn should_fail_heal_rename(_bucket: &str, _object: &str, _disk_index: usize) ->
 }
 
 #[cfg(test)]
-static HEAL_WRITER_FAILURES: std::sync::Mutex<Vec<(String, String, usize, DiskError)>> = std::sync::Mutex::new(Vec::new());
+struct HealWriterFailure {
+    bucket: String,
+    object: String,
+    part_number: Option<usize>,
+    disk_index: usize,
+    error: DiskError,
+}
+
+#[cfg(test)]
+static HEAL_WRITER_FAILURES: std::sync::Mutex<Vec<HealWriterFailure>> = std::sync::Mutex::new(Vec::new());
 
 #[cfg(test)]
 struct HealWriterFailureScope {
@@ -260,21 +270,23 @@ struct HealWriterFailureScope {
 
 #[cfg(test)]
 impl HealWriterFailureScope {
-    fn install(bucket: &str, object: &str, disk_indexes: &[usize], error: DiskError) -> Self {
+    fn install(bucket: &str, object: &str, part_number: Option<usize>, disk_indexes: &[usize], error: DiskError) -> Self {
         let mut failures = HEAL_WRITER_FAILURES
             .lock()
             .expect("heal writer failure registry should not poison");
         assert!(
-            !failures.iter().any(|(registered_bucket, registered_object, _, _)| {
-                registered_bucket == bucket && registered_object == object
-            }),
+            !failures
+                .iter()
+                .any(|failure| failure.bucket == bucket && failure.object == object),
             "heal writer failures must be installed once per object"
         );
-        failures.extend(
-            disk_indexes
-                .iter()
-                .map(|index| (bucket.to_string(), object.to_string(), *index, error.clone())),
-        );
+        failures.extend(disk_indexes.iter().map(|index| HealWriterFailure {
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            part_number,
+            disk_index: *index,
+            error: error.clone(),
+        }));
         Self {
             bucket: bucket.to_string(),
             object: object.to_string(),
@@ -288,25 +300,28 @@ impl Drop for HealWriterFailureScope {
         HEAL_WRITER_FAILURES
             .lock()
             .expect("heal writer failure registry should not poison")
-            .retain(|(bucket, object, _, _)| bucket != &self.bucket || object != &self.object);
+            .retain(|failure| failure.bucket != self.bucket || failure.object != self.object);
     }
 }
 
 #[cfg(test)]
-fn injected_heal_writer_error(bucket: &str, object: &str, disk_index: usize) -> Option<DiskError> {
+fn injected_heal_writer_error(bucket: &str, object: &str, part_number: usize, disk_index: usize) -> Option<DiskError> {
     let mut failures = HEAL_WRITER_FAILURES
         .lock()
         .expect("heal writer failure registry should not poison");
     failures
         .iter()
-        .position(|(registered_bucket, registered_object, registered_index, _)| {
-            registered_bucket == bucket && registered_object == object && *registered_index == disk_index
+        .position(|failure| {
+            failure.bucket == bucket
+                && failure.object == object
+                && failure.disk_index == disk_index
+                && failure.part_number.is_none_or(|part| part == part_number)
         })
-        .map(|position| failures.swap_remove(position).3)
+        .map(|position| failures.swap_remove(position).error)
 }
 
 #[cfg(not(test))]
-fn injected_heal_writer_error(_bucket: &str, _object: &str, _disk_index: usize) -> Option<DiskError> {
+fn injected_heal_writer_error(_bucket: &str, _object: &str, _part_number: usize, _disk_index: usize) -> Option<DiskError> {
     None
 }
 
@@ -518,30 +533,64 @@ fn is_retryable_heal_writer_error(error: &DiskError) -> bool {
     ) || error.is_retryable_internode_write_failure()
 }
 
-fn warn_heal_writer_failures(
-    bucket: &str,
-    object: &str,
-    version_id: &str,
-    writer_failure_count: usize,
-    result: &'static str,
-    first_failure: &(usize, usize, String),
-) {
-    let (first_part_number, first_disk_index, first_error) = first_failure;
-    warn!(
-        event = EVENT_SET_DISK_HEAL,
-        component = LOG_COMPONENT_ECSTORE,
-        subsystem = LOG_SUBSYSTEM_SET_DISK,
-        bucket,
-        object,
-        version_id,
-        writer_failure_count,
-        first_part_number,
-        first_disk_index,
-        error = %first_error,
-        result,
-        state = "writer_unavailable",
-        "Set disk object heal writer failures"
-    );
+struct HealWriterFailures {
+    target_count: usize,
+    count: usize,
+    first: Option<(usize, usize, String)>,
+    error: Option<DiskError>,
+}
+
+impl HealWriterFailures {
+    fn new(target_count: usize) -> Self {
+        Self {
+            target_count,
+            count: 0,
+            first: None,
+            error: None,
+        }
+    }
+
+    fn record(&mut self, part_number: usize, disk_index: usize, error: DiskError) {
+        self.count += 1;
+        if self.first.is_none() {
+            self.first = Some((part_number, disk_index, heal_writer_error_summary(&error)));
+        }
+        // A permanent failure must not be hidden behind an earlier transient
+        // target failure when the remaining targets also become unavailable.
+        if self
+            .error
+            .as_ref()
+            .is_none_or(|first| is_retryable_heal_writer_error(first) && !is_retryable_heal_writer_error(&error))
+        {
+            self.error = Some(error);
+        }
+    }
+
+    fn into_error(self) -> DiskError {
+        self.error.unwrap_or(DiskError::ErasureWriteQuorum)
+    }
+
+    fn warn(&self, bucket: &str, object: &str, version_id: &str, result: &'static str) {
+        let Some((first_part_number, first_disk_index, first_error)) = &self.first else {
+            return;
+        };
+        warn!(
+            event = EVENT_SET_DISK_HEAL,
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_SET_DISK,
+            state = "writer_unavailable",
+            result,
+            bucket,
+            object,
+            version_id,
+            target_count = self.target_count,
+            writer_failure_count = self.count,
+            first_part_number,
+            first_disk_index,
+            error = %first_error,
+            "Set disk object heal writer failures"
+        );
+    }
 }
 
 impl SetDisks {
@@ -1548,196 +1597,169 @@ impl SetDisks {
                         let dst_data_dir = data_dir;
 
                         if !latest_meta.deleted && !latest_meta.is_remote() {
-                            let erasure_info = latest_meta.erasure.clone();
-                            let mut writer_failure_count = 0usize;
-                            let mut first_writer_failure = None;
-                            let mut first_writer_error = None;
-                            let mut all_writer_failures_retryable = true;
-                            let mut writer_failure_warned = false;
+                            let stage_result: disk::error::Result<Option<DiskError>> = async {
+                                let erasure_info = latest_meta.erasure.clone();
+                                let mut writer_failures = HealWriterFailures::new(disks_to_heal_count);
 
-                            for (part_index, part) in latest_meta.parts.iter().enumerate() {
-                                use crate::io_support::shard_integrity::{PartProofReader, PreparedIntegrity, ShardVerifier};
-                                let proof = part
-                                    .integrity
-                                    .as_ref()
-                                    .map(|part| {
-                                        PartProofReader::new(part.clone(), &parts_metadata, &latest_disks, bucket, object)
-                                    })
-                                    .transpose()
-                                    .map_err(DiskError::from)?;
-                                // This reader covers the whole encoded shard. A range
-                                // offset can exclude the final encoded byte when the
-                                // logical part ends inside a data block, allowing a
-                                // one-byte tail truncation to evade deep Heal.
-                                let till_offset = usize::try_from(erasure.shard_file_size(part.size as i64))
-                                    .map_err(|_| DiskError::FileCorrupt)?;
-                                let use_mmap_read = object_mmap_read_enabled();
+                                for (part_index, part) in latest_meta.parts.iter().enumerate() {
+                                    use crate::io_support::shard_integrity::{PartProofReader, PreparedIntegrity, ShardVerifier};
+                                    let proof = part
+                                        .integrity
+                                        .as_ref()
+                                        .map(|part| {
+                                            PartProofReader::new(part.clone(), &parts_metadata, &latest_disks, bucket, object)
+                                        })
+                                        .transpose()?;
+                                    // Cover the complete encoded shard, including a logical
+                                    // part's partial final data block, when checking truncation.
+                                    let till_offset = usize::try_from(erasure.shard_file_size(part.size as i64))
+                                        .map_err(|_| DiskError::FileCorrupt)?;
+                                    let use_mmap_read = object_mmap_read_enabled();
 
-                                let mut readers = Vec::with_capacity(latest_disks.len());
-                                let mut writers = Vec::with_capacity(out_dated_disks.len());
-                                // let mut errors = Vec::with_capacity(out_dated_disks.len());
+                                    let mut writers = Vec::with_capacity(out_dated_disks.len());
 
-                                let mut prefer = vec![false; latest_disks.len()];
-                                for (index, disk) in latest_disks.iter().enumerate() {
+                                    // Preserve the committed layout: recomputing inline-ness here
+                                    // (with a hardcoded unversioned threshold) makes healed replicas
+                                    // diverge from healthy ones in quorum identity, so heal would
+                                    // flag them forever.
+                                    let is_inline_buffer = latest_meta.inline_data();
+                                    // create writers for all disk positions, but only for outdated disks
+                                    for (index, disk_op) in out_dated_disks.iter().enumerate() {
+                                        if let Some(outdated_disk) = disk_op {
+                                            let writer_result = if let Some(error) =
+                                                injected_heal_writer_error(bucket, object, part.number, index)
+                                            {
+                                                Err(error)
+                                            } else {
+                                                create_bitrot_writer(
+                                                    is_inline_buffer,
+                                                    Some(outdated_disk),
+                                                    RUSTFS_META_TMP_BUCKET,
+                                                    &path_join_buf(&[
+                                                        &tmp_id.to_string(),
+                                                        &dst_data_dir.to_string(),
+                                                        &format!("part.{}", part.number),
+                                                    ]),
+                                                    erasure.shard_file_size(part.size as i64),
+                                                    erasure.shard_size(),
+                                                    HashAlgorithm::HighwayHash256S,
+                                                )
+                                                .await
+                                            };
+                                            let writer = match writer_result {
+                                                Ok(writer) => writer,
+                                                Err(err) => {
+                                                    writer_failures.record(part.number, index, err);
+                                                    writers.push(None);
+                                                    continue;
+                                                }
+                                            };
+                                            let mut writer = writer;
+                                            if let Some(proof) = &proof {
+                                                let verifier = ShardVerifier::new(std::sync::Arc::clone(proof), index, 0, None)?;
+                                                writer.set_integrity(verifier);
+                                            }
+                                            writers.push(Some(writer));
+                                        } else {
+                                            writers.push(None);
+                                        }
+                                    }
+
+                                    // Heal each part. erasure.Heal() will write the healed
+                                    // part to .rustfs/tmp/uuid/ which needs to be renamed
+                                    // later to the final location.
+                                    if writers.iter().all(Option::is_none) {
+                                        writer_failures.warn(bucket, object, version_id, "all_targets_unavailable");
+                                        // No target can accept this part. Preserve the setup
+                                        // cause before opening source readers or reconstructing.
+                                        return Ok(Some(writer_failures.into_error()));
+                                    }
+                                    #[cfg(test)]
+                                    pause_read_repair(bucket, object, ReadRepairPausePhase::PartStaging(part.number)).await;
+
+                                    let mut readers = Vec::with_capacity(latest_disks.len());
+
+                                    let mut prefer = vec![false; latest_disks.len()];
                                     let this_part_errs =
                                         Self::shuffle_check_parts(&data_errs_by_part[&part_index], &erasure_info.distribution);
-                                    if this_part_errs[index] != CHECK_PART_SUCCESS {
-                                        trace!(
-                                            event = EVENT_SET_DISK_HEAL,
-                                            component = LOG_COMPONENT_ECSTORE,
-                                            subsystem = LOG_SUBSYSTEM_SET_DISK,
-                                            part_number = part.number,
-                                            disk_index = index,
-                                            part_status = this_part_errs[index],
-                                            state = "source_shard_skipped",
-                                            "Set disk source shard skipped"
-                                        );
-                                        readers.push(None);
-                                        continue;
-                                    }
-
-                                    if let (Some(disk), Some(metadata)) = (disk, &copy_parts_metadata[index]) {
-                                        let checksum_info = metadata.erasure.get_checksum_info(part.number);
-                                        let checksum_algo = if metadata.uses_legacy_checksum
-                                            && checksum_info.algorithm == HashAlgorithm::HighwayHash256S
-                                        {
-                                            HashAlgorithm::HighwayHash256SLegacy
-                                        } else {
-                                            checksum_info.algorithm
-                                        };
-
-                                        match create_bitrot_reader(
-                                            metadata.data.as_deref(),
-                                            Some(disk),
-                                            bucket,
-                                            &path_join_buf(&[object, &src_data_dir, &format!("part.{}", part.number)]),
-                                            0,
-                                            till_offset,
-                                            erasure.shard_size(),
-                                            checksum_algo.clone(),
-                                            false,
-                                            use_mmap_read,
-                                        )
-                                        .await
-                                        {
-                                            Ok(Some(mut reader)) => {
-                                                if let Some(proof) = &proof {
-                                                    reader
-                                                        .set_integrity(
-                                                            ShardVerifier::new(std::sync::Arc::clone(proof), index, 0, None)
-                                                                .map_err(DiskError::from)?,
-                                                        )
-                                                        .map_err(DiskError::from)?;
-                                                }
-                                                readers.push(Some(reader));
-                                            }
-                                            Ok(None) => {
-                                                readers.push(None);
-                                                continue;
-                                            }
-                                            Err(_e) => {
-                                                readers.push(None);
-                                                continue;
-                                            }
+                                    for (index, disk) in latest_disks.iter().enumerate() {
+                                        if this_part_errs[index] != CHECK_PART_SUCCESS {
+                                            trace!(
+                                                event = EVENT_SET_DISK_HEAL,
+                                                component = LOG_COMPONENT_ECSTORE,
+                                                subsystem = LOG_SUBSYSTEM_SET_DISK,
+                                                part_number = part.number,
+                                                disk_index = index,
+                                                part_status = this_part_errs[index],
+                                                state = "source_shard_skipped",
+                                                "Set disk source shard skipped"
+                                            );
+                                            readers.push(None);
+                                            continue;
                                         }
 
-                                        prefer[index] = disk.host_name().is_empty();
-                                    } else {
-                                        readers.push(None);
-                                        // errors.push(Some(DiskError::DiskNotFound));
-                                    }
-                                }
+                                        if let (Some(disk), Some(metadata)) = (disk, &copy_parts_metadata[index]) {
+                                            let checksum_info = metadata.erasure.get_checksum_info(part.number);
+                                            let checksum_algo = if metadata.uses_legacy_checksum
+                                                && checksum_info.algorithm == HashAlgorithm::HighwayHash256S
+                                            {
+                                                HashAlgorithm::HighwayHash256SLegacy
+                                            } else {
+                                                checksum_info.algorithm
+                                            };
 
-                                // Preserve the committed layout: recomputing inline-ness here
-                                // (with a hardcoded unversioned threshold) makes healed replicas
-                                // diverge from healthy ones in quorum identity, so heal would
-                                // flag them forever.
-                                let is_inline_buffer = latest_meta.inline_data();
-                                // create writers for all disk positions, but only for outdated disks
-                                for (index, disk_op) in out_dated_disks.iter().enumerate() {
-                                    if let Some(outdated_disk) = disk_op {
-                                        let writer_result = if let Some(error) = injected_heal_writer_error(bucket, object, index)
-                                        {
-                                            Err(error)
-                                        } else {
-                                            create_bitrot_writer(
-                                                is_inline_buffer,
-                                                Some(outdated_disk),
-                                                RUSTFS_META_TMP_BUCKET,
-                                                &path_join_buf(&[
-                                                    &tmp_id.to_string(),
-                                                    &dst_data_dir.to_string(),
-                                                    &format!("part.{}", part.number),
-                                                ]),
-                                                erasure.shard_file_size(part.size as i64),
+                                            match create_bitrot_reader(
+                                                metadata.data.as_deref(),
+                                                Some(disk),
+                                                bucket,
+                                                &path_join_buf(&[object, &src_data_dir, &format!("part.{}", part.number)]),
+                                                0,
+                                                till_offset,
                                                 erasure.shard_size(),
-                                                HashAlgorithm::HighwayHash256S,
+                                                checksum_algo.clone(),
+                                                false,
+                                                use_mmap_read,
                                             )
                                             .await
-                                        };
-                                        let writer = match writer_result {
-                                            Ok(writer) => writer,
-                                            Err(err) => {
-                                                writer_failure_count += 1;
-                                                all_writer_failures_retryable &= is_retryable_heal_writer_error(&err);
-                                                if first_writer_failure.is_none() {
-                                                    first_writer_failure =
-                                                        Some((part.number, index, heal_writer_error_summary(&err)));
-                                                    first_writer_error = Some(err);
+                                            {
+                                                Ok(Some(mut reader)) => {
+                                                    if let Some(proof) = &proof {
+                                                        let verifier =
+                                                            ShardVerifier::new(std::sync::Arc::clone(proof), index, 0, None)?;
+                                                        reader.set_integrity(verifier)?;
+                                                    }
+                                                    readers.push(Some(reader));
+                                                }
+                                                Ok(None) => {
+                                                    readers.push(None);
+                                                    continue;
+                                                }
+                                                Err(_e) => {
+                                                    readers.push(None);
+                                                    continue;
                                                 }
-                                                writers.push(None);
-                                                continue;
                                             }
-                                        };
-                                        let mut writer = writer;
-                                        if let Some(proof) = &proof {
-                                            writer.set_integrity(
-                                                ShardVerifier::new(std::sync::Arc::clone(proof), index, 0, None)
-                                                    .map_err(DiskError::from)?,
-                                            );
-                                        }
-                                        writers.push(Some(writer));
-                                    } else {
-                                        writers.push(None);
-                                    }
-                                }
 
-                                // Heal each part. erasure.Heal() will write the healed
-                                // part to .rustfs/tmp/uuid/ which needs to be renamed
-                                // later to the final location.
-                                if writer_failure_count > 0
-                                    && writers.iter().all(Option::is_none)
-                                    && let Some(first_failure) = first_writer_failure.as_ref()
-                                {
-                                    warn_heal_writer_failures(
-                                        bucket,
-                                        object,
-                                        version_id,
-                                        writer_failure_count,
-                                        "all_targets_unavailable",
-                                        first_failure,
-                                    );
-                                    writer_failure_warned = true;
-                                }
-                                let heal_part = erasure.heal(&mut writers, readers, part.size, &prefer);
-                                let heal_part_result = if read_repair_uses_shared_lock {
-                                    match tokio::time::timeout(READ_REPAIR_DATA_PHASE_TIMEOUT, heal_part).await {
-                                        Ok(result) => result,
-                                        Err(_) => Err(DiskError::Timeout),
+                                            prefer[index] = disk.host_name().is_empty();
+                                        } else {
+                                            readers.push(None);
+                                            // errors.push(Some(DiskError::DiskNotFound));
+                                        }
                                     }
-                                } else {
-                                    heal_part.await
-                                };
-                                if let Err(e) = heal_part_result {
-                                    // Don't leak the partially-written healed shards in
-                                    // .rustfs/tmp when heal fails midway (backlog#799 B20).
-                                    let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await;
-                                    return Err(e);
-                                }
-                                // close_bitrot_writers(&mut writers).await?;
-                                if !is_inline_buffer && let Some(proof) = &proof {
-                                    let before = out_dated_disks.iter().filter(|disk| disk.is_some()).count();
-                                    let proof_result = async {
+
+                                    let heal_part = erasure.heal(&mut writers, readers, part.size, &prefer);
+                                    let heal_part_result = if read_repair_uses_shared_lock {
+                                        match tokio::time::timeout(READ_REPAIR_DATA_PHASE_TIMEOUT, heal_part).await {
+                                            Ok(result) => result,
+                                            Err(_) => Err(DiskError::Timeout),
+                                        }
+                                    } else {
+                                        heal_part.await
+                                    };
+                                    heal_part_result?;
+                                    // close_bitrot_writers(&mut writers).await?;
+                                    if !is_inline_buffer && let Some(proof) = &proof {
+                                        let before = out_dated_disks.iter().filter(|disk| disk.is_some()).count();
                                         PreparedIntegrity::copy_from(proof)
                                             .await?
                                             .write(
@@ -1746,89 +1768,71 @@ impl SetDisks {
                                                 RUSTFS_META_TMP_BUCKET,
                                                 &format!("{tmp_id}/{dst_data_dir}"),
                                             )
-                                            .await
-                                    }
-                                    .await;
-                                    if let Err(error) = proof_result {
-                                        let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await;
-                                        return Err(error.into());
-                                    }
-                                    disks_to_heal_count -= before - out_dated_disks.iter().filter(|disk| disk.is_some()).count();
-                                }
-
-                                for (index, disk_op) in out_dated_disks.iter_mut().enumerate() {
-                                    if disk_op.is_none() {
-                                        continue;
+                                            .await?;
+                                        disks_to_heal_count -=
+                                            before - out_dated_disks.iter().filter(|disk| disk.is_some()).count();
                                     }
 
-                                    if writers[index].is_none() {
-                                        *disk_op = None;
-                                        disks_to_heal_count -= 1;
-                                        continue;
-                                    }
-
-                                    parts_metadata[index].data_dir = Some(dst_data_dir);
-                                    parts_metadata[index].add_object_part(
-                                        part.number,
-                                        part.etag.clone(),
-                                        part.size,
-                                        part.mod_time,
-                                        part.actual_size,
-                                        part.index.clone(),
-                                        part.checksums.clone(),
-                                    );
-                                    if let Some(updated) = parts_metadata[index]
-                                        .parts
-                                        .iter_mut()
-                                        .find(|updated| updated.number == part.number)
-                                    {
-                                        updated.integrity.clone_from(&part.integrity);
-                                    }
-                                    if is_inline_buffer {
-                                        if let Some(writer) = writers[index].take() {
-                                            // if let Some(w) = writer.as_any().downcast_ref::<BitrotFileWriter>() {
-                                            //     parts_metadata[index].data = Some(w.inline_data().to_vec());
-                                            // }
-                                            parts_metadata[index].data =
-                                                Some(writer.into_inline_data().map(Bytes::from).unwrap_or_default());
+                                    for (index, disk_op) in out_dated_disks.iter_mut().enumerate() {
+                                        if disk_op.is_none() {
+                                            continue;
                                         }
-                                        parts_metadata[index].set_inline_data();
-                                    } else {
-                                        parts_metadata[index].data = None;
-                                    }
-                                }
 
-                                if disks_to_heal_count == 0 {
-                                    if !writer_failure_warned && let Some(first_failure) = first_writer_failure.as_ref() {
-                                        warn_heal_writer_failures(
-                                            bucket,
-                                            object,
-                                            version_id,
-                                            writer_failure_count,
-                                            "all_targets_unavailable",
-                                            first_failure,
+                                        if writers[index].is_none() {
+                                            *disk_op = None;
+                                            disks_to_heal_count -= 1;
+                                            continue;
+                                        }
+
+                                        parts_metadata[index].data_dir = Some(dst_data_dir);
+                                        parts_metadata[index].add_object_part(
+                                            part.number,
+                                            part.etag.clone(),
+                                            part.size,
+                                            part.mod_time,
+                                            part.actual_size,
+                                            part.index.clone(),
+                                            part.checksums.clone(),
                                         );
+                                        if let Some(updated) = parts_metadata[index]
+                                            .parts
+                                            .iter_mut()
+                                            .find(|updated| updated.number == part.number)
+                                        {
+                                            updated.integrity.clone_from(&part.integrity);
+                                        }
+                                        if is_inline_buffer {
+                                            if let Some(writer) = writers[index].take() {
+                                                // if let Some(w) = writer.as_any().downcast_ref::<BitrotFileWriter>() {
+                                                //     parts_metadata[index].data = Some(w.inline_data().to_vec());
+                                                // }
+                                                parts_metadata[index].data =
+                                                    Some(writer.into_inline_data().map(Bytes::from).unwrap_or_default());
+                                            }
+                                            parts_metadata[index].set_inline_data();
+                                        } else {
+                                            parts_metadata[index].data = None;
+                                        }
                                     }
-                                    // Clean up healed shards written to .rustfs/tmp before bailing (B20).
-                                    let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await;
-                                    let error = if all_writer_failures_retryable {
-                                        first_writer_error.take().unwrap_or(DiskError::FaultyDisk)
-                                    } else {
-                                        DiskError::other(format!("all drives had write errors, unable to heal {bucket}/{object}"))
-                                    };
-                                    return Ok((result, Some(error)));
-                                }
-                            }
 
-                            if !writer_failure_warned && let Some(first_failure) = first_writer_failure.as_ref() {
-                                warn_heal_writer_failures(
-                                    bucket,
-                                    object,
-                                    version_id,
-                                    writer_failure_count,
-                                    "partial_targets_unavailable",
-                                    first_failure,
-                                );
+                                    if disks_to_heal_count == 0 {
+                                        writer_failures.warn(bucket, object, version_id, "all_targets_unavailable");
+                                        return Ok(Some(writer_failures.into_error()));
+                                    }
+                                }
+
+                                writer_failures.warn(bucket, object, version_id, "partial_targets_unavailable");
+                                Ok(None)
+                            }
+                            .await;
+                            match stage_result {
+                                Ok(None) => {}
+                                failure => {
+                                    // Release readers and writers before removing all parts
+                                    // staged under this heal's exclusively owned prefix.
+                                    let _ = self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_id).await;
+                                    return failure.map(|error| (result, error));
+                                }
                             }
                         }
                         let _read_repair_write_lock_guard = if let Some(expected_fingerprint) = read_repair_commit_fingerprint {
@@ -3531,8 +3535,8 @@ impl SetDisks {
 #[cfg(test)]
 mod heal_result_report_tests {
     use super::{
-        AbsenceProofRequest, DanglingCheckPartsFailure, DanglingDeleteFailure, DanglingDeleteSafety, ReadRepairCommitFingerprint,
-        ReadRepairPauseScope, SetDisks, heal_writer_error_summary, is_retryable_heal_writer_error,
+        AbsenceProofRequest, DanglingCheckPartsFailure, DanglingDeleteFailure, DanglingDeleteSafety, HealWriterFailures,
+        ReadRepairCommitFingerprint, ReadRepairPauseScope, SetDisks, heal_writer_error_summary, is_retryable_heal_writer_error,
     };
     use super::{HEAL_RENAME_INCOMPLETE, HealRenameFailureScope, HealWriterFailureScope};
     use crate::disk::endpoint::Endpoint;
@@ -3786,10 +3790,17 @@ mod heal_result_report_tests {
     #[tokio::test(flavor = "current_thread")]
     #[serial_test::serial]
     async fn heal_writer_failures_emit_one_aggregate_warning_per_object() {
-        for (case, failed_target_count, expected_result, expect_error) in [
-            ("partial", 1usize, "partial_targets_unavailable", false),
-            ("all", 2usize, "all_targets_unavailable", true),
+        for (case, target_count, failed_target_count, injected_error) in [
+            ("partial", 2usize, 1usize, DiskError::DiskFull),
+            ("all-full", 2, 2, DiskError::DiskFull),
+            ("all-timeout", 2, 2, DiskError::Timeout),
+            ("single-faulty", 1, 1, DiskError::FaultyDisk),
         ] {
+            let expected_result = if target_count == failed_target_count {
+                "all_targets_unavailable"
+            } else {
+                "partial_targets_unavailable"
+            };
             let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
             let bucket = format!("heal-writer-{case}");
             let object = "object.bin";
@@ -3816,11 +3827,13 @@ mod heal_result_report_tests {
                 .read_version("", &bucket, object, "", &ReadOptions::default())
                 .await
                 .expect("source metadata should be readable");
-            let mut target_slots = [source.erasure.distribution[0] - 1, source.erasure.distribution[1] - 1];
+            let mut target_slots = (0..target_count)
+                .map(|index| source.erasure.distribution[index] - 1)
+                .collect::<Vec<_>>();
             target_slots.sort_unstable();
 
-            for index in [0, 1] {
-                remove_current_object_part(&temp_dirs[index], &bucket, object)
+            for dir in temp_dirs.iter().take(target_count) {
+                remove_current_object_part(dir, &bucket, object)
                     .await
                     .expect("target shard should be removed before heal");
             }
@@ -3834,7 +3847,9 @@ mod heal_result_report_tests {
                 .without_time()
                 .finish();
             let subscriber_guard = tracing::subscriber::set_default(subscriber);
-            let failure_scope = HealWriterFailureScope::install(&bucket, object, failed_slots, DiskError::DiskFull);
+            let tmp_before = non_trash_tmp_entries(&temp_dirs).await;
+            let injected_summary = heal_writer_error_summary(&injected_error);
+            let failure_scope = HealWriterFailureScope::install(&bucket, object, None, failed_slots, injected_error.clone());
 
             let heal_outcome = set
                 .heal_object(
@@ -3851,12 +3866,22 @@ mod heal_result_report_tests {
             drop(failure_scope);
             drop(subscriber_guard);
 
+            let (_, error) = heal_outcome.expect("writer setup failures must retain a heal report and their typed error");
             assert_eq!(
-                heal_outcome.is_err(),
-                expect_error,
-                "{case}: aggregate heal result should match writer outcomes"
+                error,
+                (target_count == failed_target_count).then_some(injected_error),
+                "{case}: partial repair succeeds and all-target failure retains its cause"
+            );
+            assert_eq!(
+                non_trash_tmp_entries(&temp_dirs).await,
+                tmp_before,
+                "{case}: heal must clean temporary shards"
             );
             let output = logs.contents();
+            assert!(
+                !output.contains("Erasure encode write quorum unavailable"),
+                "{case}: unavailable heal targets must not fabricate an all-disk encode failure: {output}"
+            );
             assert_eq!(
                 output.matches("Set disk object heal writer failures").count(),
                 1,
@@ -3879,12 +3904,172 @@ mod heal_result_report_tests {
                 "{case}: warning must distinguish partial from all-target failure: {output}"
             );
             assert!(
-                output.contains("error=drive path full"),
+                output.contains(&format!("error={injected_summary}")),
                 "{case}: warning must preserve a redacted failure reason: {output}"
             );
         }
     }
 
+    #[test]
+    fn heal_writer_failure_aggregation_retains_permanent_cause() {
+        for errors in [
+            [DiskError::Timeout, DiskError::DiskFull],
+            [DiskError::DiskFull, DiskError::Timeout],
+        ] {
+            let mut failures = HealWriterFailures::new(2);
+            for (index, error) in errors.into_iter().enumerate() {
+                failures.record(1, index, error);
+            }
+            assert_eq!(failures.count, 2);
+            assert_eq!(failures.into_error(), DiskError::DiskFull);
+        }
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn heal_second_part_failures_clean_staged_parts_and_retry() {
+        use crate::set_disk::CompletePart;
+        use crate::storage_api_contracts::multipart::MultipartOperations as _;
+        use tokio::io::AsyncReadExt as _;
+
+        let (dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+        let bucket = "heal-writer-second-part";
+        let object = "multipart.bin";
+        for disk in &disks {
+            disk.make_volume(bucket).await.expect("fixture bucket");
+        }
+        let options = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+        let upload = set
+            .new_multipart_upload(bucket, object, &options)
+            .await
+            .expect("multipart upload");
+        let payloads = [vec![0x35; 5 * 1024 * 1024], vec![0x69; 1024 * 1024]];
+        let mut completed = Vec::new();
+        for (index, payload) in payloads.iter().enumerate() {
+            let part = set
+                .put_object_part(
+                    bucket,
+                    object,
+                    &upload.upload_id,
+                    index + 1,
+                    &mut PutObjReader::from_vec(payload.clone()),
+                    &options,
+                )
+                .await
+                .expect("multipart part");
+            completed.push(CompletePart {
+                part_num: part.part_num,
+                etag: part.etag,
+                ..Default::default()
+            });
+        }
+        set.clone()
+            .complete_multipart_upload(bucket, object, &upload.upload_id, completed, &options)
+            .await
+            .expect("completed multipart object");
+        let source = disks[2]
+            .read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("healthy source metadata");
+        let data_dir = source.data_dir.expect("multipart data directory").to_string();
+        let target_slots = [source.erasure.distribution[0] - 1, source.erasure.distribution[1] - 1];
+        for dir in &dirs[..2] {
+            tokio::fs::remove_dir_all(dir.path().join(bucket).join(object).join(&data_dir))
+                .await
+                .expect("remove target shards");
+        }
+        let tmp_before = non_trash_tmp_entries(&dirs).await;
+        let scope = HealWriterFailureScope::install(bucket, object, Some(2), &target_slots, DiskError::Timeout);
+        let heal_options = HealOpts {
+            no_lock: true,
+            scan_mode: HealScanMode::Deep,
+            ..Default::default()
+        };
+        let (_, error) = set
+            .heal_object(bucket, object, "", &heal_options)
+            .await
+            .expect("failed heal report");
+        drop(scope);
+        assert_eq!(error, Some(DiskError::Timeout));
+        assert_eq!(
+            non_trash_tmp_entries(&dirs).await,
+            tmp_before,
+            "first-part staging must be cleaned after the second-part setup failure"
+        );
+
+        let source_path = dirs[2].path().join(bucket).join(object).join(&data_dir).join("part.2");
+        let source_bytes = tokio::fs::read(&source_path).await.expect("committed source shard");
+        let mut metadata_before = Vec::new();
+        for dir in &dirs {
+            metadata_before.push(
+                tokio::fs::read(dir.path().join(bucket).join(object).join("xl.meta"))
+                    .await
+                    .expect("committed metadata before reconstruction failure"),
+            );
+        }
+        let (pause_scope, arrived, release) =
+            ReadRepairPauseScope::install_at(bucket, object, super::ReadRepairPausePhase::PartStaging(2));
+        let lose_source = async {
+            arrived.notified().await;
+            assert_ne!(
+                non_trash_tmp_entries(&dirs).await,
+                tmp_before,
+                "the first part must already be staged before the source loss"
+            );
+            tokio::fs::remove_file(&source_path)
+                .await
+                .expect("lose one remaining source shard");
+            release.notify_one();
+        };
+        let (outcome, ()) = tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            tokio::join!(set.heal_object(bucket, object, "", &heal_options), lose_source)
+        })
+        .await
+        .expect("heal must report source loss without waiting indefinitely");
+        drop(pause_scope);
+        assert_eq!(
+            outcome.expect_err("source loss after writer admission must preserve the outer failure"),
+            DiskError::ErasureReadQuorum
+        );
+        assert_eq!(
+            non_trash_tmp_entries(&dirs).await,
+            tmp_before,
+            "reconstruction failure must clean every staged part"
+        );
+        for (dir, expected) in dirs.iter().zip(metadata_before) {
+            assert_eq!(
+                tokio::fs::read(dir.path().join(bucket).join(object).join("xl.meta"))
+                    .await
+                    .expect("metadata after reconstruction failure"),
+                expected,
+                "a failed data stage must not publish replacement metadata"
+            );
+        }
+        tokio::fs::write(&source_path, source_bytes)
+            .await
+            .expect("restore the committed source shard");
+
+        let (_, error) = set
+            .heal_object(bucket, object, "", &heal_options)
+            .await
+            .expect("retry after target recovery");
+        assert!(error.is_none(), "recovered targets should heal both parts: {error:?}");
+        let mut reader = set
+            .get_object_reader(bucket, object, None, Default::default(), &options)
+            .await
+            .expect("read repaired multipart object");
+        let mut actual = Vec::new();
+        reader
+            .stream
+            .read_to_end(&mut actual)
+            .await
+            .expect("complete repaired object body");
+        assert_eq!(actual, payloads.concat());
+    }
+
     async fn real_disk() -> (TempDir, Endpoint, DiskStore) {
         let dir = tempfile::tempdir().expect("tempdir should be created");
         let endpoint =

crates/ecstore/src/store/init_format.rs

@@ -134,6 +134,18 @@ pub(crate) async fn connect_load_init_formats_with_instance_ctx(
     {
         return Err(Error::CorruptedFormat);
     }
+    if matches!(format_quorum, Some(Err(Error::ErasureReadQuorum)))
+        && let Some((format, members)) = select_existing_format_read_quorum(&formats, &errs, set_drive_count)
+    {
+        retain_format_quorum_members(instance_ctx, disks, &format, &members, set_drive_count).await?;
+        // Loading an existing format grants no authority to initialize pool
+        // metadata or migrate formats. The persisted storage classes and each
+        // object's EC metadata still determine service and operation quorums.
+        return Ok(LoadedFormat {
+            format,
+            pool_meta_bootstrap_authority: PoolMetaBootstrapAuthority::None,
+        });
+    }
     let resumable_partial_migration = formats_present > 0
         && errs.iter().any(|error| matches!(error, Some(DiskError::UnformattedDisk)))
         && format_quorum.as_ref().is_some_and(Result::is_err)
@@ -578,6 +590,43 @@ pub fn get_format_erasure_in_quorum(formats: &[Option<FormatV3>], slot_offset: u
     select_format_erasure_in_quorum(formats, slot_offset).map(|(format, _)| format)
 }
 
+/// Load a consistent existing layout with enough members per set to read the
+/// max-parity system metadata containing the persisted storage classes. This
+/// is not object-read admission: STANDARD, RRS and historical objects can have
+/// different EC layouts, which are validated after this bootstrap read.
+/// Format migration and healing still use strict-majority election.
+fn select_existing_format_read_quorum(
+    formats: &[Option<FormatV3>],
+    errors: &[Option<DiskError>],
+    set_drive_count: usize,
+) -> Option<(FormatV3, Vec<bool>)> {
+    if set_drive_count == 0 || formats.is_empty() || formats.len() != errors.len() {
+        return None;
+    }
+    let metadata_parity = set_drive_count / 2;
+    let metadata_read_quorum = set_drive_count - metadata_parity;
+    let reference = formats.iter().flatten().next()?;
+    if !formats_match_reference_slots(formats, reference, 0)
+        || check_format_erasure_value_for_topology(reference, formats.len(), set_drive_count).is_err()
+        || formats.iter().zip(errors).any(|(format, error)| match (format, error) {
+            (Some(_), None) => false,
+            (None, Some(DiskError::DiskNotFound)) => false,
+            (None, Some(error)) => !is_network_like_disk_error(error),
+            _ => true,
+        })
+        || formats
+            .chunks(set_drive_count)
+            .any(|set| set.iter().flatten().count() < metadata_read_quorum)
+    {
+        return None;
+    }
+
+    let mut format = reference.clone();
+    format.erasure.this = Uuid::nil();
+    format.disk_info = None;
+    Some((format, formats.iter().map(Option::is_some).collect()))
+}
+
 pub(crate) fn select_format_erasure_in_quorum(formats: &[Option<FormatV3>], slot_offset: usize) -> Result<(FormatV3, Vec<bool>)> {
     let mut candidates = HashMap::new();
     let formats_present = formats.iter().flatten().count();
@@ -1120,6 +1169,190 @@ mod tests {
         assert!(!formats_match_reference_slots(&formats, &reference, 0));
     }
 
+    fn half_pool_formats() -> (Vec<Option<FormatV3>>, Vec<Option<DiskError>>) {
+        let reference = FormatV3::new(6, 4);
+        let formats = (0..24)
+            .map(|slot| {
+                (slot % 4 < 2).then(|| {
+                    let mut format = reference.clone();
+                    format.erasure.this = reference.erasure.sets[slot / 4][slot % 4];
+                    format
+                })
+            })
+            .collect::<Vec<_>>();
+        let errors = formats
+            .iter()
+            .map(|format| format.is_none().then_some(DiskError::DiskNotFound))
+            .collect();
+        (formats, errors)
+    }
+
+    #[test]
+    fn existing_format_read_quorum_follows_each_sets_metadata_layout() {
+        // These are system-metadata EC boundaries, independent of user-object
+        // storage classes and of how many drives are hosted by each node.
+        for (set_width, read_quorum) in [(1, 1), (2, 1), (3, 2), (4, 2), (6, 3), (8, 4), (12, 6), (16, 8)] {
+            for set_count in [1, 2, 5] {
+                let reference = FormatV3::new(set_count, set_width);
+                let mut formats = (0..set_count * set_width)
+                    .map(|slot| {
+                        let set = slot / set_width;
+                        let drive = slot % set_width;
+                        // Rotate surviving slots between sets so admission
+                        // cannot depend on the first drive or node being up.
+                        ((drive + set) % set_width < read_quorum).then(|| {
+                            let mut format = reference.clone();
+                            format.erasure.this = reference.erasure.sets[set][drive];
+                            format
+                        })
+                    })
+                    .collect::<Vec<_>>();
+                let mut errors = formats
+                    .iter()
+                    .map(|format| format.is_none().then_some(DiskError::DiskNotFound))
+                    .collect::<Vec<_>>();
+                let (loaded, members) = select_existing_format_read_quorum(&formats, &errors, set_width)
+                    .unwrap_or_else(|| panic!("metadata read quorum must load {set_count} sets of {set_width} drives"));
+                assert_eq!(loaded.shared_identity(), reference.shared_identity());
+                assert!(loaded.erasure.this.is_nil());
+                assert_eq!(members.iter().filter(|member| **member).count(), set_count * read_quorum);
+
+                // More members in other sets cannot compensate for one set
+                // falling below its own metadata reconstruction boundary.
+                for slot in 0..(set_count - 1) * set_width {
+                    let mut format = reference.clone();
+                    format.erasure.this = reference.erasure.sets[slot / set_width][slot % set_width];
+                    formats[slot] = Some(format);
+                    errors[slot] = None;
+                }
+                let lost_slot = formats
+                    .iter()
+                    .rposition(Option::is_some)
+                    .expect("at least one surviving drive");
+                formats[lost_slot] = None;
+                errors[lost_slot] = Some(DiskError::DiskNotFound);
+                assert!(
+                    select_existing_format_read_quorum(&formats, &errors, set_width).is_none(),
+                    "one set below read quorum must block {set_count} sets of {set_width} drives"
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn existing_format_read_quorum_preserves_election_and_set_boundaries() {
+        let (mut formats, mut errors) = half_pool_formats();
+        assert!(select_existing_format_read_quorum(&[], &[], 0).is_none());
+        assert!(select_existing_format_read_quorum(&formats, &errors[..23], 4).is_none());
+        assert!(select_existing_format_read_quorum(&formats, &errors, 0).is_none());
+        assert!(select_existing_format_read_quorum(&formats, &errors, 6).is_none());
+        assert!(select_existing_format_read_quorum(&formats, &errors, 4).is_some());
+        assert!(matches!(get_format_erasure_in_quorum(&formats, 0), Err(Error::ErasureReadQuorum)));
+
+        let mut extra = formats[0].as_ref().expect("seed format").clone();
+        extra.erasure.this = extra.erasure.sets[0][2];
+        formats[2] = Some(extra);
+        errors[2] = None;
+        assert!(get_format_erasure_in_quorum(&formats, 0).is_ok(), "13 votes must establish a majority");
+        formats[5] = None;
+        errors[5] = Some(DiskError::DiskNotFound);
+        assert!(
+            select_existing_format_read_quorum(&formats, &errors, 4).is_none(),
+            "12 votes cannot compensate for a set with only one readable drive"
+        );
+        formats[2] = None;
+        errors[2] = Some(DiskError::DiskNotFound);
+        assert!(select_existing_format_read_quorum(&formats, &errors, 4).is_none(), "11 votes must wait");
+    }
+
+    #[test]
+    fn existing_format_read_quorum_rejects_conflicts_invalid_slots_and_blank_disks() {
+        for failure in [
+            "conflict",
+            "wrong_slot",
+            "nil",
+            "duplicate",
+            "blank",
+            "corrupt",
+            "access_denied",
+            "tie",
+        ] {
+            let (mut formats, mut errors) = half_pool_formats();
+            match failure {
+                "conflict" => formats[1].as_mut().expect("second format").id = Uuid::new_v4(),
+                "wrong_slot" => {
+                    let first_id = formats[0].as_ref().expect("first format").erasure.this;
+                    formats[1].as_mut().expect("second format").erasure.this = first_id;
+                }
+                "nil" => {
+                    for format in formats.iter_mut().flatten() {
+                        format.id = Uuid::nil();
+                    }
+                }
+                "duplicate" => {
+                    for format in formats.iter_mut().flatten() {
+                        format.erasure.sets[5][3] = format.erasure.sets[0][0];
+                    }
+                }
+                "blank" => errors[2] = Some(DiskError::UnformattedDisk),
+                "corrupt" => errors[2] = Some(DiskError::CorruptedFormat),
+                "access_denied" => errors[2] = Some(DiskError::FileAccessDenied),
+                "tie" => {
+                    let other = FormatV3::new(6, 4);
+                    for slot in 0..24 {
+                        if formats[slot].is_none() {
+                            let mut format = other.clone();
+                            format.erasure.this = other.erasure.sets[slot / 4][slot % 4];
+                            formats[slot] = Some(format);
+                            errors[slot] = None;
+                        }
+                    }
+                }
+                _ => unreachable!(),
+            }
+            assert!(
+                select_existing_format_read_quorum(&formats, &errors, 4).is_none(),
+                "{failure} must not authorize degraded startup"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn existing_format_load_accepts_each_half_of_four_by_six_without_format_writes() {
+        let (_root, all_disks) = local_disks_with_set_width(24, 4).await;
+        let reference = FormatV3::new(6, 4);
+        for (slot, disk) in all_disks.iter().enumerate() {
+            let mut format = reference.clone();
+            format.erasure.this = reference.erasure.sets[slot / 4][slot % 4];
+            save_format_file(disk, &Some(format)).await.expect("persist existing format");
+        }
+        let before = format_bytes(&all_disks).await;
+        for first in 0..4 {
+            for second in first + 1..4 {
+                let mut disks = all_disks.clone();
+                for (slot, disk) in disks.iter_mut().enumerate() {
+                    if slot % 4 != first && slot % 4 != second {
+                        *disk = None;
+                    }
+                }
+                let loaded = connect_load_init_formats_with_instance_ctx(
+                    &Arc::new(InstanceContext::new()),
+                    first == 0,
+                    &mut disks,
+                    6,
+                    4,
+                    None,
+                )
+                .await
+                .expect("two nodes must load the existing format");
+                assert_eq!(loaded.format.shared_identity(), reference.shared_identity());
+                assert!(matches!(loaded.pool_meta_bootstrap_authority, PoolMetaBootstrapAuthority::None));
+                assert_eq!(disks.iter().flatten().count(), 12);
+                assert_eq!(format_bytes(&all_disks).await, before, "read startup must never rewrite formats");
+            }
+        }
+    }
+
     #[tokio::test]
     async fn existing_format_load_succeeds_with_a_strict_majority() {
         let (_temp_dir, mut disks) = two_local_disks_with_missing_third().await;

crates/ecstore/src/store/list_objects.rs

@@ -9221,7 +9221,7 @@ mod test {
         };
         let snapshot = PersistentListMetadataObject::from_object_info(&object);
         assert!(snapshot.to_object_info("bucket").etag.is_none());
-        let mut old_snapshot = snapshot.clone();
+        let mut old_snapshot = snapshot;
         old_snapshot.etag = object.etag.clone();
         let old_contents = format!(
             "# rustfs-listobjects-key-only-v2\n{}\n",

crates/ecstore/tests/bucket_config_goldens.rs

@@ -0,0 +1,497 @@
+// 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()
+    );
+}

crates/ecstore/tests/fixtures/bucket-config-goldens/README.md

@@ -0,0 +1,58 @@
+# 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.

crates/ecstore/tests/fixtures/bucket-config-goldens/accelerate/aws.in.xml

@@ -0,0 +1,4 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<AccelerateConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <Status>Enabled</Status>
+</AccelerateConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/accelerate/aws.out.xml

@@ -0,0 +1 @@
+<AccelerateConfiguration><Status>Enabled</Status></AccelerateConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/accelerate/minio.in.xml

@@ -0,0 +1 @@
+<AccelerateConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>Suspended</Status></AccelerateConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/accelerate/minio.out.xml

@@ -0,0 +1 @@
+<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/accelerate/rustfs.in.xml

@@ -0,0 +1 @@
+<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/accelerate/rustfs.out.xml

@@ -0,0 +1 @@
+<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/cors/aws.in.xml

@@ -0,0 +1,17 @@
+<?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>

crates/ecstore/tests/fixtures/bucket-config-goldens/cors/aws.out.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/cors/minio.in.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/cors/minio.out.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/cors/rustfs.in.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/cors/rustfs.out.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/encryption/aws.in.xml

@@ -0,0 +1,10 @@
+<?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>

crates/ecstore/tests/fixtures/bucket-config-goldens/encryption/aws.out.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/encryption/minio.in.xml

@@ -0,0 +1 @@
+<ServerSideEncryptionConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm></ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/encryption/minio.out.xml

@@ -0,0 +1 @@
+<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm></ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/encryption/rustfs.in.xml

@@ -0,0 +1 @@
+<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><KMSMasterKeyID>rustfs-bucket-key</KMSMasterKeyID><SSEAlgorithm>aws:kms</SSEAlgorithm></ApplyServerSideEncryptionByDefault><BucketKeyEnabled>false</BucketKeyEnabled></Rule></ServerSideEncryptionConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/encryption/rustfs.out.xml

@@ -0,0 +1 @@
+<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><KMSMasterKeyID>rustfs-bucket-key</KMSMasterKeyID><SSEAlgorithm>aws:kms</SSEAlgorithm></ApplyServerSideEncryptionByDefault><BucketKeyEnabled>false</BucketKeyEnabled></Rule></ServerSideEncryptionConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/lifecycle/aws.in.xml

@@ -0,0 +1,49 @@
+<?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>

crates/ecstore/tests/fixtures/bucket-config-goldens/lifecycle/aws.out.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/lifecycle/minio.in.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/lifecycle/minio.out.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/lifecycle/rustfs.in.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/lifecycle/rustfs.out.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/logging/aws.in.xml

@@ -0,0 +1,20 @@
+<?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>

crates/ecstore/tests/fixtures/bucket-config-goldens/logging/aws.out.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/logging/minio.in.xml

@@ -0,0 +1 @@
+<BucketLoggingStatus xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><LoggingEnabled><TargetBucket>interop-logs</TargetBucket><TargetPrefix>interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>

crates/ecstore/tests/fixtures/bucket-config-goldens/logging/minio.out.xml

@@ -0,0 +1 @@
+<BucketLoggingStatus><LoggingEnabled><TargetBucket>interop-logs</TargetBucket><TargetPrefix>interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>

crates/ecstore/tests/fixtures/bucket-config-goldens/logging/rustfs.in.xml

@@ -0,0 +1 @@
+<BucketLoggingStatus><LoggingEnabled><TargetBucket>rustfs-access-logs</TargetBucket><TargetPrefix>buckets/interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>

crates/ecstore/tests/fixtures/bucket-config-goldens/logging/rustfs.out.xml

@@ -0,0 +1 @@
+<BucketLoggingStatus><LoggingEnabled><TargetBucket>rustfs-access-logs</TargetBucket><TargetPrefix>buckets/interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>

crates/ecstore/tests/fixtures/bucket-config-goldens/notification/aws.in.xml

@@ -0,0 +1,30 @@
+<?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>

crates/ecstore/tests/fixtures/bucket-config-goldens/notification/aws.out.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/notification/minio.in.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/notification/minio.out.xml

@@ -0,0 +1 @@
+<NotificationConfiguration><QueueConfiguration><Event>s3:ObjectCreated:*</Event><Id></Id><Queue>arn:minio:sqs::primary:webhook</Queue></QueueConfiguration></NotificationConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/notification/rustfs.in.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/notification/rustfs.out.xml

@@ -0,0 +1 @@
+<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>

crates/ecstore/tests/fixtures/bucket-config-goldens/object_lock/aws.in.xml

@@ -0,0 +1,10 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<ObjectLockConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <ObjectLockEnabled>Enabled</ObjectLockEnabled>
+  <Rule>
+    <DefaultRetention>
+      <Mode>COMPLIANCE</Mode>
+      <Years>1</Years>
+    </DefaultRetention>
+  </Rule>
+</ObjectLockConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/object_lock/aws.out.xml

@@ -0,0 +1 @@
+<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Mode>COMPLIANCE</Mode><Years>1</Years></DefaultRetention></Rule></ObjectLockConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/object_lock/minio.in.xml

@@ -0,0 +1 @@
+<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Mode>GOVERNANCE</Mode><Days>7</Days></DefaultRetention></Rule></ObjectLockConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/object_lock/minio.out.xml

@@ -0,0 +1 @@
+<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Days>7</Days><Mode>GOVERNANCE</Mode></DefaultRetention></Rule></ObjectLockConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/object_lock/rustfs.in.xml

@@ -0,0 +1 @@
+<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Days>30</Days><Mode>GOVERNANCE</Mode></DefaultRetention></Rule></ObjectLockConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/object_lock/rustfs.out.xml

@@ -0,0 +1 @@
+<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Days>30</Days><Mode>GOVERNANCE</Mode></DefaultRetention></Rule></ObjectLockConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/public_access_block/aws.in.xml

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<PublicAccessBlockConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <BlockPublicAcls>TRUE</BlockPublicAcls>
+  <IgnorePublicAcls>TRUE</IgnorePublicAcls>
+  <BlockPublicPolicy>TRUE</BlockPublicPolicy>
+  <RestrictPublicBuckets>TRUE</RestrictPublicBuckets>
+</PublicAccessBlockConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/public_access_block/aws.out.xml

@@ -0,0 +1 @@
+<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><BlockPublicPolicy>true</BlockPublicPolicy><IgnorePublicAcls>true</IgnorePublicAcls><RestrictPublicBuckets>true</RestrictPublicBuckets></PublicAccessBlockConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/public_access_block/minio.in.xml

@@ -0,0 +1 @@
+<PublicAccessBlockConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><BlockPublicAcls>true</BlockPublicAcls><IgnorePublicAcls>false</IgnorePublicAcls><BlockPublicPolicy>true</BlockPublicPolicy><RestrictPublicBuckets>false</RestrictPublicBuckets></PublicAccessBlockConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/public_access_block/minio.out.xml

@@ -0,0 +1 @@
+<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><BlockPublicPolicy>true</BlockPublicPolicy><IgnorePublicAcls>false</IgnorePublicAcls><RestrictPublicBuckets>false</RestrictPublicBuckets></PublicAccessBlockConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/public_access_block/rustfs.in.xml

@@ -0,0 +1 @@
+<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><BlockPublicPolicy>false</BlockPublicPolicy><IgnorePublicAcls>true</IgnorePublicAcls><RestrictPublicBuckets>false</RestrictPublicBuckets></PublicAccessBlockConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/public_access_block/rustfs.out.xml

@@ -0,0 +1 @@
+<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><BlockPublicPolicy>false</BlockPublicPolicy><IgnorePublicAcls>true</IgnorePublicAcls><RestrictPublicBuckets>false</RestrictPublicBuckets></PublicAccessBlockConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/replication/aws.in.xml

@@ -0,0 +1,12 @@
+
+<ReplicationConfiguration>
+  <Role></Role>
+  <FutureTopLevel>future</FutureTopLevel>
+  <Rule>
+    <ID>unknown</ID>
+    <Status>Enabled</Status>
+    <Destination>
+      <Bucket>arn:aws:s3:::destination</Bucket>
+    </Destination>
+  </Rule>
+</ReplicationConfiguration>

crates/ecstore/tests/fixtures/bucket-config-goldens/replication/aws.out.xml

@@ -0,0 +1 @@
+<ReplicationConfiguration><Role></Role><Rule><Destination><Bucket>arn:aws:s3:::destination</Bucket></Destination><ID>unknown</ID><Status>Enabled</Status></Rule></ReplicationConfiguration>