test(ecstore): trace the first unknown native publication cause

Record the first transition in test builds using existing lock and identity observations. Preserve the producer, rollback assertions, workflow, and test selection. Validation: cargo fmt --all --check; cargo check -p rustfs-ecstore --tests; all 19 original quick-check gates, including make script-tests and both Python checks. Linux diagnostic execution is pending.
test(ecstore): trace bucket creation proof rejection
2026-10-27 14:27:10 +00:00 · 2026-10-11 22:32:01 +08:00 · 2026-10-11 19:53:56 +08:00 · 2026-10-11 16:16:31 +08:00 · 2026-10-11 00:55:29 -07:00 · 2026-10-11 12:34:14 +08:00
201 changed files with 17480 additions and 3383 deletions
@@ -1,2 +1,2 @@
-sha256-darwin=98dff0961c6eb21d6128bfd2d4a6100a624ddf1fb7db05dba360a5d3d0a310c2
-sha256-linux=890f5925b87e1bf3710c356253f3b2276d901e156df9ccc04ec9315374de6d22
+sha256-darwin=b9438fd1b82558f1fcd7e5a19dc0ceb0b34b5357d4ce99b7033de3ef019333ab
+sha256-linux=d4622517c13283247a022f49962c2cdb69e5ed0e13874fb55e0ab34fb923c044
@@ -1 +1 @@
-sha256=ba8f392b617cf9f2338142233a417e43927b28096635e628233ce8a60cf49d10
+sha256=58df0d240cd74b916ce8f094ed1d8ad3e136f2f9bac157a155192a8786dae65c
@@ -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..."
@@ -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
@@ -50,7 +50,9 @@ 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_migration_gate_evidence.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/ci_gate.py --self-test
@@ -579,6 +579,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) & (
@@ -586,6 +594,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)$/)
    )
@@ -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/**
@@ -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=1584
+ecstore_files=36
@@ -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	890	18
+rustfs	5540	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	6306	1
+rustfs-extension-schema	23	0
+rustfs-filemeta	306	0
+rustfs-heal	839	3
+rustfs-heal-contracts	6	0
+rustfs-iam	371	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	87	0
+rustfs-madmin	209	0
+rustfs-notify	136	0
+rustfs-object-capacity	92	0
+rustfs-object-data-cache	140	0
+rustfs-obs	365	0
+rustfs-policy	677	0
+rustfs-protocols	736	1
+rustfs-protos	52	0
+rustfs-replication	200	0
+rustfs-rio	198	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	971	20
+rustfs-scanner-metrics	75	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	216	0
+rustfs-zip	30	0
@@ -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
@@ -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/**"
@@ -780,6 +782,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 +902,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
@@ -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`.

@@ -10284,7 +10284,6 @@ dependencies = [
 "reqwest",
 "rmp-serde",
 "rustfs-signer",
- "s3s",
 "serde",
 "serde_json",
 "sysinfo",
@@ -10509,6 +10508,7 @@ dependencies = [
 "rustfs-keystone",
 "rustfs-policy",
 "rustfs-rio",
+ "rustfs-s3-types",
 "rustfs-storage-api",
 "rustfs-test-utils",
 "rustfs-tls-runtime",
@@ -10523,6 +10523,7 @@ dependencies = [
 "sha2 0.11.0",
 "socket2",
 "subtle",
+ "sync_wrapper",
 "tempfile",
 "thiserror 2.0.21",
 "time",
@@ -10669,6 +10670,7 @@ dependencies = [
 "rustfs-checksums",
 "rustfs-config",
 "rustfs-rio",
+ "rustfs-s3-types",
 "rustfs-signer",
 "rustfs-storage-api",
 "rustfs-tls-runtime",
@@ -10702,9 +10704,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]]
@@ -10728,6 +10740,7 @@ dependencies = [
 "parking_lot",
 "rustfs-common",
 "rustfs-ecstore",
+ "rustfs-s3-types",
 "rustfs-storage-api",
 "rustfs-test-utils",
 "s3s",
@@ -10841,9 +10854,10 @@ dependencies = [
 "bytes",
 "hotpath",
 "http 1.5.0",
+ "http-body",
 "hyper",
+ "rustfs-s3-types",
 "rustfs-utils",
- "s3s",
 "serde_urlencoded",
 "thiserror 2.0.21",
 "time",
@@ -10901,7 +10915,6 @@ dependencies = [
 "rustfs-utils",
 "rustls",
 "rustls-native-certs",
- "s3s",
 "serde",
 "serde_json",
 "sha2 0.11.0",
@@ -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"
@@ -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",
+            }])
+        );
+    }
+}
@@ -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)
+        );
+    }
+}
@@ -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.
@@ -14196,7 +14196,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;
@@ -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);
    }
@@ -1553,9 +1553,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>,
@@ -1599,6 +1617,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()),
            }),
@@ -1629,6 +1648,105 @@ impl BucketMetadataSys {
            .expect("bucket metadata object store should still be live")
    }

+    pub(crate) async fn pending_bucket_creation(&self, bucket: &str) -> Option<PendingBucketCreationProof> {
+        let observed = self.pending_bucket_creations.read().await.get(bucket).cloned();
+        #[cfg(test)]
+        if bucket.starts_with("real-guard-loss-") {
+            eprintln!(
+                "[creation-proof-diag] bucket={bucket} stage=pending-read present={} map={:p}",
+                observed.is_some(),
+                Arc::as_ptr(&self.pending_bucket_creations)
+            );
+        }
+        observed
+    }
+
+    pub(crate) async fn grant_pending_bucket_creation(&self, bucket: &str, proof: PendingBucketCreationProof) {
+        self.pending_bucket_creations.write().await.insert(bucket.to_owned(), proof);
+        #[cfg(test)]
+        if bucket.starts_with("real-guard-loss-") {
+            eprintln!(
+                "[creation-proof-diag] bucket={bucket} stage=pending-inserted map={:p}",
+                Arc::as_ptr(&self.pending_bucket_creations)
+            );
+        }
+    }
+
+    pub(crate) async fn revoke_pending_bucket_creation(&self, bucket: &str) {
+        self.pending_bucket_creations.write().await.remove(bucket).inspect(|_| {
+            #[cfg(test)]
+            if bucket.starts_with("real-guard-loss-") {
+                eprintln!(
+                    "[creation-proof-diag] bucket={bucket} stage=pending-removed-by-revoke map={:p}",
+                    Arc::as_ptr(&self.pending_bucket_creations)
+                );
+            }
+        });
+    }
+
+    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()
    }
@@ -1819,6 +1937,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,
@@ -1902,6 +2021,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")
@@ -1963,6 +2083,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;
@@ -1984,6 +2105,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;
@@ -2001,6 +2123,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);
@@ -2139,6 +2262,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;
@@ -2147,6 +2271,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?;
@@ -2156,7 +2281,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;
@@ -2166,6 +2293,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
@@ -2214,6 +2342,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")
@@ -2270,6 +2399,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)]
@@ -2578,6 +2708,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))
@@ -2655,6 +2786,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}")));
@@ -3026,6 +3158,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;
@@ -3080,6 +3213,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;
@@ -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)),
@@ -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() {
@@ -2911,7 +2920,7 @@ mod tests {

        let timed = wrapper.track_disk_health_with_op(
            "write_all",
-            || std::future::pending::<Result<()>>(),
+            std::future::pending::<Result<()>>,
            SLOW_DISK_OPERATION_THRESHOLD + Duration::from_secs(1),
        );
        tokio::pin!(timed);
@@ -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)
 }

@@ -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,45 @@ 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 {
+            #[cfg(test)]
+            if volume.starts_with("real-guard-loss-") {
+                eprintln!("[creation-proof-diag] bucket={volume} stage=local-history-unavailable");
+            }
+            return Ok(None);
+        };
+        let observed = self.get_disk_id().await;
+        #[cfg(test)]
+        if volume.starts_with("real-guard-loss-")
+            && let Err(error) = &observed
+        {
+            eprintln!(
+                "[creation-proof-diag] bucket={volume} stage=local-disk-id-error discriminant={:?}",
+                std::mem::discriminant(error)
+            );
+        }
+        let Some(disk_id) = observed? else {
+            #[cfg(test)]
+            if volume.starts_with("real-guard-loss-") {
+                eprintln!("[creation-proof-diag] bucket={volume} stage=local-disk-id-unavailable");
+            }
+            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 +11606,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() {
@@ -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>>;
@@ -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,424 @@ 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>,
+    #[cfg(test)]
+    first_unknown: Option<NativeFirstUnknown>,
+}
+
+#[cfg(test)]
+#[derive(Debug)]
+struct NativeFirstUnknown {
+    reason: &'static str,
+    aliases: [Option<PathBuf>; 2],
+    old_identity: Option<NativeRootIdentity>,
+    observed_identities: [Option<NativeRootIdentity>; 2],
+}
+
+#[cfg(test)]
+impl NativeRootHistory {
+    fn record_first_unknown(
+        &mut self,
+        was_unknown: bool,
+        reason: &'static str,
+        aliases: [Option<&Path>; 2],
+        old_identity: Option<NativeRootIdentity>,
+        observed_identities: [Option<NativeRootIdentity>; 2],
+    ) {
+        if !was_unknown && self.first_unknown.is_none() {
+            self.first_unknown = Some(NativeFirstUnknown {
+                reason,
+                aliases: aliases.map(|alias| alias.map(Path::to_path_buf)),
+                old_identity,
+                observed_identities,
+            });
+        }
+    }
+}
+
+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,
+    #[cfg(test)]
+    diagnostic_previous_unknown: Option<bool>,
+}
+
+/// 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();
+    #[cfg(test)]
+    let diagnostic_previous_unknown = history.roots.get(&identity).map(|root| root.unknown);
+    history.roots.entry(identity).or_default();
+    for alias in [root, io_root] {
+        let alias: PathBuf = alias.components().collect();
+        #[cfg(test)]
+        let diagnostic_alias = alias.clone();
+        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() {
+            #[cfg(test)]
+            let diagnostic_was_unknown = history.roots.get(&identity).is_some_and(|root| root.unknown);
+            history.roots.get_mut(&identity)?.unknown = true;
+            #[cfg(test)]
+            if let Some(root) = history.roots.get_mut(&identity) {
+                root.record_first_unknown(
+                    diagnostic_was_unknown,
+                    "registration-alias-collision",
+                    [Some(&diagnostic_alias), None],
+                    collisions.first().copied(),
+                    [Some(identity), Some(identity)],
+                );
+            }
+            for old in collisions {
+                #[cfg(test)]
+                let diagnostic_was_unknown = history.roots.get(&old).is_some_and(|root| root.unknown);
+                history.roots.get_mut(&old)?.unknown = true;
+                #[cfg(test)]
+                if let Some(root) = history.roots.get_mut(&old) {
+                    root.record_first_unknown(
+                        diagnostic_was_unknown,
+                        "registration-alias-collision",
+                        [Some(&diagnostic_alias), None],
+                        Some(old),
+                        [Some(identity), Some(identity)],
+                    );
+                }
+            }
+        }
+    }
+    Some(NativePublicationRegistration {
+        identity,
+        root: root.to_path_buf(),
+        io_root: io_root.to_path_buf(),
+        #[cfg(test)]
+        diagnostic_previous_unknown,
+    })
+}
+
+/// 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) {
+            #[cfg(test)]
+            let diagnostic_old_identity = identities.iter().find(|identity| Some(**identity) != physical).copied();
+            for identity in identities {
+                #[cfg(test)]
+                let diagnostic_was_unknown = history.roots.get(&identity).is_some_and(|root| root.unknown);
+                history.roots.entry(identity).or_default().unknown = true;
+                #[cfg(test)]
+                if let Some(root) = history.roots.get_mut(&identity) {
+                    root.record_first_unknown(
+                        diagnostic_was_unknown,
+                        "publication-alias-identity-mismatch",
+                        [Some(&alias), None],
+                        Some(identity),
+                        [physical, None],
+                    );
+                }
+            }
+            if let Some(identity) = physical {
+                #[cfg(test)]
+                let diagnostic_was_unknown = history.roots.get(&identity).is_some_and(|root| root.unknown);
+                history.roots.entry(identity).or_default().unknown = true;
+                #[cfg(test)]
+                if let Some(root) = history.roots.get_mut(&identity) {
+                    root.record_first_unknown(
+                        diagnostic_was_unknown,
+                        "publication-alias-identity-mismatch",
+                        [Some(&alias), None],
+                        diagnostic_old_identity,
+                        [physical, None],
+                    );
+                }
+            }
+        } else if let Some(identity) = physical {
+            let root = history.roots.entry(identity).or_default();
+            #[cfg(test)]
+            let diagnostic_was_unknown = root.unknown;
+            #[cfg(test)]
+            let diagnostic_invalid_bucket = bucket.is_none();
+            match bucket {
+                Some(bucket) => {
+                    root.started.insert(bucket);
+                }
+                None => root.unknown = true,
+            }
+            #[cfg(test)]
+            if diagnostic_invalid_bucket {
+                root.record_first_unknown(
+                    diagnostic_was_unknown,
+                    "publication-invalid-relative-bucket",
+                    [Some(&alias), None],
+                    Some(identity),
+                    [physical, None],
+                );
+            }
+        }
+    }
+}
+
+impl NativePublicationRegistration {
+    pub(crate) fn invalidate(&self) {
+        if let Some(root) = NATIVE_PUBLICATION_HISTORY.lock().roots.get_mut(&self.identity) {
+            #[cfg(test)]
+            let diagnostic_was_unknown = root.unknown;
+            root.unknown = true;
+            #[cfg(test)]
+            root.record_first_unknown(
+                diagnostic_was_unknown,
+                "registration-invalidate",
+                [Some(&self.root), Some(&self.io_root)],
+                Some(self.identity),
+                [None, None],
+            );
+        }
+    }
+
+    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
+        {
+            #[cfg(test)]
+            if bucket.starts_with("real-guard-loss-") {
+                eprintln!("[creation-proof-diag] bucket={bucket} stage=native-input-rejected");
+            }
+            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 {
+    #[cfg(test)]
+    fn log_first_unknown(&self, root: &NativeRootHistory) {
+        if !self.bucket.starts_with("real-guard-loss-") {
+            return;
+        }
+        let first_unknown = root
+            .first_unknown
+            .as_ref()
+            .map(|cause| (cause.reason, &cause.aliases, cause.old_identity, cause.observed_identities));
+        eprintln!(
+            "[creation-proof-diag] bucket={} stage=native-history-first-unknown key={:?} first_unknown={first_unknown:?} configured_root={:?} configured_io_root={:?} registration_previous_unknown={:?}",
+            self.bucket,
+            self.registration.identity,
+            self.registration.root,
+            self.registration.io_root,
+            self.registration.diagnostic_previous_unknown,
+        );
+    }
+
+    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);
+        #[cfg(test)]
+        if self.bucket.starts_with("real-guard-loss-") {
+            eprintln!(
+                "[creation-proof-diag] bucket={} stage=native-physical-root expected={:?} observed_root={root:?} observed_io_root={io_root:?} backend={:p}",
+                self.bucket,
+                self.registration.identity,
+                Arc::as_ptr(&self.backend)
+            );
+        }
+        if root == Some(self.registration.identity) && io_root == Some(self.registration.identity) {
+            return true;
+        }
+        let mut history = NATIVE_PUBLICATION_HISTORY.lock();
+        #[cfg(test)]
+        let diagnostic_was_unknown = history
+            .roots
+            .get(&self.registration.identity)
+            .is_some_and(|root| root.unknown);
+        history.roots.entry(self.registration.identity).or_default().unknown = true;
+        #[cfg(test)]
+        if let Some(state) = history.roots.get_mut(&self.registration.identity) {
+            state.record_first_unknown(
+                diagnostic_was_unknown,
+                "witness-physical-root-drift",
+                [Some(&self.registration.root), Some(&self.registration.io_root)],
+                Some(self.registration.identity),
+                [root, io_root],
+            );
+        }
+        for identity in [root, io_root].into_iter().flatten() {
+            #[cfg(test)]
+            let diagnostic_was_unknown = history.roots.get(&identity).is_some_and(|root| root.unknown);
+            history.roots.entry(identity).or_default().unknown = true;
+            #[cfg(test)]
+            if let Some(state) = history.roots.get_mut(&identity) {
+                state.record_first_unknown(
+                    diagnostic_was_unknown,
+                    "witness-physical-root-drift",
+                    [Some(&self.registration.root), Some(&self.registration.io_root)],
+                    Some(self.registration.identity),
+                    [root, io_root],
+                );
+            }
+        }
+        #[cfg(test)]
+        if let Some(state) = history.roots.get(&self.registration.identity) {
+            self.log_first_unknown(state);
+        }
+        false
+    }
+
+    fn valid_in(&self, history: &NativePublicationHistory) -> bool {
+        let Some(root) = history.roots.get(&self.registration.identity) else {
+            #[cfg(test)]
+            if self.bucket.starts_with("real-guard-loss-") {
+                eprintln!("[creation-proof-diag] bucket={} stage=native-history-missing", self.bucket);
+            }
+            return false;
+        };
+        let known = !root.unknown;
+        #[cfg(test)]
+        self.log_first_unknown(root);
+        #[cfg(test)]
+        if self.bucket.starts_with("real-guard-loss-") {
+            eprintln!("[creation-proof-diag] bucket={} stage=native-history-known known={known}", self.bucket);
+        }
+        known && {
+            let unstarted = !root.started.contains(&self.bucket);
+            #[cfg(test)]
+            if self.bucket.starts_with("real-guard-loss-") {
+                eprintln!(
+                    "[creation-proof-diag] bucket={} stage=native-history-unstarted unstarted={unstarted}",
+                    self.bucket
+                );
+            }
+            unstarted
+        }
+    }
+}
+
+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 +2509,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 +2531,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 +2554,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 +2566,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 +2591,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 +2836,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 +2891,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 +2915,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 +3318,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 +3367,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 +4656,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 +6161,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 +8251,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"
+        );
+    }
+}
@@ -17,8 +17,8 @@
 use crate::bucket::error::BucketMetadataError;
 use crate::disk::error::DiskError;
 use crate::storage_api_contracts::{error::StorageErrorCode, range::HTTPRangeError};
+use rustfs_s3_types::S3ErrorCode;
 use rustfs_utils::path::decode_dir_object;
-use s3s::S3ErrorCode;

 pub type Error = StorageError;
 pub type Result<T> = core::result::Result<T, Error>;
@@ -29,7 +29,7 @@ use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions};
 use crate::storage_api_contracts::range::HTTPRangeSpec;
 use rustfs_filemeta::ObjectPartInfo;
 use rustfs_rio::HashReader;
-use 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
@@ -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},
@@ -2368,8 +2368,7 @@ fn list_objects_paginate_versions<'a>(
    disk_has_more: bool,
    cache_id: Option<&str>,
    marker: Option<&str>,
-    last_scanned_key: Option<&str>,
-    raw_keys: impl Iterator<Item = &'a str>,
+    (last_scanned_key, raw_keys): (Option<&str>, impl Iterator<Item = &'a str>),
 ) -> (Vec<ObjectInfo>, Vec<String>, bool, Option<String>, Option<String>) {
    filter_versions_common_prefixes_after_marker(&mut get_objects, delimiter.as_deref(), marker);
    let prefix_at_page_boundary =
@@ -4331,8 +4330,7 @@ impl ECStore {
            disk_has_more,
            next_cache_id.as_deref(),
            opts.marker.as_deref(),
-            last_scanned_key.as_deref(),
-            entries.entries().iter().map(|entry| entry.name.as_str()),
+            (last_scanned_key.as_deref(), entries.entries().iter().map(|entry| entry.name.as_str())),
        );

        Ok(ListObjectVersionsInfo {
@@ -5759,8 +5757,7 @@ impl Sets {
            disk_has_more,
            next_cache_id.as_deref(),
            opts.marker.as_deref(),
-            last_scanned_key.as_deref(),
-            entries.entries().iter().map(|entry| entry.name.as_str()),
+            (last_scanned_key.as_deref(), entries.entries().iter().map(|entry| entry.name.as_str())),
        );

        Ok(ListObjectVersionsInfo {
@@ -6573,8 +6570,7 @@ impl SetDisks {
            disk_has_more,
            next_cache_id.as_deref(),
            opts.marker.as_deref(),
-            last_scanned_key.as_deref(),
-            entries.entries().iter().map(|entry| entry.name.as_str()),
+            (last_scanned_key.as_deref(), entries.entries().iter().map(|entry| entry.name.as_str())),
        );

        Ok(ListObjectVersionsInfo {
@@ -8853,8 +8849,7 @@ mod test {
                disk_has_more,
                None,
                marker.as_deref(),
-                last_scanned.as_deref(),
-                window.iter().map(String::as_str),
+                (last_scanned.as_deref(), window.iter().map(String::as_str)),
            );

            assert!(
@@ -8932,8 +8927,7 @@ mod test {
                disk_has_more,
                None,
                marker.as_deref(),
-                last_scanned.as_deref(),
-                window.iter().map(String::as_str),
+                (last_scanned.as_deref(), window.iter().map(String::as_str)),
            );

            assert_eq!(objects.len() + prefixes.len(), 1, "a page must respect max-keys");
@@ -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()
+    );
+}
@@ -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.
@@ -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>
@@ -0,0 +1 @@
+<AccelerateConfiguration><Status>Enabled</Status></AccelerateConfiguration>
@@ -0,0 +1 @@
+<AccelerateConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>Suspended</Status></AccelerateConfiguration>
@@ -0,0 +1 @@
+<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>
@@ -0,0 +1 @@
+<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>
@@ -0,0 +1 @@
+<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>
@@ -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>
@@ -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>
@@ -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>
@@ -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>
@@ -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>
@@ -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>
@@ -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>
@@ -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>
@@ -0,0 +1 @@
+<ServerSideEncryptionConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm></ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>
@@ -0,0 +1 @@
+<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm></ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>
@@ -0,0 +1 @@
+<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><KMSMasterKeyID>rustfs-bucket-key</KMSMasterKeyID><SSEAlgorithm>aws:kms</SSEAlgorithm></ApplyServerSideEncryptionByDefault><BucketKeyEnabled>false</BucketKeyEnabled></Rule></ServerSideEncryptionConfiguration>
@@ -0,0 +1 @@
+<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><KMSMasterKeyID>rustfs-bucket-key</KMSMasterKeyID><SSEAlgorithm>aws:kms</SSEAlgorithm></ApplyServerSideEncryptionByDefault><BucketKeyEnabled>false</BucketKeyEnabled></Rule></ServerSideEncryptionConfiguration>
@@ -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>
@@ -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>
@@ -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>
@@ -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>
@@ -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>
@@ -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>
@@ -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>
@@ -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>
@@ -0,0 +1 @@
+<BucketLoggingStatus xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><LoggingEnabled><TargetBucket>interop-logs</TargetBucket><TargetPrefix>interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -0,0 +1 @@
+<BucketLoggingStatus><LoggingEnabled><TargetBucket>interop-logs</TargetBucket><TargetPrefix>interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -0,0 +1 @@
+<BucketLoggingStatus><LoggingEnabled><TargetBucket>rustfs-access-logs</TargetBucket><TargetPrefix>buckets/interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -0,0 +1 @@
+<BucketLoggingStatus><LoggingEnabled><TargetBucket>rustfs-access-logs</TargetBucket><TargetPrefix>buckets/interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -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>
@@ -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>
@@ -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>
@@ -0,0 +1 @@
+<NotificationConfiguration><QueueConfiguration><Event>s3:ObjectCreated:*</Event><Id></Id><Queue>arn:minio:sqs::primary:webhook</Queue></QueueConfiguration></NotificationConfiguration>
@@ -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>
@@ -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>
@@ -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>
@@ -0,0 +1 @@
+<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Mode>COMPLIANCE</Mode><Years>1</Years></DefaultRetention></Rule></ObjectLockConfiguration>
@@ -0,0 +1 @@
+<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Mode>GOVERNANCE</Mode><Days>7</Days></DefaultRetention></Rule></ObjectLockConfiguration>
@@ -0,0 +1 @@
+<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Days>7</Days><Mode>GOVERNANCE</Mode></DefaultRetention></Rule></ObjectLockConfiguration>
@@ -0,0 +1 @@
+<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Days>30</Days><Mode>GOVERNANCE</Mode></DefaultRetention></Rule></ObjectLockConfiguration>
@@ -0,0 +1 @@
+<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Days>30</Days><Mode>GOVERNANCE</Mode></DefaultRetention></Rule></ObjectLockConfiguration>
@@ -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>
@@ -0,0 +1 @@
+<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><BlockPublicPolicy>true</BlockPublicPolicy><IgnorePublicAcls>true</IgnorePublicAcls><RestrictPublicBuckets>true</RestrictPublicBuckets></PublicAccessBlockConfiguration>
@@ -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>
@@ -0,0 +1 @@
+<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><BlockPublicPolicy>true</BlockPublicPolicy><IgnorePublicAcls>false</IgnorePublicAcls><RestrictPublicBuckets>false</RestrictPublicBuckets></PublicAccessBlockConfiguration>
@@ -0,0 +1 @@
+<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><BlockPublicPolicy>false</BlockPublicPolicy><IgnorePublicAcls>true</IgnorePublicAcls><RestrictPublicBuckets>false</RestrictPublicBuckets></PublicAccessBlockConfiguration>
@@ -0,0 +1 @@
+<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><BlockPublicPolicy>false</BlockPublicPolicy><IgnorePublicAcls>true</IgnorePublicAcls><RestrictPublicBuckets>false</RestrictPublicBuckets></PublicAccessBlockConfiguration>
@@ -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>
@@ -0,0 +1 @@
+<ReplicationConfiguration><Role></Role><Rule><Destination><Bucket>arn:aws:s3:::destination</Bucket></Destination><ID>unknown</ID><Status>Enabled</Status></Rule></ReplicationConfiguration>
@@ -0,0 +1 @@
+<ReplicationConfiguration><Rule><ID>d96i4m09k8h2vldifkag</ID><Status>Enabled</Status><Priority>1</Priority><DeleteMarkerReplication><Status>Enabled</Status></DeleteMarkerReplication><DeleteReplication><Status>Enabled</Status></DeleteReplication><Destination><Bucket>arn:minio:replication::ef5859af-120a-4218-94b5-be23470f3c60:interop-dr</Bucket></Destination><SourceSelectionCriteria><ReplicaModifications><Status>Enabled</Status></ReplicaModifications></SourceSelectionCriteria><Filter><Prefix></Prefix></Filter><ExistingObjectReplication><Status>Enabled</Status></ExistingObjectReplication></Rule><Role></Role></ReplicationConfiguration>
@@ -0,0 +1 @@
+<ReplicationConfiguration><Role></Role><Rule><DeleteMarkerReplication><Status>Enabled</Status></DeleteMarkerReplication><DeleteReplication><Status>Enabled</Status></DeleteReplication><Destination><Bucket>arn:minio:replication::ef5859af-120a-4218-94b5-be23470f3c60:interop-dr</Bucket></Destination><ExistingObjectReplication><Status>Enabled</Status></ExistingObjectReplication><Filter><Prefix></Prefix></Filter><ID>d96i4m09k8h2vldifkag</ID><Priority>1</Priority><SourceSelectionCriteria><ReplicaModifications><Status>Enabled</Status></ReplicaModifications></SourceSelectionCriteria><Status>Enabled</Status></Rule></ReplicationConfiguration>
@@ -0,0 +1 @@
+<ReplicationConfiguration><Role></Role><Rule><DeleteMarkerReplication><Status>Enabled</Status></DeleteMarkerReplication><DeleteReplication><Status>Enabled</Status></DeleteReplication><Destination><Bucket>arn:rustfs:replication::0b6f3c2a-5e6a-4d2b-9c1e-7a8f9d0e1b2c:interop-dr</Bucket><StorageClass>STANDARD</StorageClass></Destination><ExistingObjectReplication><Status>Enabled</Status></ExistingObjectReplication><Filter><And><Prefix>docs/</Prefix><Tag><Key>replicate</Key><Value>yes</Value></Tag></And></Filter><ID>rustfs-dr</ID><Priority>1</Priority><SourceSelectionCriteria><ReplicaModifications><Status>Enabled</Status></ReplicaModifications></SourceSelectionCriteria><Status>Enabled</Status></Rule></ReplicationConfiguration>
@@ -0,0 +1 @@
+<ReplicationConfiguration><Role></Role><Rule><DeleteMarkerReplication><Status>Enabled</Status></DeleteMarkerReplication><DeleteReplication><Status>Enabled</Status></DeleteReplication><Destination><Bucket>arn:rustfs:replication::0b6f3c2a-5e6a-4d2b-9c1e-7a8f9d0e1b2c:interop-dr</Bucket><StorageClass>STANDARD</StorageClass></Destination><ExistingObjectReplication><Status>Enabled</Status></ExistingObjectReplication><Filter><And><Prefix>docs/</Prefix><Tag><Key>replicate</Key><Value>yes</Value></Tag></And></Filter><ID>rustfs-dr</ID><Priority>1</Priority><SourceSelectionCriteria><ReplicaModifications><Status>Enabled</Status></ReplicaModifications></SourceSelectionCriteria><Status>Enabled</Status></Rule></ReplicationConfiguration>
@@ -0,0 +1,4 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<RequestPaymentConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <Payer>Requester</Payer>
+</RequestPaymentConfiguration>
@@ -0,0 +1 @@
+<RequestPaymentConfiguration><Payer>Requester</Payer></RequestPaymentConfiguration>
@@ -0,0 +1 @@
+<RequestPaymentConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Payer>BucketOwner</Payer></RequestPaymentConfiguration>
@@ -0,0 +1 @@
+<RequestPaymentConfiguration><Payer>BucketOwner</Payer></RequestPaymentConfiguration>
@@ -0,0 +1 @@
+<RequestPaymentConfiguration><Payer>BucketOwner</Payer></RequestPaymentConfiguration>
@@ -0,0 +1 @@
+<RequestPaymentConfiguration><Payer>BucketOwner</Payer></RequestPaymentConfiguration>
@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<Tagging xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <TagSet>
+    <Tag>
+      <Key>Project</Key>
+      <Value>Project One</Value>
+    </Tag>
+    <Tag>
+      <Key>User</Key>
+      <Value>jsmith</Value>
+    </Tag>
+  </TagSet>
+</Tagging>
@@ -0,0 +1 @@
+<Tagging><TagSet><Tag><Key>Project</Key><Value>Project One</Value></Tag><Tag><Key>User</Key><Value>jsmith</Value></Tag></TagSet></Tagging>
@@ -0,0 +1 @@
+<Tagging><TagSet><Tag><Key>team</Key><Value>storage</Value></Tag><Tag><Key>env</Key><Value>prod</Value></Tag></TagSet></Tagging>
@@ -0,0 +1 @@
+<Tagging><TagSet><Tag><Key>team</Key><Value>storage</Value></Tag><Tag><Key>env</Key><Value>prod</Value></Tag></TagSet></Tagging>
@@ -0,0 +1 @@
+<Tagging><TagSet><Tag><Key>team</Key><Value>storage</Value></Tag><Tag><Key>env</Key><Value>staging</Value></Tag></TagSet></Tagging>
--- a/Show More
+++ b/Show More
				`@@ -0,0 +1 @@`
				`<AccelerateConfiguration><Status>Enabled</Status></AccelerateConfiguration>`
				`@@ -0,0 +1 @@`
				`<AccelerateConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>Suspended</Status></AccelerateConfiguration>`
				`@@ -0,0 +1 @@`
				`<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>`
				`@@ -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>`
				`@@ -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>`
				`@@ -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>`
				`@@ -0,0 +1 @@`
				`<ServerSideEncryptionConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm></ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>`
				`@@ -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>
				`@@ -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>`
				`@@ -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>`