feat(capture): add explicit schema-bound checkpoint capabilities

feat(capture): persist first-decision provenance
feat(capture): atomically migrate legacy Created history
2026-10-27 22:37:13 +00:00 · 2026-10-11 21:28:05 +08:00 · 2026-10-11 18:58:12 +08:00 · 2026-10-11 16:39:10 +08:00 · 2026-10-11 16:39:10 +08:00 · 2026-10-11 16:30:12 +08:00
254 changed files with 15685 additions and 13742 deletions
@@ -1,2 +1,2 @@
-sha256-darwin=b9438fd1b82558f1fcd7e5a19dc0ceb0b34b5357d4ce99b7033de3ef019333ab
-sha256-linux=d4622517c13283247a022f49962c2cdb69e5ed0e13874fb55e0ab34fb923c044
+sha256-darwin=98dff0961c6eb21d6128bfd2d4a6100a624ddf1fb7db05dba360a5d3d0a310c2
+sha256-linux=890f5925b87e1bf3710c356253f3b2276d901e156df9ccc04ec9315374de6d22
@@ -1 +1 @@
-sha256=58df0d240cd74b916ce8f094ed1d8ad3e136f2f9bac157a155192a8786dae65c
+sha256=ba8f392b617cf9f2338142233a417e43927b28096635e628233ce8a60cf49d10
@@ -70,11 +70,6 @@ s3s-footprint-check: ## Check the s3s dependency footprint ratchet stays frozen
 	@echo "📦 Checking s3s footprint ratchet..."
 	./scripts/check_s3s_footprint.sh

-.PHONY: test-count-ratchet-check
-test-count-ratchet-check: ## Check no workspace crate lost tests or gained #[ignore] against the committed baseline
-	@echo "🧪 Checking per-crate test-count ratchet..."
-	./scripts/check_test_count_ratchet.sh
-
 .PHONY: fips-wording-check
 fips-wording-check: ## Check docs and crates/kms do not over-claim crypto capabilities
 	@echo "📣 Checking cryptographic capability wording guard..."
@@ -20,11 +20,11 @@ planning-docs-check: ## Check that no planning-type documents are committed
 	./scripts/check_no_planning_docs.sh

 .PHONY: pre-commit
-pre-commit: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check error-other-ratchet-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check test-count-ratchet-check fips-wording-check embedded-secrets-check test-wiring-check doc-paths-check planning-docs-check quick-check ## Run fast pre-commit checks without clippy/full tests
+pre-commit: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check error-other-ratchet-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check embedded-secrets-check test-wiring-check doc-paths-check planning-docs-check quick-check ## Run fast pre-commit checks without clippy/full tests
 	@echo "✅ All pre-commit checks passed!"

 .PHONY: pre-pr
-pre-pr: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check error-other-ratchet-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check test-count-ratchet-check fips-wording-check embedded-secrets-check test-wiring-check doc-paths-check planning-docs-check log-analyzer-rules-check offline-enrollment-e2e-check clippy-check test ## Run full pre-PR checks with clippy and tests
+pre-pr: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check error-other-ratchet-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check embedded-secrets-check test-wiring-check doc-paths-check planning-docs-check log-analyzer-rules-check offline-enrollment-e2e-check clippy-check test ## Run full pre-PR checks with clippy and tests
 	@echo "✅ All pre-PR checks passed!"

 .PHONY: dev-check
@@ -50,9 +50,7 @@ script-tests: ## Run shell script tests
 	./scripts/test_fuzz_runner.sh
 	./scripts/test_python_bin.sh
 	./scripts/check_embedded_secrets.sh --self-test
-	./scripts/check_s3s_footprint.sh --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_test_wiring.py --self-test
-	$(RUSTFS_PYTHON_BIN) ./scripts/codemods/s3_error_to_s3_types.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_e2e_binary.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_migration_gate_evidence.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/ci_gate.py --self-test
@@ -92,6 +92,12 @@ setup = 'ecstore-large-stack'
 filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
 setup = 'ecstore-base-stack'

+# These handler/store fixtures and the relocated-pool GET compose deep storage
+# futures in libtest threads; on Linux they overflow the default 2 MiB stack.
+[[profile.default.scripts]]
+filter = 'package(rustfs) & (test(/^admin::handlers::table_catalog::tests::/) | test(/^table_catalog::store::tests::/) | test(=app::object::get::tests::execute_get_object_resumes_from_relocated_pool_without_splicing_body))'
+setup = 'ecstore-base-stack'
+
 [[profile.default.scripts]]
 filter = 'binary(lifecycle_integration_test) | (package(rustfs) & test(/^app::lifecycle_transition_api_test::/))'
 setup = 'lifecycle-large-stack'
@@ -337,6 +343,10 @@ setup = 'ecstore-large-stack'
 filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
 setup = 'ecstore-base-stack'

+[[profile.ci.scripts]]
+filter = 'package(rustfs) & (test(/^admin::handlers::table_catalog::tests::/) | test(/^table_catalog::store::tests::/) | test(=app::object::get::tests::execute_get_object_resumes_from_relocated_pool_without_splicing_body))'
+setup = 'ecstore-base-stack'
+
 [[profile.ci.scripts]]
 filter = 'binary(lifecycle_integration_test) | (package(rustfs) & test(/^app::lifecycle_transition_api_test::/))'
 setup = 'lifecycle-large-stack'
@@ -579,14 +589,6 @@ test-group = 'ecstore-serial-flaky'
 # the PR-lane e2e coverage for scanner usage snapshots consumed by quota and
 # admin surfaces. It uses the same single-node, random-port, isolated-temp-dir
 # fixture as the existing smoke modules.
-#
-# Dual-stack differential (rustfs/backlog#2740): `dual_stack_differential::`
-# starts two single-node servers from the same binary, one per
-# RUSTFS_S3_STACK value, on random ports with isolated temp dirs, and runs one
-# SDK script covering every operation of `impl S3 for FS` against both. Its
-# comparator unit tests need no server. It meets the admission criteria
-# unchanged; the module clause admits a new differential test without another
-# filter edit.
 [profile.e2e-smoke]
 default-filter = """
    package(e2e_test) & (
@@ -594,7 +596,6 @@ default-filter = """
        | test(/^replication_extension_test::(test_replication_check_succeeds_with_remote_target|test_replication_check_rejects_target_without_object_lock|test_set_remote_target_rejects_unversioned_source_bucket|test_replication_check_rejects_unversioned_source_bucket|test_replication_check_rejects_missing_replication_config|test_replication_check_rejects_invalid_bucket|test_set_remote_target_rejects_same_bucket_on_same_deployment|test_set_remote_target_rejects_unversioned_target_bucket|test_set_remote_target_update_requires_arn|test_set_remote_target_update_rejects_missing_target|test_set_remote_target_rejects_invalid_target_url|test_set_remote_target_rejects_self_signed_https_target_without_skip_tls_verify|test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_pem|test_list_remote_targets_rejects_empty_bucket|test_list_remote_targets_rejects_invalid_bucket|test_remove_remote_target_rejects_missing_target|test_remove_remote_target_rejects_missing_arn|test_remove_remote_target_rejects_invalid_bucket|test_remove_remote_target_rejects_target_used_by_replication|test_delete_bucket_replication_removes_remote_target)$/)
        | test(/^reliant::lifecycle::/)
        | test(/^reliant::tiering::/)
-        | test(/^dual_stack_differential::/)
        | test(/^upgrade_compatibility_test::upgrade_write_readiness_tests::/)
        | test(/^on_demand_migration::(get_basic_test::(get_miss_pulls_inline_and_serves_locally_afterwards|head_miss_answers_from_the_source_without_persisting)|interaction_test::test_odm_admin_config_is_redacted_and_status_counts_match_the_source)$/)
    )
@@ -1,37 +0,0 @@
-# Files that may still reference s3s (`s3s::` or `s3s_sigv4`) when Phase 1 of
-# the gateway migration ends (rustfs/backlog#2734 task T0.6, rustfs/backlog#2741).
-# Consumed only by `scripts/check_s3s_footprint.sh --mode allowlist`.
-#
-# One entry per line: a repository-relative file path or a git `:(glob)`
-# pathspec. `**` matches across directories, `*` stays within one path
-# component, and a directory entry matches everything under it. Blank lines
-# and lines starting with `#` are ignored; an entry may not contain whitespace,
-# so a note belongs on its own comment line.
-#
-# Every Rust file listed by `git grep -l '\bs3s\(::\|_sigv4\)' -- '*.rs'` that
-# no entry matches fails the allowlist mode. Task T1.9 switches CI to that
-# mode; until then `--dry-run` prints the files still outside this list, which
-# is the Phase 1 tracking list.
-#
-# Never add crates/** other than crates/e2e_test: the engine crates leave s3s
-# entirely, and the script rejects such an entry. Only the s3s-facing edge of
-# the rustfs binary and the e2e oracle keep it.
-
-# rustfs binary: the s3s-facing edge that survives Phase 1
-rustfs/src/storage/ecfs.rs
-rustfs/src/storage/s3_compat/**
-rustfs/src/storage/access.rs
-rustfs/src/auth.rs
-rustfs/src/error.rs
-rustfs/src/storage_api.rs
-rustfs/src/app/storage_api.rs
-rustfs/src/admin/router.rs
-rustfs/src/app/metadata_route.rs
-rustfs/src/server/http.rs
-rustfs/src/server/layer.rs
-rustfs/src/server/hybrid.rs
-rustfs/src/protocols/client.rs
-
-# e2e oracle: the test target that verifies S3 behaviour against s3s
-crates/e2e_test/src/fake_s3_target/**
-crates/e2e_test/src/**
@@ -1,17 +0,0 @@
-# Exact baseline for the direct s3s footprint, consumed only by
-# scripts/check_s3s_footprint.sh (rustfs/backlog#1677; rustfs/backlog#2734 T0.4).
-#
-# Each value is the exact measurement of the tree, not a ceiling. CI fails when a
-# measured counter differs from its line here in either direction:
-#   - a larger measurement means new code widened the s3s surface: route it
-#     through the gateway abstractions instead; never raise a value here;
-#   - a smaller measurement means the PR shrank the surface: lower the matching
-#     value to the measured number in the same PR.
-#
-# files           Rust files referencing s3s paths, repo-wide minus crates/e2e_test
-# s3_error_lines  lines holding s3_error!, same scope, minus those the script
-#                 header's rule clears (rustfs_s3_types's macro); fails closed
-# ecstore_files   Rust files referencing s3s paths under crates/ecstore/src
-files=195
-s3_error_lines=1584
-ecstore_files=36
@@ -1,63 +0,0 @@
-# Per-crate test-count baseline consumed by scripts/check_test_count_ratchet.sh
-# (rustfs/backlog#2734 section 1 rule 3; T0.3 in rustfs/backlog#2738).
-#
-# One row per workspace member: <package>, tests, ignored, tab-separated.
-# tests counts #[test], #[tokio::test], #[tokio::test(...)] and #[test_case(...)]
-# attribute lines under the package directory; ignored counts #[ignore] and
-# #[ignore = "..."]. Both columns are exact: CI fails when a crate's tests fall
-# below or rise above its row, or when its ignored count changes. Raise a row in
-# the PR that adds tests; lowering one is a reviewed statement that tests were
-# intentionally removed. Regenerate with:
-#   scripts/check_test_count_ratchet.sh --update
-e2e_test	890	18
-rustfs	5532	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	6270	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,10 +86,6 @@ runs:
      shell: bash
      run: ./scripts/check_s3s_footprint.sh

-    - name: Check per-crate test-count ratchet
-      shell: bash
-      run: ./scripts/check_test_count_ratchet.sh
-
    - name: Check cryptographic capability wording
      shell: bash
      run: ./scripts/check_fips_wording.sh
@@ -440,6 +440,17 @@ jobs:
            fi
          done

+      - name: Verify macOS binary has no dynamic liblzma dependency
+        if: matrix.platform == 'macos'
+        shell: bash
+        run: |
+          binary="target/${{ matrix.target }}/release/rustfs"
+          if otool -L "$binary" | grep -E 'liblzma.*\.dylib'; then
+            echo "macOS release binary must not depend on a dynamic liblzma library" >&2
+            exit 1
+          fi
+          "$binary" --version
+
      - name: Generate the CPU symbol catalog
        if: endsWith(matrix.target, '-linux-gnu') && matrix.features == 'pyroscope'
        shell: bash
@@ -16,9 +16,7 @@ name: Continuous Integration

 on:
  push:
-    # gateway/integration receives the daily merge of main as a direct push
-    # (rustfs/backlog#2734), so its pushes must prove the result like main's.
-    branches: [ main, gateway/integration ]
+    branches: [ main ]
    paths-ignore:
      - "**.md"
      - "docs/**"
@@ -782,8 +780,6 @@ jobs:
    needs: [ build-rustfs-debug-binary ]
    runs-on: sm-standard-2
    timeout-minutes: 30
-    env:
-      RUSTFS_S3_STACK: legacy
    steps:
      - name: Checkout repository
        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
@@ -902,105 +898,18 @@ jobs:
          path: ${{ runner.temp }}/rustfs-e2e-*/rustfs.log
          retention-days: 3

-  # The same e2e-smoke suite against the same debug binary with the gateway
-  # S3 stack selected (rustfs/backlog#2734). Report-only: it is outside the
-  # Test and Lint aggregate and becomes required at T2.10. The archive and run
-  # steps must stay identical to e2e-tests; scripts/ci_gate.py enforces that.
-  e2e-tests-gateway:
-    name: e2e-smoke (gateway, report-only until T2.10)
-    needs: [ build-rustfs-debug-binary ]
-    runs-on: sm-standard-2
-    timeout-minutes: 30
-    continue-on-error: true
-    env:
-      RUSTFS_S3_STACK: gateway
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
-        with:
-          persist-credentials: false
-
-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
-        with:
-          rust-version: stable
-          cache-shared-key: ci-dev
-          cache-save-if: 'false'
-          install-build-packaging-tools: 'false'
-
-      - name: Set up Python
-        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
-        with:
-          python-version: "3.12"
-
-      - name: Install awscurl
-        run: |
-          python3 -m pip install --user --upgrade pip "awscurl==0.44"
-          echo "AWSCURL_PATH=$HOME/.local/bin/awscurl" >> "$GITHUB_ENV"
-
-      - name: Verify awscurl
-        run: test -x "$AWSCURL_PATH"
-
-      - name: Download debug binary
-        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
-        with:
-          name: rustfs-debug-binary
-          path: target/debug
-
-      - name: Make binary executable
-        run: chmod +x ./target/debug/rustfs
-
-      - name: Archive e2e smoke test binaries
-        env:
-          NEXTEST_ARCHIVE: ${{ runner.temp }}/rustfs-e2e-smoke.tar.zst
-          NEXTEST_LISTING: ${{ runner.temp }}/rustfs-e2e-smoke-list.json
-        run: |
-          cargo nextest archive --profile e2e-smoke -p e2e_test --archive-file "${NEXTEST_ARCHIVE}"
-          cargo nextest list --profile e2e-smoke --archive-file "${NEXTEST_ARCHIVE}" --message-format json > "${NEXTEST_LISTING}"
-          python3 ./scripts/check_test_wiring.py --check-profile e2e-smoke "${NEXTEST_LISTING}"
-          ./scripts/check_security_smoke_count.sh check "${NEXTEST_LISTING}"
-
-      - name: Run e2e smoke suite
-        env:
-          NEXTEST_ARCHIVE: ${{ runner.temp }}/rustfs-e2e-smoke.tar.zst
-          RUSTFS_E2E_LOG_DIR: ${{ runner.temp }}/rustfs-e2e-smoke-logs
-          SAMPLER_INTERVAL_SECS: "15"
-        run: |
-          ./scripts/ci/resource_sampler.sh start e2e-smoke
-          trap './scripts/ci/resource_sampler.sh stop' EXIT
-          python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-smoke --archive-file "${NEXTEST_ARCHIVE}" \
-            --status-level all --final-status-level all --failure-output final
-
-      - name: Upload e2e smoke diagnostics
-        if: always()
-        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
-        with:
-          name: e2e-smoke-gateway-diagnostics-${{ github.run_number }}
-          path: |
-            ${{ runner.temp }}/rustfs-e2e-smoke-logs/
-            ${{ runner.temp }}/rustfs-e2e-smoke-list.json
-          if-no-files-found: warn
-
-      - name: Upload e2e smoke JUnit report
-        if: always()
-        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
-        with:
-          name: e2e-smoke-gateway-junit-${{ github.run_number }}
-          path: target/nextest/e2e-smoke/junit.xml
-          if-no-files-found: warn
-
  e2e-full:
    name: End-to-End Tests (full merge gate)
    # Merge gate only (backlog#1149 ci-5): the never-automated user-visible
    # suites — KMS, object_lock, multipart_auth, quota, checksum, encryption,
    # security-boundary, ... — via the e2e-full nextest profile. Too heavy for
-    # every PR, so it is gated to main and gateway/integration pushes, the merge
-    # queue, and manual dispatch. protocols / the 7 cluster suites / replication
-    # / #[ignore] are owned by other lanes (see .config/nextest.toml profile.e2e-full).
+    # every PR, so it is gated to main pushes, the merge queue, and manual
+    # dispatch. protocols / the 7 cluster suites / replication / #[ignore] are
+    # owned by other lanes (see .config/nextest.toml profile.e2e-full).
    if: >-
      github.event_name == 'workflow_dispatch' ||
      github.event_name == 'merge_group' ||
-      (github.event_name == 'push' && (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/gateway/integration'))
+      (github.event_name == 'push' && github.ref == 'refs/heads/main')
    needs: [ build-rustfs-debug-binary ]
    runs-on: sm-standard-2
    timeout-minutes: 55
@@ -155,7 +155,7 @@ default build (lifecycle:
   - ⚠️ PARTIALLY VIOLATED: serving-side `s3s` references remain in ecstore
     (bucket metadata/replication/lifecycle DTOs and error mapping). The
     count is ratcheted shrink-only by `scripts/check_s3s_footprint.sh`
-     (the `ecstore_files` line of `.config/s3s-footprint-baseline.txt`; the `object_lock` module was converted to
+     (`S3S_ECSTORE_FILES_BASELINE`; the `object_lock` module was converted to
     storage-level types as the first ratchet step). Target state: the
     baseline reaches zero and ecstore's `Cargo.toml` drops `s3s`.

@@ -275,6 +275,15 @@ dependencies = [
 "windows-sys 0.61.2",
 ]

+[[package]]
+name = "anyerror"
+version = "0.1.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "71add24cc141a1e8326f249b74c41cfd217aeb2a67c9c6cf9134d175469afd49"
+dependencies = [
+ "serde",
+]
+
 [[package]]
 name = "anyhow"
 version = "1.0.104"
@@ -1706,6 +1715,30 @@ dependencies = [
 "syn 3.0.6",
 ]

+[[package]]
+name = "borsh"
+version = "1.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "553c5d846a6ba5150c65e3b1b8ec073bcf1abc20f9b7220de384a4443ea4e20a"
+dependencies = [
+ "borsh-derive",
+ "bytes",
+ "cfg_aliases",
+]
+
+[[package]]
+name = "borsh-derive"
+version = "1.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "12cdfe656708a01f89b451a7d36466e6fe6c414de0aa18fc54f864f6f9ca9f56"
+dependencies = [
+ "once_cell",
+ "proc-macro-crate",
+ "proc-macro2",
+ "quote",
+ "syn 3.0.6",
+]
+
 [[package]]
 name = "brotli"
 version = "8.0.4"
@@ -1772,6 +1805,18 @@ version = "3.20.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"

+[[package]]
+name = "byte-unit"
+version = "5.2.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c719d56f7e96194cfc53460976d3ba51c85719747c9c62ed99981847b551152b"
+dependencies = [
+ "rust_decimal",
+ "schemars 1.2.2",
+ "serde",
+ "utf8-width",
+]
+
 [[package]]
 name = "bytemuck"
 version = "1.25.2"
@@ -3724,13 +3769,34 @@ dependencies = [
 "syn 2.0.119",
 ]

+[[package]]
+name = "derive_more"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4a9b99b9cbbe49445b21764dc0625032a89b145a2642e67603e1c936f5458d05"
+dependencies = [
+ "derive_more-impl 1.0.0",
+]
+
 [[package]]
 name = "derive_more"
 version = "2.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d751e9e49156b02b44f9c1815bcb94b984cdcc4396ecc32521c739452808b134"
 dependencies = [
- "derive_more-impl",
+ "derive_more-impl 2.1.1",
+]
+
+[[package]]
+name = "derive_more-impl"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cb7330aeadfbe296029522e6c40f315320aba36fc43a5b3632f3795348f3bd22"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+ "unicode-xid",
 ]

 [[package]]
@@ -6234,7 +6300,7 @@ dependencies = [
 "bytes",
 "chrono",
 "dashmap",
- "derive_more",
+ "derive_more 2.1.1",
 "futures-util",
 "getrandom 0.4.3",
 "lazy_static",
@@ -6388,6 +6454,12 @@ dependencies = [
 "quote",
 ]

+[[package]]
+name = "maplit"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3e2e65a1a2e43cfcb47a895c4c8b10d1f4a61097f9f254f183aee60cad9c651d"
+
 [[package]]
 name = "matchers"
 version = "0.2.0"
@@ -7329,6 +7401,42 @@ dependencies = [
 "url",
 ]

+[[package]]
+name = "openraft"
+version = "0.9.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a97014fb78acb77be3a40ac2da305f6dd3a6b243f3a908ace87d29b3972eaafd"
+dependencies = [
+ "anyerror",
+ "byte-unit",
+ "chrono",
+ "clap",
+ "derive_more 1.0.0",
+ "futures",
+ "maplit",
+ "openraft-macros",
+ "rand 0.8.8",
+ "serde",
+ "thiserror 1.0.69",
+ "tokio",
+ "tracing",
+ "tracing-futures",
+ "validit",
+]
+
+[[package]]
+name = "openraft-macros"
+version = "0.9.25"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "55417d79f46a787dd88c043d831d8b7a0713f7c69d1925f9de51cc42c437250f"
+dependencies = [
+ "chrono",
+ "proc-macro2",
+ "quote",
+ "semver",
+ "syn 2.0.119",
+]
+
 [[package]]
 name = "openssl-probe"
 version = "0.2.1"
@@ -8958,6 +9066,15 @@ dependencies = [
 "syn 2.0.119",
 ]

+[[package]]
+name = "redb"
+version = "4.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fb338a6c67830a61bed824b78c2bb034ab1ff401a20616bd7957524f4fdc2f22"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "redis"
 version = "1.7.1"
@@ -9464,6 +9581,23 @@ dependencies = [
 "walkdir",
 ]

+[[package]]
+name = "rust_decimal"
+version = "1.43.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7653272e75dcac41dc199fbea6f5797633994fafd339943c06c9af16bf29cd3a"
+dependencies = [
+ "arrayvec",
+ "borsh",
+ "bytes",
+ "num-traits",
+ "rand 0.8.8",
+ "rand 0.9.5",
+ "serde",
+ "serde_json",
+ "wasm-bindgen",
+]
+
 [[package]]
 name = "rustc-demangle"
 version = "0.1.28"
@@ -9666,6 +9800,21 @@ dependencies = [
 "url",
 ]

+[[package]]
+name = "rustfs-capture-authority"
+version = "1.0.1"
+dependencies = [
+ "openraft",
+ "redb",
+ "serde",
+ "serde_json",
+ "sha2 0.11.0",
+ "tempfile",
+ "thiserror 2.0.21",
+ "tokio",
+ "url",
+]
+
 [[package]]
 name = "rustfs-checksums"
 version = "1.0.1"
@@ -10284,6 +10433,7 @@ dependencies = [
 "reqwest",
 "rmp-serde",
 "rustfs-signer",
+ "s3s",
 "serde",
 "serde_json",
 "sysinfo",
@@ -10472,6 +10622,7 @@ dependencies = [
 "tokio",
 "tracing",
 "tracing-subscriber",
+ "uuid",
 ]

 [[package]]
@@ -10508,7 +10659,6 @@ dependencies = [
 "rustfs-keystone",
 "rustfs-policy",
 "rustfs-rio",
- "rustfs-s3-types",
 "rustfs-storage-api",
 "rustfs-test-utils",
 "rustfs-tls-runtime",
@@ -10523,7 +10673,6 @@ dependencies = [
 "sha2 0.11.0",
 "socket2",
 "subtle",
- "sync_wrapper",
 "tempfile",
 "thiserror 2.0.21",
 "time",
@@ -10595,6 +10744,7 @@ dependencies = [
 "faster-hex",
 "futures",
 "hex-simd",
+ "hmac 0.13.0",
 "hotpath",
 "http 1.5.0",
 "http-body-util",
@@ -10670,7 +10820,6 @@ dependencies = [
 "rustfs-checksums",
 "rustfs-config",
 "rustfs-rio",
- "rustfs-s3-types",
 "rustfs-signer",
 "rustfs-storage-api",
 "rustfs-tls-runtime",
@@ -10704,19 +10853,9 @@ 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]]
@@ -10740,7 +10879,6 @@ dependencies = [
 "parking_lot",
 "rustfs-common",
 "rustfs-ecstore",
- "rustfs-s3-types",
 "rustfs-storage-api",
 "rustfs-test-utils",
 "s3s",
@@ -10854,10 +10992,9 @@ dependencies = [
 "bytes",
 "hotpath",
 "http 1.5.0",
- "http-body",
 "hyper",
- "rustfs-s3-types",
 "rustfs-utils",
+ "s3s",
 "serde_urlencoded",
 "thiserror 2.0.21",
 "time",
@@ -10915,6 +11052,7 @@ dependencies = [
 "rustfs-utils",
 "rustls",
 "rustls-native-certs",
+ "s3s",
 "serde",
 "serde_json",
 "sha2 0.11.0",
@@ -11073,6 +11211,7 @@ dependencies = [
 "async-compression",
 "futures",
 "hotpath",
+ "liblzma",
 "rustfs-rio",
 "rustfs-tokio-tar",
 "serde",
@@ -13057,6 +13196,16 @@ dependencies = [
 "tracing-subscriber",
 ]

+[[package]]
+name = "tracing-futures"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "97d095ae15e245a057c8e8451bab9b3ee1e1f68e9ba2b4fbc18d0ac5237835f2"
+dependencies = [
+ "pin-project",
+ "tracing",
+]
+
 [[package]]
 name = "tracing-log"
 version = "0.2.0"
@@ -13207,7 +13356,7 @@ checksum = "79acf4b257aa56dd7f00466a81693d6c79c362c55092b3137909d6e76249ad80"
 dependencies = [
 "async-trait",
 "chrono",
- "derive_more",
+ "derive_more 2.1.1",
 "libc",
 "md-5 0.10.6",
 "thiserror 2.0.21",
@@ -13307,6 +13456,12 @@ version = "2.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"

+[[package]]
+name = "utf8-width"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "159a7cadce548703edd50d24069bc294c5415ecab0a480e0cd1ca06d112dc94a"
+
 [[package]]
 name = "utf8_iter"
 version = "1.0.4"
@@ -13333,6 +13488,15 @@ dependencies = [
 "wasm-bindgen",
 ]

+[[package]]
+name = "validit"
+version = "0.2.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4417835826b78aa6265724624d424df2b1e6e71a33329227d8b04a88950780dc"
+dependencies = [
+ "anyerror",
+]
+
 [[package]]
 name = "valuable"
 version = "0.1.1"
@@ -18,6 +18,7 @@ members = [
    "crates/audit", # Audit target management system with multi-target fan-out
    "crates/checksums", # client checksums
    "crates/common", # Shared utilities and data structures
+    "crates/capture-authority", # Durable capture authority storage
    "crates/data-usage", # Shared data usage models and algorithms
    "crates/config", # Configuration management
    "crates/connect-symbol-catalog", # Release-only CPU profile symbol catalog generator
@@ -48,7 +49,7 @@ members = [
    "crates/replication", # Replication contracts and wire formats
    "crates/concurrency", # Concurrency management for RustFS - timeout, locking, backpressure, and I/O scheduling
    "crates/s3-client", # S3 client for engine-side consumption of remote S3 endpoints (tiering, transition targets)
-    "crates/s3-types", # S3 event and error type definitions
+    "crates/s3-types", # S3 event type definitions
    "crates/s3-ops", # S3 operation definitions and mapping
    "crates/s3select-api", # S3 Select API interface
    "crates/s3select-query", # S3 Select query engine
@@ -314,6 +315,8 @@ ratelimit = "2.0.0"
 rayon = "1.12.0"
 rustfs-erasure-codec = { version = "9.0.0" }
 reed-solomon-simd = "3.1.0"
+openraft = { version = "=0.9.25", default-features = false, features = ["serde"] }
+redb = "=4.3.0"
 regex = { version = "1.13.1" }
 rumqttc = { package = "rumqttc-next", version = "0.34.0" }
 redis = { version = "1.7.1" }
@@ -323,10 +326,6 @@ 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"
@@ -335,7 +334,6 @@ compact_str = "0.10.0"
 snap = "1.1.2"
 starshard = { version = "2.4.0" }
 strum = { version = "0.28.0" }
-sync_wrapper = "1.0.2"
 sysinfo = "0.39.6"
 temp-env = "0.3.6"
 tempfile = "3.27.0"
@@ -0,0 +1,24 @@
+[package]
+name = "rustfs-capture-authority"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+rust-version.workspace = true
+repository.workspace = true
+description = "Durable local storage for capture authority"
+
+[dependencies]
+openraft = { workspace = true }
+redb = { workspace = true }
+sha2 = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+url = { workspace = true }
+
+[dev-dependencies]
+serde_json = { workspace = true }
+tempfile = { workspace = true }
+
+[lints]
+workspace = true
@@ -0,0 +1,531 @@
+use super::*;
+use crate::application::{APPLY_RESULTS, EVENTS, OPERATIONS, VERSIONS};
+use crate::tests::CrashBackend;
+use redb::ReadableTableMetadata;
+use std::{
+    io::{BufRead, BufReader, Write},
+    process::{Command, Stdio},
+    sync::mpsc,
+    time::Duration,
+};
+fn identity() -> NodeIdentity {
+    NodeIdentity {
+        cluster_id: [7; 16],
+        node_id: 0,
+    }
+}
+fn binding() -> CaptureBinding {
+    CaptureBinding {
+        revision: 9,
+        binding_id: [8; 16],
+        target_id: [9; 16],
+    }
+}
+fn command() -> DecideCreated {
+    DecideCreated {
+        operation_id: [1; 16],
+        object: ObjectIdentity {
+            bucket_incarnation: [2; 16],
+            key: vec![0, 255, 3],
+        },
+        expected_head: None,
+        prepared: PreparedIdentity {
+            version_id: [3; 16],
+            preparation_id: [4; 16],
+            content_digest: [5; 32],
+            content_length: 123,
+        },
+        binding_revision: 9,
+    }
+}
+fn id(index: u64) -> LogId {
+    LogId {
+        term: 8,
+        leader_node: 0,
+        index,
+    }
+}
+fn append(store: &mut Store, index: u64, c: &DecideCreated) {
+    store
+        .append(&[LogEntry {
+            id: id(index),
+            payload: c.encode().expect("encode command"),
+        }])
+        .expect("append command");
+}
+fn apply(store: &mut Store, index: u64, c: &DecideCreated) -> CreatedResult {
+    append(store, index, c);
+    store.save_committed(id(index)).expect("commit log");
+    store.apply_next_created(id(index)).expect("apply command")
+}
+fn success(c: &DecideCreated, index: u64) -> CreatedResult {
+    CreatedResult::Created {
+        version_id: c.prepared.version_id,
+        event_id: c.operation_id,
+        decision_log_id: id(index),
+    }
+}
+fn complete(store: &Store) {
+    let c = command();
+    assert_eq!(store.capture_binding().expect("binding"), Some(binding()));
+    assert_eq!(
+        store.created_head(&c.object).expect("head"),
+        Some(c.prepared.version_id),
+        "U11 acknowledged head lost"
+    );
+    assert_eq!(
+        store.created_version(&c.object, c.prepared.version_id).expect("version"),
+        Some(CreatedVersion {
+            prepared: c.prepared.clone(),
+            decision_log_id: id(0),
+            binding_revision: 9
+        })
+    );
+    assert_eq!(
+        store.created_event(c.operation_id).expect("event"),
+        Some(CreatedEvent {
+            object: c.object.clone(),
+            prepared: c.prepared.clone(),
+            decision_log_id: id(0),
+            binding: binding()
+        }),
+        "U11 complete event missing"
+    );
+    assert_eq!(
+        store.created_operation(c.operation_id).expect("operation"),
+        Some(CreatedOperation {
+            canonical_command: c.encode().expect("encode"),
+            result: success(&c, 0)
+        })
+    );
+    assert_eq!(
+        store.created_apply_result(0).expect("result"),
+        Some(CreatedApplyResult {
+            log_id: id(0),
+            result: success(&c, 0)
+        })
+    );
+    assert_eq!(store.last_applied().expect("last applied"), Some(id(0)));
+}
+fn fixture() -> (tempfile::TempDir, Store) {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let mut store = Store::initialize(dir.path().join("db"), identity()).expect("initialize");
+    store.pin_capture_binding(binding()).expect("pin");
+    (dir, store)
+}
+fn reopen(path: &std::path::Path) -> Store {
+    let result = Store::open(path, identity());
+    assert!(result.is_ok(), "U11 complete reopen: {:?}", result.as_ref().err());
+    result.expect("asserted successful reopen")
+}
+#[test]
+fn created_codec_is_canonical_and_strict() {
+    let c = command();
+    let b = c.encode().expect("encode");
+    let mut golden = b"CRTD\x01".to_vec();
+    golden.extend([1; 16]);
+    golden.extend([2; 16]);
+    golden.extend(3u32.to_be_bytes());
+    golden.extend([0, 255, 3]);
+    golden.push(0);
+    golden.extend([3; 16]);
+    golden.extend([4; 16]);
+    golden.extend([5; 32]);
+    golden.extend(123u64.to_be_bytes());
+    golden.extend(9u64.to_be_bytes());
+    assert_eq!(b, golden);
+    assert_eq!(DecideCreated::decode(&b).expect("decode"), c);
+    let mut variants = Vec::new();
+    let mut v = c.clone();
+    v.operation_id = [6; 16];
+    variants.push(v);
+    let mut v = c.clone();
+    v.object.bucket_incarnation = [6; 16];
+    variants.push(v);
+    let mut v = c.clone();
+    v.object.key.push(4);
+    variants.push(v);
+    let mut v = c.clone();
+    v.expected_head = Some([6; 16]);
+    variants.push(v);
+    let mut v = c.clone();
+    v.prepared.version_id = [6; 16];
+    variants.push(v);
+    let mut v = c.clone();
+    v.prepared.preparation_id = [6; 16];
+    variants.push(v);
+    let mut v = c.clone();
+    v.prepared.content_digest = [6; 32];
+    variants.push(v);
+    let mut v = c.clone();
+    v.prepared.content_length += 1;
+    variants.push(v);
+    let mut v = c.clone();
+    v.binding_revision += 1;
+    variants.push(v);
+    for v in variants {
+        assert_ne!(v.encode().expect("variant"), b)
+    }
+    for n in 0..b.len() {
+        assert!(DecideCreated::decode(&b[..n]).is_err())
+    }
+    let mut bad = b.clone();
+    bad[4] = 2;
+    assert!(DecideCreated::decode(&bad).is_err());
+    let mut bad = b.clone();
+    bad[44] = 2;
+    assert!(DecideCreated::decode(&bad).is_err());
+    let mut bad = b.clone();
+    bad.push(0);
+    assert!(DecideCreated::decode(&bad).is_err());
+    let mut bad = b;
+    bad[37..41].copy_from_slice(&4097u32.to_be_bytes());
+    assert!(DecideCreated::decode(&bad).is_err());
+    let mut bad = c.clone();
+    bad.operation_id = [0; 16];
+    assert!(bad.encode().is_err());
+    bad = c;
+    bad.object.key = vec![0; MAX_OBJECT_KEY_LENGTH + 1];
+    assert!(bad.encode().is_err());
+}
+#[test]
+fn created_committed_order_and_identity_are_enforced() {
+    let (_dir, mut s) = fixture();
+    let c = command();
+    append(&mut s, 0, &c);
+    append(&mut s, 1, &c);
+    assert!(s.apply_next_created(id(0)).is_err());
+    assert_eq!(s.last_applied().expect("no apply"), None);
+    s.save_committed(id(1)).expect("committed");
+    assert!(matches!(s.apply_next_created(id(1)), Err(StoreError::InvalidApplySequence)));
+    let mut wrong = id(0);
+    wrong.term += 1;
+    assert!(s.apply_next_created(wrong).is_err());
+    wrong = id(0);
+    wrong.leader_node += 1;
+    assert!(s.apply_next_created(wrong).is_err());
+    assert_eq!(s.apply_next_created(id(0)).expect("first"), success(&c, 0));
+    complete(&s);
+    assert_eq!(s.apply_next_created(id(0)).expect("retry"), success(&c, 0));
+    complete(&s);
+    assert!(s.apply_next_created(wrong).is_err());
+    assert_eq!(s.apply_next_created(id(1)).expect("duplicate log"), success(&c, 0));
+    assert_eq!(s.last_applied().expect("last"), Some(id(1)));
+    assert_eq!(
+        s.created_apply_result(1).expect("result"),
+        Some(CreatedApplyResult {
+            log_id: id(1),
+            result: success(&c, 0)
+        })
+    );
+    let tx = s.database.begin_read().expect("read");
+    assert_eq!(tx.open_table(EVENTS).expect("events").len().expect("len"), 1);
+}
+#[test]
+fn created_head_version_event_and_dedup_reopen_together() {
+    let (dir, mut s) = fixture();
+    let c = command();
+    assert_eq!(apply(&mut s, 0, &c), success(&c, 0));
+    drop(s);
+    let mut s = reopen(&dir.path().join("db"));
+    complete(&s);
+    assert_eq!(apply(&mut s, 1, &c), success(&c, 0));
+    let mut changed = c.clone();
+    changed.prepared.content_digest = [6; 32];
+    assert_eq!(
+        apply(&mut s, 2, &changed),
+        CreatedResult::OperationConflict,
+        "U11 full canonical operation conflict"
+    );
+    assert_eq!(
+        s.created_operation(c.operation_id)
+            .expect("original")
+            .expect("operation")
+            .result,
+        success(&c, 0)
+    );
+    let mut stale = c.clone();
+    stale.operation_id = [6; 16];
+    stale.prepared.version_id = [6; 16];
+    assert_eq!(
+        apply(&mut s, 3, &stale),
+        CreatedResult::HeadMismatch {
+            actual_head: Some(c.prepared.version_id)
+        }
+    );
+    let mut reuse = c.clone();
+    reuse.operation_id = [7; 16];
+    reuse.expected_head = Some(c.prepared.version_id);
+    assert_eq!(apply(&mut s, 4, &reuse), CreatedResult::VersionConflict);
+    let mut replacement = stale.clone();
+    replacement.operation_id = [10; 16];
+    replacement.expected_head = Some(c.prepared.version_id);
+    assert_eq!(apply(&mut s, 5, &replacement), success(&replacement, 5));
+    assert_eq!(s.created_head(&c.object).expect("head"), Some(replacement.prepared.version_id));
+    assert_eq!(
+        s.created_version(&c.object, c.prepared.version_id)
+            .expect("old version")
+            .expect("version")
+            .decision_log_id,
+        id(0)
+    );
+    assert!(s.created_event(c.operation_id).expect("old event").is_some());
+    assert!(s.created_event(replacement.operation_id).expect("new event").is_some());
+    assert!(s.created_event(stale.operation_id).expect("rejection event").is_none());
+    assert_eq!(
+        apply(&mut s, 6, &stale),
+        CreatedResult::HeadMismatch {
+            actual_head: Some(c.prepared.version_id)
+        }
+    );
+    drop(s);
+    let s = reopen(&dir.path().join("db"));
+    assert_eq!(s.last_applied().expect("last"), Some(id(6)));
+    let tx = s.database.begin_read().expect("read");
+    assert_eq!(tx.open_table(EVENTS).expect("events").len().expect("len"), 2);
+}
+#[test]
+fn created_binding_is_pinned_and_not_caller_selected() {
+    let (dir, mut s) = fixture();
+    let mut bad = command();
+    bad.binding_revision = 10;
+    assert_eq!(apply(&mut s, 0, &bad), CreatedResult::BindingMismatch);
+    assert!(s.created_head(&bad.object).expect("head").is_none());
+    assert!(s.created_event(bad.operation_id).expect("event").is_none());
+    assert_eq!(
+        s.created_operation(bad.operation_id)
+            .expect("operation")
+            .expect("saved")
+            .result,
+        CreatedResult::BindingMismatch
+    );
+    s.pin_capture_binding(binding()).expect("idempotent pin");
+    let mut changed = binding();
+    changed.target_id = [10; 16];
+    assert!(matches!(s.pin_capture_binding(changed), Err(StoreError::BindingConflict)));
+    drop(s);
+    let s = reopen(&dir.path().join("db"));
+    assert_eq!(s.capture_binding().expect("binding"), Some(binding()));
+}
+#[test]
+fn created_schema_compatibility_and_corruption_are_checked() {
+    let dir = tempfile::tempdir().expect("dir");
+    let path = dir.path().join("db");
+    let s = Store::initialize(&path, identity()).expect("schema1");
+    assert_eq!(s.capture_binding().expect("unmigrated"), None);
+    drop(s);
+    let mut s = reopen(&path);
+    assert_eq!(s.last_applied().expect("no apply"), None);
+    append(&mut s, 0, &command());
+    s.save_committed(id(0)).expect("schema2");
+    drop(s);
+    let mut s = reopen(&path);
+    assert_eq!(s.capture_binding().expect("schema2 unchanged"), None);
+    s.pin_capture_binding(binding()).expect("explicit upgrade");
+    drop(s);
+    let mut s = reopen(&path);
+    assert_eq!(s.capture_binding().expect("schema3"), Some(binding()));
+    s.apply_next_created(id(0)).expect("apply");
+    drop(s);
+    let db = Database::open(&path).expect("raw db");
+    let tx = db.begin_write().expect("write corruption");
+    tx.open_table(EVENTS)
+        .expect("events")
+        .remove(command().operation_id.as_slice())
+        .expect("remove reference");
+    tx.commit().expect("commit corruption");
+    drop(db);
+    let result = Store::open(&path, identity());
+    assert!(
+        matches!(result, Err(StoreError::CorruptRecord)),
+        "U11 missing event rejected before expect"
+    );
+    let (dir, mut s) = fixture();
+    assert_eq!(s.read_committed().expect("empty committed"), None);
+    append(&mut s, 0, &command());
+    s.save_committed(id(0)).expect("committed schema3");
+    assert_eq!(s.capture_binding().expect("no downgrade"), Some(binding()));
+    drop(s);
+    let path = dir.path().join("db");
+    let db = Database::open(&path).expect("raw db");
+    let tx = db.begin_write().expect("write");
+    let mut marker = vec![1];
+    marker.extend(encode_id(id(0)));
+    tx.open_table(META)
+        .expect("meta")
+        .insert("last_applied", marker.as_slice())
+        .expect("corrupt applied");
+    tx.commit().expect("commit");
+    drop(db);
+    assert!(
+        matches!(Store::open(&path, identity()), Err(StoreError::CorruptRecord)),
+        "U11 last applied missing result rejected"
+    );
+    let (dir, s) = fixture();
+    drop(s);
+    let path = dir.path().join("db");
+    let db = Database::open(&path).expect("raw pinned db");
+    let tx = db.begin_write().expect("missing binding write");
+    tx.open_table(META)
+        .expect("meta")
+        .remove("capture_binding")
+        .expect("remove binding");
+    tx.commit().expect("binding corruption commit");
+    drop(db);
+    assert!(
+        matches!(Store::open(&path, identity()), Err(StoreError::CorruptRecord)),
+        "U11 missing pinned binding rejected"
+    );
+
+    let (dir, mut s) = fixture();
+    let c = command();
+    assert_eq!(apply(&mut s, 0, &c), success(&c, 0));
+    assert_eq!(apply(&mut s, 1, &c), success(&c, 0));
+    drop(s);
+    let path = dir.path().join("db");
+    let db = Database::open(&path).expect("raw duplicate db");
+    let tx = db.begin_write().expect("retarget decision write");
+    let mut key = created::object_bytes(&c.object).expect("object key");
+    key.extend(c.prepared.version_id);
+    let mut version = Vec::new();
+    created::put_prepared(&mut version, &c.prepared);
+    version.extend(encode_id(id(1)));
+    version.extend(binding().revision.to_be_bytes());
+    tx.open_table(VERSIONS)
+        .expect("versions")
+        .insert(key.as_slice(), version.as_slice())
+        .expect("retarget version");
+    let mut event = created::object_bytes(&c.object).expect("event object");
+    created::put_prepared(&mut event, &c.prepared);
+    event.extend(encode_id(id(1)));
+    event.extend(created::binding_bytes(&binding()).expect("binding bytes"));
+    tx.open_table(EVENTS)
+        .expect("events")
+        .insert(c.operation_id.as_slice(), event.as_slice())
+        .expect("retarget event");
+    let canonical = c.encode().expect("canonical");
+    let mut operation = u64::try_from(canonical.len()).expect("length").to_be_bytes().to_vec();
+    operation.extend(canonical);
+    operation.extend(created::result_bytes(&success(&c, 1)));
+    tx.open_table(OPERATIONS)
+        .expect("operations")
+        .insert(c.operation_id.as_slice(), operation.as_slice())
+        .expect("retarget original response");
+    for index in 0..=1 {
+        let mut result = encode_id(id(index));
+        result.extend(created::result_bytes(&success(&c, 1)));
+        tx.open_table(APPLY_RESULTS)
+            .expect("results")
+            .insert(index, result.as_slice())
+            .expect("retarget duplicate response");
+    }
+    tx.commit().expect("coherent retarget commit");
+    drop(db);
+    assert!(
+        matches!(Store::open(&path, identity()), Err(StoreError::CorruptRecord)),
+        "U11 original decision cannot retarget duplicate log"
+    );
+}
+#[test]
+fn created_process_child() {
+    if let Some(path) = std::env::var_os("U11_CHILD_DB") {
+        let mut s = Store::initialize(path, identity()).expect("child init");
+        s.pin_capture_binding(binding()).expect("child pin");
+        assert_eq!(apply(&mut s, 0, &command()), success(&command(), 0));
+        complete(&s);
+        println!("U11_ACK");
+        std::io::stdout().flush().expect("flush ACK");
+        loop {
+            std::thread::park()
+        }
+    }
+    let (_dir, mut s) = fixture();
+    assert_eq!(apply(&mut s, 0, &command()), success(&command(), 0));
+    complete(&s);
+}
+#[test]
+fn created_acknowledged_apply_survives_child_kill() {
+    let dir = tempfile::tempdir().expect("dir");
+    let path = dir.path().join("db");
+    let mut child = Command::new(std::env::current_exe().expect("test executable"))
+        .args(["--exact", "application_tests::created_process_child", "--nocapture"])
+        .env("U11_CHILD_DB", &path)
+        .stdout(Stdio::piped())
+        .stderr(Stdio::inherit())
+        .spawn()
+        .expect("spawn");
+    let stdout = child.stdout.take().expect("stdout");
+    let (tx, rx) = mpsc::channel();
+    let reader = std::thread::spawn(move || {
+        for line in BufReader::new(stdout).lines() {
+            let line = line.expect("child line");
+            if line.ends_with("U11_ACK") {
+                let _ = tx.send(());
+                break;
+            }
+        }
+    });
+    let acknowledged = rx.recv_timeout(Duration::from_secs(20));
+    child.kill().expect("kill child");
+    child.wait().expect("wait child");
+    reader.join().expect("reader cleanup");
+    assert!(acknowledged.is_ok(), "U11 bounded child ACK missing");
+    complete(&reopen(&path));
+}
+#[test]
+fn created_power_cut_before_and_after_apply_flush() {
+    let (dir, mut s) = fixture();
+    append(&mut s, 0, &command());
+    s.save_committed(id(0)).expect("committed");
+    drop(s);
+    let path = dir.path().join("db");
+    let backend = CrashBackend::new(&path);
+    let mut s = Store::from_database(backend.database(), identity()).expect("backend");
+    assert_eq!(s.apply_next_created(id(0)).expect("apply ACK"), success(&command(), 0));
+    backend.cut();
+    drop(s);
+    complete(&reopen(&path));
+    for sync in 1..=2 {
+        for after in [false, true] {
+            let (dir, mut s) = fixture();
+            append(&mut s, 0, &command());
+            s.save_committed(id(0)).expect("committed");
+            drop(s);
+            let path = dir.path().join("db");
+            let backend = CrashBackend::new(&path);
+            let mut s = Store::from_database(backend.database(), identity()).expect("backend");
+            let (entered, resume) = backend.pause_sync(sync, after);
+            let writer = std::thread::spawn(move || {
+                let result = s.apply_next_created(id(0));
+                (s, result)
+            });
+            entered.wait();
+            backend.cut();
+            resume.wait();
+            let (s, result) = writer.join().expect("writer");
+            drop(s);
+            let reopened = Store::open(&path, identity());
+            assert!(
+                reopened.is_ok(),
+                "U11 apply cut must reopen whole old/new state sync={sync} after={after}: {:?}",
+                reopened.as_ref().err()
+            );
+            let s = reopened.expect("asserted reopen");
+            if s.last_applied().expect("last").is_some() {
+                complete(&s)
+            } else {
+                assert!(result.is_err());
+                assert_eq!(s.created_head(&command().object).expect("old head"), None);
+                assert_eq!(
+                    s.created_version(&command().object, command().prepared.version_id)
+                        .expect("old version"),
+                    None
+                );
+                assert_eq!(s.created_event(command().operation_id).expect("old event"), None);
+                assert_eq!(s.created_operation(command().operation_id).expect("old op"), None);
+                assert_eq!(s.created_apply_result(0).expect("old result"), None)
+            }
+            eprintln!("U11 cut boundary sync={sync} after={after}");
+        }
+    }
+}
@@ -0,0 +1,63 @@
+use crate::{LOGS, LogId, META, Store, StoreError, decode_id, decode_log, encode_id, number};
+use redb::{ReadableDatabase, ReadableTable};
+
+pub(crate) fn validate_committed(
+    meta: &impl ReadableTable<&'static str, &'static [u8]>,
+    logs: &impl ReadableTable<u64, &'static [u8]>,
+) -> Result<Option<LogId>, StoreError> {
+    let schema = number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?;
+    let marker = meta.get("committed")?;
+    match (schema, marker) {
+        (1 | 3 | 4 | 5 | 6, None) => {
+            if matches!(schema, 3..=6) && meta.get("last_applied")?.is_none_or(|v| v.value() != [0]) {
+                return Err(StoreError::CorruptRecord);
+            }
+            Ok(None)
+        }
+        (2 | 3 | 4 | 5 | 6, Some(marker)) => {
+            let id = decode_id(marker.value())?;
+            let record = logs.get(id.index)?.ok_or(StoreError::CorruptRecord)?;
+            if decode_log(record.value())?.id != id {
+                return Err(StoreError::CorruptRecord);
+            }
+            Ok(Some(id))
+        }
+        (1 | 2, _) => Err(StoreError::CorruptRecord),
+        _ => Err(StoreError::UnsupportedSchema),
+    }
+}
+
+impl Store {
+    /// Read the exact committed boundary, validating it against the stored log.
+    pub fn read_committed(&self) -> Result<Option<LogId>, StoreError> {
+        let tx = self.database.begin_read()?;
+        validate_committed(&tx.open_table(META)?, &tx.open_table(LOGS)?)
+    }
+
+    /// Persist a monotonic committed boundary supplied by the authority owner.
+    pub fn save_committed(&mut self, id: LogId) -> Result<(), StoreError> {
+        let tx = self.write_transaction()?;
+        {
+            let mut meta = tx.open_table(META)?;
+            let logs = tx.open_table(LOGS)?;
+            let previous = validate_committed(&meta, &logs)?;
+            let record = logs.get(id.index)?.ok_or(StoreError::InvalidCommittedLog)?;
+            if decode_log(record.value())?.id != id {
+                return Err(StoreError::InvalidCommittedLog);
+            }
+            if let Some(previous) = previous {
+                if id.index < previous.index {
+                    return Err(StoreError::CommittedRegression);
+                }
+                if id == previous {
+                    return Ok(());
+                }
+            }
+            meta.insert("committed", encode_id(id).as_slice())?;
+            if previous.is_none() && !matches!(number(meta.get("schema")?.ok_or(StoreError::CorruptRecord)?.value())?, 3..=6) {
+                meta.insert("schema", 2u64.to_be_bytes().as_slice())?;
+            }
+        }
+        self.commit(tx)
+    }
+}
@@ -0,0 +1,247 @@
+use crate::{LogId, StoreError, decode_id, encode_id};
+
+pub const MAX_OBJECT_KEY_LENGTH: usize = 4096;
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ObjectIdentity {
+    pub bucket_incarnation: [u8; 16],
+    pub key: Vec<u8>,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PreparedIdentity {
+    pub version_id: [u8; 16],
+    pub preparation_id: [u8; 16],
+    pub content_digest: [u8; 32],
+    pub content_length: u64,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DecideCreated {
+    pub operation_id: [u8; 16],
+    pub object: ObjectIdentity,
+    pub expected_head: Option<[u8; 16]>,
+    pub prepared: PreparedIdentity,
+    pub binding_revision: u64,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CaptureBinding {
+    pub revision: u64,
+    pub binding_id: [u8; 16],
+    pub target_id: [u8; 16],
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum CreatedResult {
+    Created {
+        version_id: [u8; 16],
+        event_id: [u8; 16],
+        decision_log_id: LogId,
+    },
+    HeadMismatch {
+        actual_head: Option<[u8; 16]>,
+    },
+    OperationConflict,
+    BindingMismatch,
+    VersionConflict,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CreatedVersion {
+    pub prepared: PreparedIdentity,
+    pub decision_log_id: LogId,
+    pub binding_revision: u64,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CreatedEvent {
+    pub object: ObjectIdentity,
+    pub prepared: PreparedIdentity,
+    pub decision_log_id: LogId,
+    pub binding: CaptureBinding,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CreatedOperation {
+    pub canonical_command: Vec<u8>,
+    pub result: CreatedResult,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CreatedOperationProvenance {
+    pub operation: CreatedOperation,
+    pub first_decision_log_id: LogId,
+}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CreatedApplyResult {
+    pub log_id: LogId,
+    pub result: CreatedResult,
+}
+
+pub(crate) struct Decoder<'a> {
+    bytes: &'a [u8],
+}
+impl<'a> Decoder<'a> {
+    pub(crate) fn new(bytes: &'a [u8]) -> Self {
+        Self { bytes }
+    }
+    pub(crate) fn take<const N: usize>(&mut self) -> Result<[u8; N], StoreError> {
+        let a = self
+            .bytes
+            .get(..N)
+            .ok_or(StoreError::CorruptRecord)?
+            .try_into()
+            .map_err(|_| StoreError::CorruptRecord)?;
+        self.bytes = &self.bytes[N..];
+        Ok(a)
+    }
+    pub(crate) fn u64(&mut self) -> Result<u64, StoreError> {
+        Ok(u64::from_be_bytes(self.take()?))
+    }
+    pub(crate) fn uuid(&mut self) -> Result<[u8; 16], StoreError> {
+        let id = self.take()?;
+        if id == [0; 16] {
+            return Err(StoreError::CorruptRecord);
+        }
+        Ok(id)
+    }
+    pub(crate) fn option(&mut self) -> Result<Option<[u8; 16]>, StoreError> {
+        match self.take::<1>()?[0] {
+            0 => Ok(None),
+            1 => Ok(Some(self.uuid()?)),
+            _ => Err(StoreError::CorruptRecord),
+        }
+    }
+    pub(crate) fn object(&mut self) -> Result<ObjectIdentity, StoreError> {
+        let bucket_incarnation = self.uuid()?;
+        let len = usize::try_from(u32::from_be_bytes(self.take()?)).map_err(|_| StoreError::CorruptRecord)?;
+        if len > MAX_OBJECT_KEY_LENGTH {
+            return Err(StoreError::CorruptRecord);
+        }
+        let key = self.bytes.get(..len).ok_or(StoreError::CorruptRecord)?.to_vec();
+        self.bytes = &self.bytes[len..];
+        Ok(ObjectIdentity { bucket_incarnation, key })
+    }
+    pub(crate) fn prepared(&mut self) -> Result<PreparedIdentity, StoreError> {
+        Ok(PreparedIdentity {
+            version_id: self.uuid()?,
+            preparation_id: self.uuid()?,
+            content_digest: self.take()?,
+            content_length: self.u64()?,
+        })
+    }
+    pub(crate) fn id(&mut self) -> Result<LogId, StoreError> {
+        decode_id(&self.take::<24>()?)
+    }
+    pub(crate) fn binding(&mut self) -> Result<CaptureBinding, StoreError> {
+        Ok(CaptureBinding {
+            revision: self.u64()?,
+            binding_id: self.uuid()?,
+            target_id: self.uuid()?,
+        })
+    }
+    pub(crate) fn finish(self) -> Result<(), StoreError> {
+        if self.bytes.is_empty() {
+            Ok(())
+        } else {
+            Err(StoreError::CorruptRecord)
+        }
+    }
+    pub(crate) fn result(&mut self) -> Result<CreatedResult, StoreError> {
+        match self.take::<1>()?[0] {
+            0 => Ok(CreatedResult::Created {
+                version_id: self.uuid()?,
+                event_id: self.uuid()?,
+                decision_log_id: self.id()?,
+            }),
+            1 => Ok(CreatedResult::HeadMismatch {
+                actual_head: self.option()?,
+            }),
+            2 => Ok(CreatedResult::OperationConflict),
+            3 => Ok(CreatedResult::BindingMismatch),
+            4 => Ok(CreatedResult::VersionConflict),
+            _ => Err(StoreError::CorruptRecord),
+        }
+    }
+}
+pub(crate) fn put_option(b: &mut Vec<u8>, id: Option<[u8; 16]>) {
+    b.push(u8::from(id.is_some()));
+    if let Some(id) = id {
+        b.extend(id)
+    }
+}
+pub(crate) fn object_bytes(o: &ObjectIdentity) -> Result<Vec<u8>, StoreError> {
+    if o.bucket_incarnation == [0; 16] || o.key.len() > MAX_OBJECT_KEY_LENGTH {
+        return Err(StoreError::CorruptRecord);
+    }
+    let capacity = 20usize.checked_add(o.key.len()).ok_or(StoreError::CorruptRecord)?;
+    let mut b = Vec::with_capacity(capacity);
+    b.extend(o.bucket_incarnation);
+    b.extend(
+        u32::try_from(o.key.len())
+            .map_err(|_| StoreError::CorruptRecord)?
+            .to_be_bytes(),
+    );
+    b.extend(&o.key);
+    Ok(b)
+}
+pub(crate) fn put_prepared(b: &mut Vec<u8>, p: &PreparedIdentity) {
+    b.extend(p.version_id);
+    b.extend(p.preparation_id);
+    b.extend(p.content_digest);
+    b.extend(p.content_length.to_be_bytes())
+}
+pub(crate) fn binding_bytes(v: &CaptureBinding) -> Result<Vec<u8>, StoreError> {
+    let mut b = v.revision.to_be_bytes().to_vec();
+    b.extend(v.binding_id);
+    b.extend(v.target_id);
+    let mut d = Decoder::new(&b);
+    d.binding()?;
+    d.finish()?;
+    Ok(b)
+}
+pub(crate) fn result_bytes(r: &CreatedResult) -> Vec<u8> {
+    let mut b = Vec::new();
+    match r {
+        CreatedResult::Created {
+            version_id,
+            event_id,
+            decision_log_id,
+        } => {
+            b.push(0);
+            b.extend(version_id);
+            b.extend(event_id);
+            b.extend(encode_id(*decision_log_id))
+        }
+        CreatedResult::HeadMismatch { actual_head } => {
+            b.push(1);
+            put_option(&mut b, *actual_head)
+        }
+        CreatedResult::OperationConflict => b.push(2),
+        CreatedResult::BindingMismatch => b.push(3),
+        CreatedResult::VersionConflict => b.push(4),
+    }
+    b
+}
+impl DecideCreated {
+    pub fn encode(&self) -> Result<Vec<u8>, StoreError> {
+        let object = object_bytes(&self.object)?;
+        let capacity = object.len().checked_add(134).ok_or(StoreError::CorruptRecord)?;
+        let mut b = Vec::with_capacity(capacity);
+        b.extend(b"CRTD\x01");
+        b.extend(self.operation_id);
+        b.extend(object);
+        put_option(&mut b, self.expected_head);
+        put_prepared(&mut b, &self.prepared);
+        b.extend(self.binding_revision.to_be_bytes());
+        Self::decode(&b)?;
+        Ok(b)
+    }
+    pub fn decode(bytes: &[u8]) -> Result<Self, StoreError> {
+        let mut d = Decoder::new(bytes);
+        if d.take::<5>()? != *b"CRTD\x01" {
+            return Err(StoreError::CorruptRecord);
+        }
+        let c = Self {
+            operation_id: d.uuid()?,
+            object: d.object()?,
+            expected_head: d.option()?,
+            prepared: d.prepared()?,
+            binding_revision: d.u64()?,
+        };
+        d.finish()?;
+        Ok(c)
+    }
+}
@@ -0,0 +1,118 @@
+//! Owned consistent native application snapshot view.
+use crate::{Store, StoreError, application, snapshot};
+use redb::{ReadTransaction, ReadableDatabase, ReadableTable, ReadableTableMetadata};
+use std::{
+    io::Write,
+    sync::{Arc, Mutex},
+};
+
+#[derive(Debug, thiserror::Error)]
+pub enum NativeSnapshotError {
+    #[error(transparent)]
+    Store(#[from] StoreError),
+    #[error(transparent)]
+    Snapshot(#[from] snapshot::SnapshotError),
+    #[error("native snapshot Store lock poisoned")]
+    Poisoned,
+    #[error("native snapshot worker failed")]
+    Worker(#[source] tokio::task::JoinError),
+}
+
+pub struct NativeSnapshotView {
+    // Drop the transaction before the owner so the Database stays open throughout the view.
+    transaction: ReadTransaction,
+    header: snapshot::Header,
+    counts: [u64; 4],
+    schema: u64,
+    _owner: Arc<Mutex<Store>>,
+}
+impl NativeSnapshotView {
+    pub(crate) fn acquire(owner: Arc<Mutex<Store>>) -> Result<Self, NativeSnapshotError> {
+        let transaction = {
+            let store = owner.lock().map_err(|_| NativeSnapshotError::Poisoned)?;
+            if !store.writable {
+                return Err(StoreError::WriteDisabled.into());
+            }
+            store.database.begin_read().map_err(StoreError::from)?
+        };
+        let (applied, membership) = application::native_application_state(&transaction)?;
+        let (header, schema) = {
+            let meta = transaction.open_table(crate::META).map_err(StoreError::from)?;
+            let identity = meta
+                .get("identity")
+                .map_err(StoreError::from)?
+                .ok_or(StoreError::CorruptRecord)?;
+            let schema = crate::number(
+                meta.get("schema")
+                    .map_err(StoreError::from)?
+                    .ok_or(StoreError::CorruptRecord)?
+                    .value(),
+            )?;
+            (
+                snapshot::Header {
+                    cluster_id: crate::decode_identity(identity.value())?.cluster_id,
+                    applied,
+                    membership,
+                    binding: application::read_binding(&meta)?,
+                },
+                schema,
+            )
+        };
+        let mut counts = [0; 4];
+        for (index, definition) in [
+            application::HEADS,
+            application::VERSIONS,
+            application::EVENTS,
+            application::OPERATIONS,
+        ]
+        .into_iter()
+        .enumerate()
+        {
+            counts[index] = transaction
+                .open_table(definition)
+                .map_err(StoreError::from)?
+                .len()
+                .map_err(StoreError::from)?;
+        }
+        Ok(Self {
+            transaction,
+            header,
+            counts,
+            schema,
+            _owner: owner,
+        })
+    }
+
+    pub fn stream_capability_artifact<W: Write>(
+        self,
+        writer: W,
+        limits: snapshot::Limits,
+    ) -> Result<snapshot::CapabilityArtifact<W>, NativeSnapshotError> {
+        if self.schema != 6 {
+            return Err(StoreError::UnsupportedSchema.into());
+        }
+        let output = snapshot::CapabilityOutput::new(writer, limits.total_bytes);
+        Ok(self.stream(output, limits)?.finish())
+    }
+
+    pub fn stream<W: Write>(self, writer: W, limits: snapshot::Limits) -> Result<W, NativeSnapshotError> {
+        let mut encoder = snapshot::Encoder::new(writer, &self.header, self.counts, limits)?;
+        for (definition, section) in [
+            (application::HEADS, snapshot::Section::Heads),
+            (application::VERSIONS, snapshot::Section::Versions),
+            (application::EVENTS, snapshot::Section::Events),
+            (application::OPERATIONS, snapshot::Section::Operations),
+        ] {
+            let table = self.transaction.open_table(definition).map_err(StoreError::from)?;
+            for row in table.iter().map_err(StoreError::from)? {
+                let (key, value) = row.map_err(StoreError::from)?;
+                encoder.row(snapshot::Row {
+                    section,
+                    key: key.value(),
+                    value: value.value(),
+                })?;
+            }
+        }
+        Ok(encoder.finish()?)
+    }
+}
@@ -0,0 +1,371 @@
+//! Validated capture peer metadata and its independent CNOD v1 codec.
+use serde::{Deserialize, Deserializer, Serialize, Serializer, ser::SerializeStruct};
+
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct CaptureNode {
+    rpc_uri: String,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum CaptureNodeError {
+    #[error("invalid RPC URI")]
+    InvalidRpcUri,
+    #[error("invalid RPC URI")]
+    InvalidRpcUriParse(#[source] url::ParseError),
+    #[error("RPC URI too long")]
+    RpcUriTooLong,
+    #[error("invalid node magic")]
+    InvalidNodeMagic,
+    #[error("unsupported node version")]
+    UnsupportedNodeVersion,
+    #[error("invalid node length")]
+    InvalidNodeLength,
+    #[error("truncated node record")]
+    TruncatedNodeRecord,
+    #[error("invalid node UTF-8")]
+    InvalidNodeUtf8(#[source] std::str::Utf8Error),
+    #[error("trailing node bytes")]
+    TrailingNodeBytes,
+}
+impl CaptureNodeError {
+    pub fn code(&self) -> &'static str {
+        match self {
+            Self::InvalidRpcUri | Self::InvalidRpcUriParse(_) => "invalid_rpc_uri",
+            Self::RpcUriTooLong => "rpc_uri_too_long",
+            Self::InvalidNodeMagic => "invalid_node_magic",
+            Self::UnsupportedNodeVersion => "unsupported_node_version",
+            Self::InvalidNodeLength => "invalid_node_length",
+            Self::TruncatedNodeRecord => "truncated_node_record",
+            Self::InvalidNodeUtf8(_) => "invalid_node_utf8",
+            Self::TrailingNodeBytes => "trailing_node_bytes",
+        }
+    }
+}
+impl CaptureNode {
+    pub fn new(rpc_uri: String) -> Result<Self, CaptureNodeError> {
+        let node = Self { rpc_uri };
+        node.validate()?;
+        Ok(node)
+    }
+
+    pub fn rpc_uri(&self) -> &str {
+        &self.rpc_uri
+    }
+
+    pub fn validate(&self) -> Result<(), CaptureNodeError> {
+        let uri = &self.rpc_uri;
+        if uri.len() > 4096 {
+            return Err(CaptureNodeError::RpcUriTooLong);
+        }
+        if uri.is_empty() || uri.chars().any(|c| c.is_whitespace() || c.is_control() || c == '\\') {
+            return Err(CaptureNodeError::InvalidRpcUri);
+        }
+        let rest = uri
+            .strip_prefix("http://")
+            .or_else(|| uri.strip_prefix("https://"))
+            .ok_or(CaptureNodeError::InvalidRpcUri)?;
+        if uri.contains('?') || uri.contains('#') {
+            return Err(CaptureNodeError::InvalidRpcUri);
+        }
+        let (authority, path) = rest.split_once('/').map_or((rest, ""), |(a, p)| (a, p));
+        // An empty suffix after the first slash is the only allowed path.
+        if !path.is_empty() || authority.contains('@') {
+            return Err(CaptureNodeError::InvalidRpcUri);
+        }
+        let (host, port) = if authority.starts_with('[') {
+            let end = authority.find(']').ok_or(CaptureNodeError::InvalidRpcUri)?;
+            let port = authority
+                .get(end + 1..)
+                .and_then(|suffix| suffix.strip_prefix(':'))
+                .ok_or(CaptureNodeError::InvalidRpcUri)?;
+            (&authority[..=end], port)
+        } else {
+            authority.split_once(':').ok_or(CaptureNodeError::InvalidRpcUri)?
+        };
+        if host.is_empty() || port.is_empty() || !port.bytes().all(|b| b.is_ascii_digit()) {
+            return Err(CaptureNodeError::InvalidRpcUri);
+        }
+        let port = port.parse::<u16>().map_err(|_| CaptureNodeError::InvalidRpcUri)?;
+        if port == 0 {
+            return Err(CaptureNodeError::InvalidRpcUri);
+        }
+        let parsed_url = url::Url::parse(uri).map_err(CaptureNodeError::InvalidRpcUriParse)?;
+        if parsed_url.host().is_none() {
+            return Err(CaptureNodeError::InvalidRpcUri);
+        }
+        Ok(())
+    }
+
+    pub fn encode(&self) -> Result<Vec<u8>, CaptureNodeError> {
+        self.validate()?;
+        let length = u32::try_from(self.rpc_uri.len()).map_err(|_| CaptureNodeError::InvalidNodeLength)?;
+        let mut bytes = b"CNOD\x01".to_vec();
+        bytes.extend_from_slice(&length.to_be_bytes());
+        bytes.extend_from_slice(self.rpc_uri.as_bytes());
+        Ok(bytes)
+    }
+
+    pub fn decode(bytes: &[u8]) -> Result<Self, CaptureNodeError> {
+        if bytes.len() < 9 {
+            return Err(CaptureNodeError::TruncatedNodeRecord);
+        }
+        if &bytes[..4] != b"CNOD" {
+            return Err(CaptureNodeError::InvalidNodeMagic);
+        }
+        if bytes[4] != 1 {
+            return Err(CaptureNodeError::UnsupportedNodeVersion);
+        }
+        let length = u32::from_be_bytes(bytes[5..9].try_into().map_err(|_| CaptureNodeError::InvalidNodeLength)?);
+        if !(1..=4096).contains(&length) {
+            return Err(CaptureNodeError::InvalidNodeLength);
+        }
+        let length = usize::try_from(length).map_err(|_| CaptureNodeError::InvalidNodeLength)?;
+        let end = 9_usize.checked_add(length).ok_or(CaptureNodeError::InvalidNodeLength)?;
+        if bytes.len() < end {
+            return Err(CaptureNodeError::TruncatedNodeRecord);
+        }
+        if bytes.len() != end {
+            return Err(CaptureNodeError::TrailingNodeBytes);
+        }
+        let uri = std::str::from_utf8(&bytes[9..end]).map_err(CaptureNodeError::InvalidNodeUtf8)?;
+        Self::new(uri.to_owned())
+    }
+}
+impl Serialize for CaptureNode {
+    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        self.validate().map_err(serde::ser::Error::custom)?;
+        let mut state = serializer.serialize_struct("CaptureNode", 1)?;
+        state.serialize_field("rpc_uri", &self.rpc_uri)?;
+        state.end()
+    }
+}
+impl<'de> Deserialize<'de> for CaptureNode {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        #[derive(Deserialize)]
+        #[serde(deny_unknown_fields)]
+        struct Fields {
+            rpc_uri: String,
+        }
+        let fields = Fields::deserialize(deserializer).map_err(|_| serde::de::Error::custom("invalid capture node fields"))?;
+        Self::new(fields.rpc_uri).map_err(serde::de::Error::custom)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::error::Error;
+
+    fn record(payload: &[u8]) -> Vec<u8> {
+        let mut bytes = b"CNOD\x01".to_vec();
+        bytes.extend_from_slice(&u32::try_from(payload.len()).expect("test payload length").to_be_bytes());
+        bytes.extend_from_slice(payload);
+        bytes
+    }
+    fn reject_uri(uri: &str) {
+        let result = CaptureNode::new(uri.to_owned());
+        assert!(result.is_err(), "URI should reject: {uri:?}");
+        assert_eq!(result.expect_err("rejected URI").code(), "invalid_rpc_uri");
+    }
+    fn reject_record(bytes: &[u8], code: &str) {
+        let result = CaptureNode::decode(bytes);
+        assert!(result.is_err(), "record should reject with {code}");
+        assert_eq!(result.expect_err("rejected record").code(), code);
+    }
+    #[test]
+    fn actual_openraft_node_bound() {
+        fn require_node<T: openraft::Node>() {}
+        require_node::<CaptureNode>();
+        assert_eq!(CaptureNode::default().rpc_uri(), "");
+    }
+    #[test]
+    fn default_and_empty_rejected_at_every_boundary() {
+        let node = CaptureNode::default();
+        assert_eq!(node.rpc_uri(), "");
+        reject_uri("");
+        assert_eq!(node.validate().expect_err("empty validation").code(), "invalid_rpc_uri");
+        assert!(serde_json::to_string(&node).is_err());
+        assert!(serde_json::from_str::<CaptureNode>(r#"{"rpc_uri":""}"#).is_err());
+        assert_eq!(node.encode().expect_err("empty encoding").code(), "invalid_rpc_uri");
+        reject_record(&record(b""), "invalid_node_length");
+    }
+    #[test]
+    fn accepted_uris_preserve_exact_bytes() {
+        for uri in [
+            "http://node:80",
+            "https://node:443",
+            "http://node:1",
+            "https://node:65535/",
+            "http://127.0.0.1:9000",
+            "https://[::1]:443/",
+            "http://[2001:db8::1]:1",
+            "http://NoDe:00080/",
+        ] {
+            let result = CaptureNode::new(uri.to_owned());
+            assert!(result.is_ok(), "valid URI construction: {uri}");
+            let node = result.expect("valid URI");
+            assert_eq!(node.rpc_uri(), uri, "preserve explicit default port and raw URI");
+            assert_eq!(node.encode().expect("encode URI"), record(uri.as_bytes()));
+            assert_eq!(CaptureNode::decode(&record(uri.as_bytes())).expect("decode URI"), node);
+        }
+    }
+    #[test]
+    fn golden_cnod_bytes() {
+        let node = CaptureNode {
+            rpc_uri: "http://node:80".to_owned(),
+        };
+        let golden = b"CNOD\x01\x00\x00\x00\x0ehttp://node:80";
+        assert_eq!(node.encode().ok().as_deref(), Some(golden.as_slice()), "golden CNOD bytes");
+    }
+    #[test]
+    fn valid_cnod_decode() {
+        let golden = b"CNOD\x01\x00\x00\x00\x0ehttp://node:80";
+        let decoded = CaptureNode::decode(golden);
+        assert!(decoded.is_ok(), "golden decode succeeds");
+        assert_eq!(decoded.expect("golden decoded node").rpc_uri(), "http://node:80");
+    }
+    #[test]
+    fn serde_shape_roundtrip_and_strict_fields() {
+        let node = CaptureNode::new("https://node:443/".to_owned()).expect("serde node");
+        let json = r#"{"rpc_uri":"https://node:443/"}"#;
+        assert_eq!(serde_json::to_string(&node).expect("serialize node"), json);
+        assert_eq!(serde_json::from_str::<CaptureNode>(json).expect("deserialize node"), node);
+        for json in [
+            r#"{}"#,
+            r#"{"rpc_uri":1}"#,
+            r#"{"rpc_uri":null}"#,
+            r#"{"rpc_uri":"http://node:1","extra":0}"#,
+            r#"{"rpc_uri":"http://node:1","rpc_uri":"http://node:2"}"#,
+        ] {
+            assert!(serde_json::from_str::<CaptureNode>(json).is_err(), "strict fields: {json}");
+        }
+    }
+    #[test]
+    fn credentials_and_raw_uri_shapes_reject() {
+        for uri in [
+            "http://user@node:1",
+            "http://user:password@node:1",
+            "http://@node:1",
+            "HTTP://node:1",
+            "Https://node:1",
+            "http://node",
+            "http://node:",
+            "http://node:0",
+            "http://node:-1",
+            "http://node:+1",
+            "http://node:65536",
+            "http://node:999999999999999999999",
+            "http://[::1:1",
+            "http://[::1]extra:1",
+            "http://[invalid]:1",
+            "http://::1:1",
+            "http://:1",
+            "/disk/path",
+            "file:///disk/path",
+            "http://node:1/./",
+            "http://node:1/../",
+            "http://node:1/%2e/",
+            "http://node:1/%2E%2e/",
+            "http://node:1/path",
+            "http://node:1\\",
+            " http://node:1",
+            "http://node:1\n",
+            "http://no\tde:1",
+            "http://node:1\u{00a0}",
+            "http://node:1\u{0085}",
+            "http://node:1\0",
+        ] {
+            reject_uri(uri);
+        }
+    }
+    #[test]
+    fn query_and_fragment_markers_reject() {
+        for uri in [
+            "http://node:1?",
+            "http://node:1?x=y",
+            "http://node:1/?",
+            "http://node:1#",
+            "http://node:1#fragment",
+            "http://node:1/#",
+        ] {
+            reject_uri(uri);
+        }
+    }
+    #[test]
+    fn utf8_byte_length_boundaries() {
+        for length in [4095, 4096] {
+            let prefix = "http://node:";
+            let uri = format!("{prefix}{}1", "0".repeat(length - prefix.len() - 1));
+            assert_eq!(uri.len(), length);
+            assert!(url::Url::parse(&uri).is_ok(), "boundary fixture accepted by actual URL parser");
+            let node = CaptureNode::new(uri.clone()).expect("boundary node");
+            assert_eq!(node.rpc_uri(), uri);
+            assert_eq!(
+                CaptureNode::decode(&node.encode().expect("boundary encoding")).expect("boundary decoding"),
+                node
+            );
+        }
+        let uri = format!("http://node:{}1", "0".repeat(4097 - "http://node:".len() - 1));
+        assert_eq!(CaptureNode::new(uri).expect_err("oversize URI").code(), "rpc_uri_too_long");
+        let unicode = format!("http://{}:1", "é".repeat(2044));
+        assert!(unicode.chars().count() < 4096);
+        assert!(unicode.len() > 4096);
+        assert_eq!(CaptureNode::new(unicode).expect_err("UTF-8 byte limit").code(), "rpc_uri_too_long");
+    }
+    #[test]
+    fn cnod_headers_and_lengths_reject_exact_codes() {
+        let valid = record(b"http://node:80");
+        for end in 0..9 {
+            reject_record(&valid[..end], "truncated_node_record");
+        }
+        let mut bad = valid.clone();
+        bad[0] = b'X';
+        reject_record(&bad, "invalid_node_magic");
+        bad = valid.clone();
+        bad[4] = 2;
+        reject_record(&bad, "unsupported_node_version");
+        for length in [0_u32, 4097, u32::MAX] {
+            bad = b"CNOD\x01".to_vec();
+            bad.extend_from_slice(&length.to_be_bytes());
+            reject_record(&bad, "invalid_node_length");
+        }
+        for end in 9..valid.len() {
+            reject_record(&valid[..end], "truncated_node_record");
+        }
+        reject_record(&record(&[0xff]), "invalid_node_utf8");
+        reject_record(&record(&[0xc3, 0x28]), "invalid_node_utf8");
+        reject_record(&record(b"invalid URI"), "invalid_rpc_uri");
+    }
+    #[test]
+    fn trailing_bytes_reject() {
+        for tail in [&b"x"[..], &b"xyz"[..]] {
+            let mut bytes = record(b"http://node:80");
+            bytes.extend_from_slice(tail);
+            reject_record(&bytes, "trailing_node_bytes");
+        }
+    }
+    #[test]
+    fn errors_and_sources_redact_input() {
+        let sentinel = "http://SECRET_USER:SECRET_PASSWORD@node:1";
+        let error = CaptureNode::new(sentinel.to_owned()).expect_err("credentials reject");
+        let mut current: &dyn Error = &error;
+        loop {
+            let rendered = format!("{current} {current:?}");
+            assert!(!rendered.contains("SECRET"));
+            match current.source() {
+                Some(source) => current = source,
+                None => break,
+            }
+        }
+        let json = format!(r#"{{"rpc_uri":"{sentinel}"}}"#);
+        let error = serde_json::from_str::<CaptureNode>(&json).expect_err("serde credentials reject");
+        assert!(!format!("{error} {error:?}").contains("SECRET"));
+        let json = format!(r#"{{"{sentinel}":0}}"#);
+        let error = serde_json::from_str::<CaptureNode>(&json).expect_err("unknown fields reject");
+        assert!(!format!("{error} {error:?}").contains("SECRET"));
+        let error = CaptureNode::new("http://[bad]:1".to_owned()).expect_err("parser rejection");
+        assert!(error.source().is_some(), "retain URL parser source");
+        let error = CaptureNode::decode(&record(&[0xff])).expect_err("UTF-8 rejection");
+        assert!(error.source().is_some(), "retain UTF-8 source");
+    }
+}
@@ -0,0 +1,43 @@
+use crate::{LOGS, META, Store, StoreError, decode_id, decode_log, encode_id};
+use redb::ReadableTable;
+
+impl Store {
+    /// Atomically remove every log entry at or after `from`.
+    pub fn truncate_suffix(&mut self, from: u64) -> Result<(), StoreError> {
+        let tx = self.write_transaction()?;
+        {
+            let mut meta = tx.open_table(META)?;
+            let logs = tx.open_table(LOGS)?;
+            if crate::committed::validate_committed(&meta, &logs)?.is_some_and(|id| from <= id.index) {
+                return Err(StoreError::CommittedLogProtected);
+            }
+            drop(logs);
+            let tail = meta.get("tail")?.map(|v| decode_id(v.value())).transpose()?;
+            if tail.is_none_or(|id| from > id.index) {
+                return Ok(());
+            }
+            let mut logs = tx.open_table(LOGS)?;
+            let new_tail = if from == 0 {
+                None
+            } else {
+                let index = from - 1;
+                let record = logs.get(index)?.ok_or(StoreError::InvalidLogSequence)?;
+                let id = decode_log(record.value())?.id;
+                if id.index != index {
+                    return Err(StoreError::CorruptRecord);
+                }
+                Some(id)
+            };
+            logs.retain(|index, _| index < from)?;
+            match new_tail {
+                Some(id) => {
+                    meta.insert("tail", encode_id(id).as_slice())?;
+                }
+                None => {
+                    meta.remove("tail")?;
+                }
+            }
+        }
+        self.commit(tx)
+    }
+}
@@ -1,512 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! What one stack answered, and how two answers are compared.
-//!
-//! An [`Observation`] keeps only what a client acts on: the HTTP status, the S3 error code, the
-//! allowlisted response headers and a few typed fields decoded from the response. The error
-//! message is kept for diagnostics and never compared: message wording is judged by the wire
-//! corpus, not here. The comparator is pure so it can be tested without a server.
-
-use serde::Serialize;
-use std::collections::{BTreeMap, BTreeSet};
-use std::path::{Path, PathBuf};
-
-/// Response headers compared exactly (after [`normalized_header_value`]).
-const HEADER_ALLOWLIST: [&str; 7] = [
-    "content-type",
-    "x-amz-version-id",
-    "etag",
-    "x-amz-delete-marker",
-    "content-range",
-    "accept-ranges",
-    "x-amz-request-charged",
-];
-
-/// Every response header with this prefix is compared as well.
-const HEADER_ALLOWLIST_PREFIX: &str = "x-amz-server-side-encryption";
-
-/// Placeholder for a field one side did not send.
-const ABSENT: &str = "<absent>";
-
-/// The lowercase header name if the comparator looks at this header.
-pub(crate) fn compared_header(name: &str) -> Option<String> {
-    let name = name.to_ascii_lowercase();
-    (HEADER_ALLOWLIST.contains(&name.as_str()) || name.starts_with(HEADER_ALLOWLIST_PREFIX)).then_some(name)
-}
-
-/// A version ID is minted per process, so its value never matches across two servers; only
-/// whether it is the literal `null` (an unversioned write) carries meaning.
-fn normalized_header_value(name: &str, value: &str) -> String {
-    if name == "x-amz-version-id" && value != "null" {
-        "<version-id>".to_owned()
-    } else {
-        value.to_owned()
-    }
-}
-
-/// One request's outcome as a client sees it.
-#[derive(Clone, Debug, Serialize)]
-pub(crate) struct Observation {
-    /// The S3 operation, named as in the `s3s::S3` trait (`GetObject`).
-    pub op: &'static str,
-    /// Which request of the script this was (`normal`, `negative`, or a setup label).
-    pub case: &'static str,
-    /// HTTP status; 0 when no response arrived.
-    pub status: u16,
-    /// The S3 error code, or the SDK's error kind when there was no service error.
-    pub error_code: Option<String>,
-    /// Allowlisted headers only, lowercase names, normalized values.
-    pub headers: BTreeMap<String, String>,
-    /// Typed values decoded from the response body, never free text.
-    pub fields: BTreeMap<String, String>,
-    /// Diagnostic only; never compared.
-    #[serde(skip)]
-    pub message: Option<String>,
-}
-
-impl Observation {
-    /// Builds an observation, keeping only the allowlisted headers.
-    pub(crate) fn new<'h>(
-        op: &'static str,
-        case: &'static str,
-        status: u16,
-        error_code: Option<String>,
-        headers: impl IntoIterator<Item = (&'h str, &'h str)>,
-    ) -> Self {
-        let headers = headers
-            .into_iter()
-            .filter_map(|(name, value)| compared_header(name).map(|name| (name, value)))
-            .map(|(name, value)| {
-                let value = normalized_header_value(&name, value);
-                (name, value)
-            })
-            .collect();
-        Self {
-            op,
-            case,
-            status,
-            error_code,
-            headers,
-            fields: BTreeMap::new(),
-            message: None,
-        }
-    }
-
-    pub(crate) fn with_field(mut self, name: &str, value: impl Into<String>) -> Self {
-        self.fields.insert(name.to_owned(), value.into());
-        self
-    }
-
-    pub(crate) fn is_success(&self) -> bool {
-        (200..300).contains(&self.status)
-    }
-}
-
-/// One field on which the two stacks answered differently.
-#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
-pub(crate) struct Difference {
-    pub op: String,
-    pub case: String,
-    pub field: String,
-    pub a: String,
-    pub b: String,
-}
-
-/// Every difference between two transcripts of the same script, matched by `(op, case)`.
-pub(crate) fn compare(a: &[Observation], b: &[Observation]) -> Vec<Difference> {
-    let mut differences = Vec::new();
-    let (index_a, duplicates_a) = index(a);
-    let (index_b, duplicates_b) = index(b);
-    for (op, case) in duplicates_a.union(&duplicates_b) {
-        let count = |side: &[Observation]| {
-            side.iter()
-                .filter(|observation| (observation.op, observation.case) == (*op, *case))
-                .count()
-                .to_string()
-        };
-        differences.push(difference(op, case, "duplicate_case", count(a), count(b)));
-    }
-    let keys: BTreeSet<_> = index_a.keys().chain(index_b.keys()).copied().collect();
-    for key @ (op, case) in keys {
-        match (index_a.get(&key), index_b.get(&key)) {
-            (Some(a), Some(b)) => compare_one(a, b, &mut differences),
-            (a, b) => {
-                let presence = |side: Option<&&Observation>| if side.is_some() { "present" } else { ABSENT }.to_owned();
-                differences.push(difference(op, case, "observation", presence(a), presence(b)));
-            }
-        }
-    }
-    differences
-}
-
-type Key = (&'static str, &'static str);
-
-fn index(side: &[Observation]) -> (BTreeMap<Key, &Observation>, BTreeSet<Key>) {
-    let mut index = BTreeMap::new();
-    let mut duplicates = BTreeSet::new();
-    for observation in side {
-        let key = (observation.op, observation.case);
-        if index.insert(key, observation).is_some() {
-            duplicates.insert(key);
-        }
-    }
-    (index, duplicates)
-}
-
-fn difference(op: &str, case: &str, field: &str, a: String, b: String) -> Difference {
-    Difference {
-        op: op.to_owned(),
-        case: case.to_owned(),
-        field: field.to_owned(),
-        a,
-        b,
-    }
-}
-
-fn compare_one(a: &Observation, b: &Observation, differences: &mut Vec<Difference>) {
-    let mut field = |name: String, value_a: Option<&str>, value_b: Option<&str>| {
-        if value_a != value_b {
-            let show = |value: Option<&str>| value.unwrap_or(ABSENT).to_owned();
-            differences.push(difference(a.op, a.case, &name, show(value_a), show(value_b)));
-        }
-    };
-    field(
-        "status".to_owned(),
-        Some(a.status.to_string().as_str()),
-        Some(b.status.to_string().as_str()),
-    );
-    field("error_code".to_owned(), a.error_code.as_deref(), b.error_code.as_deref());
-    let headers: BTreeSet<_> = a.headers.keys().chain(b.headers.keys()).collect();
-    for name in headers {
-        field(
-            format!("header.{name}"),
-            a.headers.get(name).map(String::as_str),
-            b.headers.get(name).map(String::as_str),
-        );
-    }
-    let fields: BTreeSet<_> = a.fields.keys().chain(b.fields.keys()).collect();
-    for name in fields {
-        field(
-            format!("body.{name}"),
-            a.fields.get(name).map(String::as_str),
-            b.fields.get(name).map(String::as_str),
-        );
-    }
-}
-
-/// Operations observed on both sides.
-pub(crate) fn compared_ops(a: &[Observation], b: &[Observation]) -> BTreeSet<&'static str> {
-    let ops_b: BTreeSet<_> = b.iter().map(|observation| observation.op).collect();
-    a.iter()
-        .map(|observation| observation.op)
-        .filter(|op| ops_b.contains(op))
-        .collect()
-}
-
-/// `(op, case)` pairs observed on both sides.
-pub(crate) fn compared_cases(a: &[Observation], b: &[Observation]) -> usize {
-    let cases_b: BTreeSet<_> = b.iter().map(|observation| (observation.op, observation.case)).collect();
-    a.iter()
-        .map(|observation| (observation.op, observation.case))
-        .collect::<BTreeSet<_>>()
-        .intersection(&cases_b)
-        .count()
-}
-
-/// The JSON report written for every run.
-#[derive(Debug, Serialize)]
-pub(crate) struct Report {
-    pub schema: u32,
-    pub run: String,
-    pub stack_a: String,
-    pub stack_b: String,
-    pub region_a: Option<String>,
-    pub region_b: Option<String>,
-    pub compared_ops: usize,
-    pub compared_cases: usize,
-    pub differences: Vec<Difference>,
-}
-
-impl Report {
-    /// Writes the report as `<directory>/<run>.json` and returns its path.
-    pub(crate) fn write_to(&self, directory: &Path) -> std::io::Result<PathBuf> {
-        std::fs::create_dir_all(directory)?;
-        let path = directory.join(format!("{}.json", self.run));
-        let json = serde_json::to_vec_pretty(self).map_err(std::io::Error::other)?;
-        std::fs::write(&path, json)?;
-        Ok(path)
-    }
-}
-
-/// One line per difference, for a failing assertion.
-pub(crate) fn render(differences: &[Difference]) -> String {
-    differences
-        .iter()
-        .map(|d| format!("({}, {}, {}, a={:?}, b={:?})", d.op, d.case, d.field, d.a, d.b))
-        .collect::<Vec<_>>()
-        .join("\n")
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    fn ok(op: &'static str) -> Observation {
-        Observation::new(op, "normal", 200, None, [("content-type", "application/xml")])
-    }
-
-    fn fields(differences: &[Difference]) -> Vec<&str> {
-        differences.iter().map(|d| d.field.as_str()).collect()
-    }
-
-    #[test]
-    fn identical_transcripts_have_no_difference() {
-        let a = vec![
-            ok("HeadBucket"),
-            ok("GetBucketLocation").with_field("location_constraint", "us-east-1"),
-        ];
-        let b = a.clone();
-        assert_eq!(compare(&a, &b), Vec::new());
-        assert_eq!(compared_ops(&a, &b).len(), 2);
-    }
-
-    #[test]
-    fn a_status_difference_is_reported() {
-        let a = vec![ok("PutBucketAcl")];
-        let b = vec![Observation::new(
-            "PutBucketAcl",
-            "normal",
-            403,
-            None,
-            [("content-type", "application/xml")],
-        )];
-        assert_eq!(
-            compare(&a, &b),
-            vec![Difference {
-                op: "PutBucketAcl".into(),
-                case: "normal".into(),
-                field: "status".into(),
-                a: "200".into(),
-                b: "403".into(),
-            }]
-        );
-    }
-
-    #[test]
-    fn an_error_code_difference_is_reported() {
-        let a = vec![Observation::new("GetObject", "negative", 404, Some("NoSuchKey".into()), [])];
-        let b = vec![Observation::new(
-            "GetObject",
-            "negative",
-            404,
-            Some("NoSuchBucket".into()),
-            [],
-        )];
-        let differences = compare(&a, &b);
-        assert_eq!(fields(&differences), ["error_code"]);
-        assert_eq!((differences[0].a.as_str(), differences[0].b.as_str()), ("NoSuchKey", "NoSuchBucket"));
-    }
-
-    #[test]
-    fn an_error_code_on_one_side_only_is_reported() {
-        let a = vec![Observation::new("HeadBucket", "negative", 404, Some("NotFound".into()), [])];
-        let b = vec![Observation::new("HeadBucket", "negative", 404, None, [])];
-        let differences = compare(&a, &b);
-        assert_eq!(fields(&differences), ["error_code"]);
-        assert_eq!(differences[0].b, ABSENT);
-    }
-
-    #[test]
-    fn an_allowlisted_header_difference_is_reported() {
-        let a = vec![Observation::new("HeadObject", "normal", 200, None, [("ETag", "\"aa\"")])];
-        let b = vec![Observation::new("HeadObject", "normal", 200, None, [("etag", "\"bb\"")])];
-        assert_eq!(fields(&compare(&a, &b)), ["header.etag"]);
-    }
-
-    #[test]
-    fn a_header_sent_by_one_side_only_is_reported() {
-        let a = vec![Observation::new(
-            "DeleteObject",
-            "normal",
-            204,
-            None,
-            [("x-amz-delete-marker", "true")],
-        )];
-        let b = vec![Observation::new("DeleteObject", "normal", 204, None, [])];
-        let differences = compare(&a, &b);
-        assert_eq!(fields(&differences), ["header.x-amz-delete-marker"]);
-        assert_eq!((differences[0].a.as_str(), differences[0].b.as_str()), ("true", ABSENT));
-    }
-
-    #[test]
-    fn every_server_side_encryption_header_is_compared() {
-        let header = "x-amz-server-side-encryption-customer-key-md5";
-        let a = vec![Observation::new("PutObject", "sse-c", 200, None, [(header, "one")])];
-        let b = vec![Observation::new("PutObject", "sse-c", 200, None, [(header, "two")])];
-        assert_eq!(fields(&compare(&a, &b)), [format!("header.{header}")]);
-    }
-
-    #[test]
-    fn headers_outside_the_allowlist_are_not_compared() {
-        let a = vec![Observation::new(
-            "ListBuckets",
-            "normal",
-            200,
-            None,
-            [
-                ("date", "Mon"),
-                ("x-amz-request-id", "1"),
-                ("server", "a"),
-                ("content-length", "10"),
-            ],
-        )];
-        let b = vec![Observation::new(
-            "ListBuckets",
-            "normal",
-            200,
-            None,
-            [
-                ("date", "Tue"),
-                ("x-amz-request-id", "2"),
-                ("server", "b"),
-                ("content-length", "11"),
-            ],
-        )];
-        assert_eq!(compare(&a, &b), Vec::new());
-        assert!(a[0].headers.is_empty(), "non-allowlisted headers were kept: {:?}", a[0].headers);
-    }
-
-    #[test]
-    fn distinct_version_ids_are_not_a_difference() {
-        let a = vec![Observation::new(
-            "PutObject",
-            "normal",
-            200,
-            None,
-            [("x-amz-version-id", "0b1c")],
-        )];
-        let b = vec![Observation::new(
-            "PutObject",
-            "normal",
-            200,
-            None,
-            [("x-amz-version-id", "9f8e")],
-        )];
-        assert_eq!(compare(&a, &b), Vec::new());
-    }
-
-    #[test]
-    fn a_null_version_id_against_a_real_one_is_reported() {
-        let a = vec![Observation::new(
-            "PutObject",
-            "normal",
-            200,
-            None,
-            [("x-amz-version-id", "null")],
-        )];
-        let b = vec![Observation::new(
-            "PutObject",
-            "normal",
-            200,
-            None,
-            [("x-amz-version-id", "9f8e")],
-        )];
-        assert_eq!(fields(&compare(&a, &b)), ["header.x-amz-version-id"]);
-    }
-
-    #[test]
-    fn a_body_field_difference_is_reported() {
-        let a = vec![ok("GetBucketLocation").with_field("location_constraint", "us-east-1")];
-        let b = vec![ok("GetBucketLocation").with_field("location_constraint", "eu-west-1")];
-        let differences = compare(&a, &b);
-        assert_eq!(fields(&differences), ["body.location_constraint"]);
-        assert_eq!((differences[0].a.as_str(), differences[0].b.as_str()), ("us-east-1", "eu-west-1"));
-    }
-
-    #[test]
-    fn a_body_field_on_one_side_only_is_reported() {
-        let a = vec![ok("ListObjectsV2").with_field("keys", "[]")];
-        let b = vec![ok("ListObjectsV2")];
-        assert_eq!(fields(&compare(&a, &b)), ["body.keys"]);
-    }
-
-    #[test]
-    fn error_messages_are_never_compared() {
-        let mut a = Observation::new("GetObject", "negative", 404, Some("NoSuchKey".into()), []);
-        let mut b = a.clone();
-        a.message = Some("The specified key does not exist.".into());
-        b.message = Some("Object not found".into());
-        assert_eq!(compare(&[a], &[b]), Vec::new());
-    }
-
-    #[test]
-    fn a_case_observed_on_one_side_only_is_reported() {
-        let a = vec![ok("HeadBucket"), ok("ListBuckets")];
-        let b = vec![ok("HeadBucket")];
-        let differences = compare(&a, &b);
-        assert_eq!(fields(&differences), ["observation"]);
-        assert_eq!(differences[0].op, "ListBuckets");
-        assert_eq!(compared_ops(&a, &b).into_iter().collect::<Vec<_>>(), ["HeadBucket"]);
-    }
-
-    #[test]
-    fn a_case_recorded_twice_is_reported() {
-        let a = vec![ok("HeadBucket"), ok("HeadBucket")];
-        let b = vec![ok("HeadBucket")];
-        assert_eq!(fields(&compare(&a, &b)), ["duplicate_case"]);
-    }
-
-    #[test]
-    fn compared_ops_counts_operations_not_cases() {
-        let negative = Observation::new("HeadBucket", "negative", 404, Some("NotFound".into()), []);
-        let a = vec![ok("HeadBucket"), negative, ok("ListBuckets")];
-        let b = a.clone();
-        assert_eq!(compared_ops(&a, &b).len(), 2);
-    }
-
-    #[test]
-    fn the_report_lists_each_difference_as_op_field_a_b() {
-        let directory = std::env::temp_dir().join(format!("dual-stack-report-{}", uuid::Uuid::new_v4()));
-        let report = Report {
-            schema: 1,
-            run: "unit".into(),
-            stack_a: "legacy".into(),
-            stack_b: "legacy".into(),
-            region_a: None,
-            region_b: Some("eu-west-1".into()),
-            compared_ops: 1,
-            compared_cases: 1,
-            differences: compare(
-                &[ok("GetBucketLocation").with_field("location_constraint", "us-east-1")],
-                &[ok("GetBucketLocation").with_field("location_constraint", "eu-west-1")],
-            ),
-        };
-        let path = report.write_to(&directory).expect("write the report");
-        let json: serde_json::Value = serde_json::from_slice(&std::fs::read(&path).expect("read the report")).expect("parse");
-        let _ = std::fs::remove_dir_all(&directory);
-        assert_eq!(path.file_name().and_then(|name| name.to_str()), Some("unit.json"));
-        assert_eq!(json["compared_ops"], 1);
-        assert_eq!(
-            json["differences"],
-            serde_json::json!([{
-                "op": "GetBucketLocation",
-                "case": "normal",
-                "field": "body.location_constraint",
-                "a": "us-east-1",
-                "b": "eu-west-1",
-            }])
-        );
-    }
-}
@@ -1,197 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Dual-stack differential: one `rustfs` binary, two processes, one SDK script (rustfs/backlog#2740).
-//!
-//! Stack A always runs `RUSTFS_S3_STACK=legacy`. Stack B runs the value of `RUSTFS_E2E_STACK_B`
-//! (default `legacy` until the gateway stack is assembled), and `RUSTFS_E2E_STACK_B_REGION`, when
-//! set, starts B in another region so that a difference can be injected by hand. Both servers
-//! answer the script in `scenarios`, which covers every operation of `impl S3 for FS`; `compare`
-//! reports each `(op, field, a, b)` on which the answers differ, and every run writes its report
-//! to `<target>/dual-stack/<run>.json`. Two processes, not two embedded servers: bucket metadata
-//! is process-global, so two stacks in one process would read each other's buckets.
-
-mod compare;
-mod scenarios;
-
-use crate::common::{RustFSTestEnvironment, init_logging, workspace_root};
-use compare::{Difference, Observation, Report, compare, compared_cases, compared_ops, render};
-use std::path::PathBuf;
-use std::time::{SystemTime, UNIX_EPOCH};
-
-type BoxError = Box<dyn std::error::Error + Send + Sync>;
-
-const STACK_A: &str = "legacy";
-const ENV_STACK_B: &str = "RUSTFS_E2E_STACK_B";
-const ENV_STACK_B_REGION: &str = "RUSTFS_E2E_STACK_B_REGION";
-/// Becomes `gateway` once rustfs/backlog#2734 T2.5 assembles the gateway stack.
-const DEFAULT_STACK_B: &str = "legacy";
-/// The region the negative control gives stack A explicitly, so an inherited `RUSTFS_REGION`
-/// cannot erase the injected difference.
-const CONTROL_REGION_A: &str = "us-east-1";
-const CONTROL_REGION_B: &str = "eu-west-1";
-/// `normal` cases the legacy stack must answer with a 2xx. The comparison alone cannot tell a
-/// script that exercises real behavior from one in which both stacks fail the same way. Five of
-/// the 73 fail on legacy by design of this script: PutBucketReplication (no registered remote
-/// target), GetBucketReplication (so no configuration), PutBucketNotificationConfiguration (no
-/// notification target configured), RestoreObject (the object is not archived) and
-/// GetObjectTorrent (legacy answers `NoSuchKey`).
-const MIN_NORMAL_SUCCESSES: usize = 68;
-
-struct PairConfig {
-    stack_b: String,
-    region_a: Option<String>,
-    region_b: Option<String>,
-}
-
-impl PairConfig {
-    fn from_env() -> Self {
-        Self {
-            stack_b: std::env::var(ENV_STACK_B).unwrap_or_else(|_| DEFAULT_STACK_B.to_owned()),
-            region_a: None,
-            region_b: std::env::var(ENV_STACK_B_REGION).ok(),
-        }
-    }
-}
-
-/// Starts one server on `stack`, overriding its region only when asked.
-async fn start_stack(stack: &str, region: Option<&str>) -> Result<RustFSTestEnvironment, BoxError> {
-    let mut env = RustFSTestEnvironment::new().await?;
-    let mut vars = vec![("RUSTFS_S3_STACK", stack)];
-    if let Some(region) = region {
-        vars.push(("RUSTFS_REGION", region));
-    }
-    env.start_rustfs_server_without_cleanup_with_env(&vars).await?;
-    Ok(env)
-}
-
-/// Starts stack A and stack B side by side, each with its own port and data directory.
-async fn spawn_pair(config: &PairConfig) -> Result<(RustFSTestEnvironment, RustFSTestEnvironment), BoxError> {
-    tokio::try_join!(
-        start_stack(STACK_A, config.region_a.as_deref()),
-        start_stack(&config.stack_b, config.region_b.as_deref())
-    )
-}
-
-fn report_directory() -> PathBuf {
-    let target = std::env::var_os("CARGO_TARGET_DIR").map_or_else(|| PathBuf::from("target"), PathBuf::from);
-    let target = if target.is_absolute() {
-        target
-    } else {
-        workspace_root().join(target)
-    };
-    target.join("dual-stack")
-}
-
-struct Run {
-    report: Report,
-    path: PathBuf,
-    a: Vec<Observation>,
-}
-
-/// Runs the script against both stacks concurrently and writes the report.
-async fn run_differential(label: &str, config: &PairConfig) -> Run {
-    let (server_a, server_b) = spawn_pair(config).await.expect("start stack A and stack B");
-    let (a, b) = tokio::join!(Box::pin(scenarios::run(&server_a)), Box::pin(scenarios::run(&server_b)));
-    let millis = SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .map_or(0, |elapsed| elapsed.as_millis());
-    let report = Report {
-        schema: 1,
-        run: format!("{label}-{millis}-{}", std::process::id()),
-        stack_a: STACK_A.to_owned(),
-        stack_b: config.stack_b.clone(),
-        region_a: config.region_a.clone(),
-        region_b: config.region_b.clone(),
-        compared_ops: compared_ops(&a, &b).len(),
-        compared_cases: compared_cases(&a, &b),
-        differences: compare(&a, &b),
-    };
-    let path = report.write_to(&report_directory()).expect("write the dual-stack report");
-    Run { report, path, a }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[tokio::test]
-    async fn both_stacks_answer_every_operation_identically() {
-        init_logging();
-        let config = PairConfig::from_env();
-        let run = run_differential("self-comparison", &config).await;
-        let report = &run.report;
-
-        let mut observed: Vec<_> = run.a.iter().map(|observation| observation.op).collect();
-        observed.sort_unstable();
-        observed.dedup();
-        assert_eq!(
-            observed,
-            scenarios::OPERATIONS,
-            "the script must drive exactly the operations of impl S3 for FS"
-        );
-        assert_eq!(report.compared_ops, scenarios::OPERATIONS.len(), "report: {}", run.path.display());
-        assert!(
-            report.differences.is_empty(),
-            "stack {} and stack {} differ on {} of {} cases (report: {}):\n{}",
-            report.stack_a,
-            report.stack_b,
-            report.differences.len(),
-            report.compared_cases,
-            run.path.display(),
-            render(&report.differences)
-        );
-
-        let not_4xx: Vec<_> = run
-            .a
-            .iter()
-            .filter(|observation| observation.case == "negative" && !(400..500).contains(&observation.status))
-            .collect();
-        assert!(not_4xx.is_empty(), "negative cases must fail with a 4xx on {STACK_A}: {not_4xx:#?}");
-        let successes = run
-            .a
-            .iter()
-            .filter(|observation| observation.case == "normal" && observation.is_success())
-            .count();
-        assert!(
-            successes >= MIN_NORMAL_SUCCESSES,
-            "only {successes} normal cases succeeded on {STACK_A}: {:#?}",
-            run.a
-        );
-    }
-
-    #[tokio::test]
-    async fn differential_detects_an_injected_difference() {
-        init_logging();
-        let config = PairConfig {
-            stack_b: STACK_A.to_owned(),
-            region_a: Some(CONTROL_REGION_A.to_owned()),
-            region_b: Some(CONTROL_REGION_B.to_owned()),
-        };
-        let run = run_differential("injected-region", &config).await;
-        let expected = Difference {
-            op: "GetBucketLocation".to_owned(),
-            case: "normal".to_owned(),
-            field: "body.location_constraint".to_owned(),
-            a: CONTROL_REGION_A.to_owned(),
-            b: CONTROL_REGION_B.to_owned(),
-        };
-        assert!(
-            run.report.differences.contains(&expected),
-            "B started in {CONTROL_REGION_B} but GetBucketLocation was not reported (report: {}):\n{}",
-            run.path.display(),
-            render(&run.report.differences)
-        );
-    }
-}
@@ -31,12 +31,6 @@ pub mod chaos;
 #[cfg(test)]
 pub mod fake_s3_target;

-// Dual-stack differential (rustfs/backlog#2740): one binary, two processes on
-// different RUSTFS_S3_STACK values, the same SDK script for every S3 operation,
-// compared on status, error code, allowlisted headers and typed body fields.
-#[cfg(test)]
-mod dual_stack_differential;
-
 // On-demand migration (backlog#2147): shared two-server environment, admin
 // wrappers, and the harness self-test (backlog#2151). Behavior scenarios are
 // added by later ODM tasks.
@@ -371,6 +371,24 @@ pub mod config {
            with_config_object_read_lock, with_config_object_write_lock, with_server_config_read_lock,
            with_server_config_write_lock,
        };
+
+        /// Read config bytes and metadata without treating an empty payload as absent.
+        pub async fn read_config_preserve_empty_with_metadata_opts(
+            api: std::sync::Arc<crate::store::ECStore>,
+            file: &str,
+            opts: &crate::object_api::ObjectOptions,
+        ) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
+            crate::config::com::read_config_preserve_empty_with_metadata_opts(api, file, opts).await
+        }
+
+        pub async fn read_config_limited_preserve_empty_with_metadata_opts(
+            api: std::sync::Arc<crate::store::ECStore>,
+            file: &str,
+            opts: &crate::object_api::ObjectOptions,
+            max_bytes: usize,
+        ) -> crate::error::Result<(Vec<u8>, crate::object_api::ObjectInfo)> {
+            crate::config::com::read_config_limited_preserve_empty_with_metadata_opts(api, file, opts, max_bytes).await
+        }
    }

    pub mod storageclass {
@@ -2939,6 +2939,20 @@ fn resolve_tier_free_version_recovery_enabled(value: Result<String, env::VarErro
 }

 fn spawn_tier_free_version_recovery_once(api: Arc<ECStore>, started: &OnceLock<()>) -> Option<JoinHandle<()>> {
+    spawn_tier_free_version_recovery_once_with_options(
+        api,
+        started,
+        TIER_FREE_VERSION_RECOVERY_LEADER_LOCK.to_string(),
+        tier_free_version_recovery_leader_retry_delay,
+    )
+}
+
+fn spawn_tier_free_version_recovery_once_with_options(
+    api: Arc<ECStore>,
+    started: &OnceLock<()>,
+    leader_lock_name: String,
+    retry_delay: fn() -> StdDuration,
+) -> Option<JoinHandle<()>> {
    if !tier_free_version_recovery_enabled() {
        warn!(
            event = EVENT_LIFECYCLE_WORKER_STATE,
@@ -2956,15 +2970,15 @@ fn spawn_tier_free_version_recovery_once(api: Arc<ECStore>, started: &OnceLock<(
    Some(tokio::spawn(async move {
        let cancel_token = api.ctx.background_cancel_token().unwrap_or_default();
        let acquire_api = Arc::clone(&api);
+        let acquire_lock_name = Arc::<str>::from(leader_lock_name);
        run_tier_free_version_recovery_coordinator(
            cancel_token,
-            tier_free_version_recovery_leader_retry_delay,
+            retry_delay,
            move || {
                let api = Arc::clone(&acquire_api);
+                let lock_name = Arc::clone(&acquire_lock_name);
                async move {
-                    let lock = api
-                        .new_ns_lock(RUSTFS_META_BUCKET, TIER_FREE_VERSION_RECOVERY_LEADER_LOCK)
-                        .await?;
+                    let lock = api.new_ns_lock(RUSTFS_META_BUCKET, &lock_name).await?;
                    match lock.get_write_lock_quiet(TIER_FREE_VERSION_RECOVERY_LOCK_TIMEOUT).await {
                        Ok(guard) => Ok(Some(guard)),
                        Err(LockError::Timeout { .. }) => Ok(None),
@@ -14193,6 +14207,10 @@ mod tests {
        .await;
    }

+    fn fast_tier_free_version_recovery_leader_retry_delay() -> StdDuration {
+        StdDuration::from_millis(10)
+    }
+
    #[tokio::test]
    #[serial]
    async fn tier_free_version_recovery_production_entrypoint_enqueues_seeded_item() {
@@ -14208,8 +14226,13 @@ mod tests {
        seed_recoverable_free_version(&disk_paths, &bucket, object, None, None).await;

        let started = OnceLock::new();
-        let recovery = super::spawn_tier_free_version_recovery_once(Arc::clone(&ecstore), &started)
-            .expect("production recovery entrypoint should start once");
+        let recovery = super::spawn_tier_free_version_recovery_once_with_options(
+            Arc::clone(&ecstore),
+            &started,
+            format!("tier-free-version-recovery/test/{}", Uuid::new_v4()),
+            fast_tier_free_version_recovery_leader_retry_delay,
+        )
+        .expect("production recovery entrypoint should start once");
        let task = tokio::time::timeout(StdDuration::from_secs(30), async { recovery_rx.lock().await.recv().await })
            .await
            .expect("production recovery entrypoint should enqueue the seeded item")
@@ -1292,14 +1292,34 @@ pub(crate) async fn get_object_lock_config_state_in(
 pub(crate) async fn get_object_lock_config_and_incarnation_from_disk_in(
    ctx: &crate::runtime::instance::InstanceContext,
    bucket: &str,
+) -> Result<(ObjectLockConfigState, Uuid, OffsetDateTime)> {
+    get_object_lock_config_and_incarnation_from_disk_with_guard_in(ctx, bucket, None).await
+}
+
+/// Bucket deletion already owns the namespace/lifecycle fences. Reuse its
+/// live guard rather than recursively acquiring the bucket namespace lock.
+/// The caller must also hold the metadata transaction read lock.
+pub(crate) async fn get_object_lock_config_and_incarnation_from_disk_with_guard_in(
+    ctx: &crate::runtime::instance::InstanceContext,
+    bucket: &str,
+    guard: Option<&rustfs_lock::NamespaceLockGuard>,
 ) -> Result<(ObjectLockConfigState, Uuid, OffsetDateTime)> {
    let bucket_meta_sys_lock = bucket_metadata_sys_of(ctx)?;
    let bucket_meta_sys = bucket_meta_sys_lock.read().await.clone();

-    match bucket_meta_sys
-        .read_authoritative_metadata_from_disk_under_transaction_lock(bucket)
-        .await?
-    {
+    let authority = match guard {
+        Some(guard) => {
+            bucket_meta_sys
+                .read_authoritative_metadata_from_disk_under_guard(bucket, guard)
+                .await?
+        }
+        None => {
+            bucket_meta_sys
+                .read_authoritative_metadata_from_disk_under_transaction_lock(bucket)
+                .await?
+        }
+    };
+    match authority {
        BucketMetadataAuthority::Authoritative(metadata)
            if metadata.bucket_incarnation_sidecar && !metadata.bucket_incarnation_id.is_nil() =>
        {
@@ -495,10 +495,22 @@ pub(crate) async fn read_config_preserve_empty<S>(api: Arc<S>, file: &str) -> Re
 where
    S: EcstoreObjectIO,
 {
-    let (data, _obj) = read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, None).await?;
+    let (data, _obj) = read_config_preserve_empty_with_metadata_opts(api, file, &ObjectOptions::default()).await?;
    Ok(data)
 }

+/// Preserve an empty config payload, its metadata, and caller-supplied read options.
+pub(crate) async fn read_config_preserve_empty_with_metadata_opts<S>(
+    api: Arc<S>,
+    file: &str,
+    opts: &ObjectOptions,
+) -> Result<(Vec<u8>, ObjectInfo)>
+where
+    S: EcstoreObjectIO,
+{
+    read_config_with_metadata_inner(api, file, opts, true, None).await
+}
+
 pub async fn read_config_no_lock<S>(api: Arc<S>, file: &str) -> Result<Vec<u8>>
 where
    S: ObjectIO<
@@ -527,15 +539,13 @@ pub(crate) async fn read_config_no_lock_preserve_empty_with_metadata<S>(api: Arc
 where
    S: EcstoreObjectIO,
 {
-    read_config_with_metadata_inner(
+    read_config_preserve_empty_with_metadata_opts(
        api,
        file,
        &ObjectOptions {
            no_lock: true,
            ..Default::default()
        },
-        true,
-        None,
    )
    .await
 }
@@ -2815,9 +2825,10 @@ mod tests {
        config_task_join_error, configs_semantically_equal, decode_server_config_blob, encode_server_config_blob,
        heal_config_descriptor, is_standard_object_server_config, lookup_configs, map_system_metadata_write_error,
        new_and_save_server_config, read_config, read_config_no_lock_preserve_empty_with_metadata, read_config_preserve_empty,
-        read_config_with_metadata, read_config_without_migrate, read_server_config_snapshot, save_config_with_opts_inner,
-        save_server_config, save_server_config_snapshot, save_server_config_snapshot_with_generation,
-        server_config_transaction_lock_path, should_warn_ignored_scalar_section, storage_class_kvs_mut,
+        read_config_preserve_empty_with_metadata_opts, read_config_with_metadata, read_config_without_migrate,
+        read_server_config_snapshot, save_config_with_opts_inner, save_server_config, save_server_config_snapshot,
+        save_server_config_snapshot_with_generation, server_config_transaction_lock_path, should_warn_ignored_scalar_section,
+        storage_class_kvs_mut,
    };
    use crate::config::{audit, heal, notify, oidc, scanner};
    use crate::disk::{RUSTFS_META_BUCKET, endpoint::Endpoint};
@@ -5065,6 +5076,7 @@ mod tests {
        heal_replacement: Option<Vec<u8>>,
        heal_calls: AtomicUsize,
        write_calls: AtomicUsize,
+        last_get_no_lock: AtomicBool,
        last_put_no_lock: AtomicBool,
        last_put_preconditions: Mutex<Option<HTTPPreconditions>>,
        revision: AtomicUsize,
@@ -5080,6 +5092,7 @@ mod tests {
                heal_replacement,
                heal_calls: AtomicUsize::new(0),
                write_calls: AtomicUsize::new(0),
+                last_get_no_lock: AtomicBool::new(false),
                last_put_no_lock: AtomicBool::new(false),
                last_put_preconditions: Mutex::new(None),
                revision: AtomicUsize::new(1),
@@ -5135,8 +5148,9 @@ mod tests {
            _object: &str,
            _range: Option<HTTPRangeSpec>,
            _h: HeaderMap,
-            _opts: &ObjectOptions,
+            opts: &ObjectOptions,
        ) -> Result<GetObjectReader> {
+            self.last_get_no_lock.store(opts.no_lock, Ordering::SeqCst);
            let data = match &*self.state.lock().expect("state lock poisoned") {
                RecoveryReadState::Missing => return Err(Error::ConfigNotFound),
                RecoveryReadState::Blob(data) => data.clone(),
@@ -5431,6 +5445,60 @@ mod tests {
        assert!(data.is_empty());
    }

+    #[tokio::test]
+    async fn read_config_preserve_empty_metadata_keeps_empty_and_options() {
+        let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(Vec::new()), None));
+        for no_lock in [false, true] {
+            let opts = ObjectOptions {
+                no_lock,
+                ..Default::default()
+            };
+            let (data, info) = read_config_preserve_empty_with_metadata_opts(store.clone(), "config/empty.json", &opts)
+                .await
+                .expect("an existing empty config must retain its payload and metadata");
+            assert!(data.is_empty());
+            assert_eq!(info.size, 0);
+            assert_eq!(info.actual_size, 0);
+            assert_eq!(info.etag.as_deref(), Some("config-1"));
+            assert_eq!(info.data_dir, Some(uuid::Uuid::from_u128(1)));
+            assert_eq!(store.last_get_no_lock.load(Ordering::SeqCst), no_lock);
+        }
+        let err = read_config(store, "config/empty.json")
+            .await
+            .expect_err("the existing reader must still treat empty configs as missing");
+        assert!(matches!(err, Error::ConfigNotFound));
+    }
+
+    #[tokio::test]
+    async fn read_config_preserve_empty_metadata_keeps_payload_and_metadata() {
+        let payload = b"original config bytes".to_vec();
+        let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(payload.clone()), None));
+        let (data, info) =
+            read_config_preserve_empty_with_metadata_opts(store, "config/nonempty.json", &ObjectOptions::default())
+                .await
+                .expect("a nonempty config must retain its payload and metadata");
+        assert_eq!(data, payload);
+        let size = i64::try_from(payload.len()).expect("test payload size fits i64");
+        assert_eq!(info.size, size);
+        assert_eq!(info.actual_size, size);
+        assert_eq!(info.etag.as_deref(), Some("config-1"));
+        assert_eq!(info.data_dir, Some(uuid::Uuid::from_u128(1)));
+    }
+
+    #[tokio::test]
+    async fn read_config_preserve_empty_metadata_preserves_read_errors() {
+        let missing = Arc::new(RecoveryMockStore::new(RecoveryReadState::Missing, None));
+        let err = read_config_preserve_empty_with_metadata_opts(missing, "config/missing.json", &ObjectOptions::default())
+            .await
+            .expect_err("physical absence must remain ConfigNotFound");
+        assert!(matches!(err, Error::ConfigNotFound));
+        let quorum = Arc::new(RecoveryMockStore::new(RecoveryReadState::QuorumError, None));
+        let err = read_config_preserve_empty_with_metadata_opts(quorum, "config/quorum.json", &ObjectOptions::default())
+            .await
+            .expect_err("a quorum failure must retain its typed read error");
+        assert!(matches!(err, Error::ErasureReadQuorum));
+    }
+
    #[async_trait::async_trait]
    impl StorageAdminApi for RecoveryMockStore {
        type BackendInfo = rustfs_madmin::BackendInfo;
@@ -2911,7 +2911,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);
@@ -12477,6 +12477,227 @@ mod test {
        }
    }

+    // SIGKILL tests process death on a healthy filesystem, not power loss.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn rename_data_process_kill_preserves_publication_boundary() {
+        use std::io::Write as _;
+        use std::os::unix::process::ExitStatusExt;
+        use std::process::{Child, Command, ExitStatus, Stdio};
+        use std::time::{Duration, Instant};
+
+        const TEST: &str = "disk::local::test::rename_data_process_kill_preserves_publication_boundary";
+        const ROLE: &str = "RUSTFS_TEST_PUBLICATION_KILL_ROLE";
+        const ROOT: &str = "RUSTFS_TEST_PUBLICATION_KILL_ROOT";
+        const NONCE: &str = "RUSTFS_TEST_PUBLICATION_KILL_NONCE";
+        const BUCKET: &str = "bucket";
+        const OBJECT: &str = "process-kill-object";
+        const STAGED: &str = "process-kill-staged";
+        const VERSION: Uuid = Uuid::from_u128(0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa);
+        const OLD: Uuid = Uuid::from_u128(0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb);
+        const NEW: Uuid = Uuid::from_u128(0xcccccccccccccccccccccccccccccccc);
+
+        struct ChildGuard(Child);
+        impl Drop for ChildGuard {
+            fn drop(&mut self) {
+                let _ = self.0.kill();
+                let _ = self.0.wait();
+            }
+        }
+
+        async fn wait_exit(child: &mut ChildGuard) -> ExitStatus {
+            let deadline = Instant::now() + Duration::from_secs(60);
+            loop {
+                if child.0.try_wait().expect("poll owned child").is_some() {
+                    return child.0.wait().expect("reap owned child");
+                }
+                assert!(Instant::now() < deadline, "owned child exceeded exit deadline");
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+        }
+
+        fn spawn(root: &std::path::Path, nonce: &str, role: &str) -> ChildGuard {
+            let stdout = std::fs::File::create(root.join(format!("{role}.stdout"))).expect("child stdout file");
+            let stderr = std::fs::File::create(root.join(format!("{role}.stderr"))).expect("child stderr file");
+            ChildGuard(
+                Command::new(std::env::current_exe().expect("test executable"))
+                    .args(["--exact", TEST, "--nocapture"])
+                    .env(ROLE, role)
+                    .env(ROOT, root)
+                    .env(NONCE, nonce)
+                    .stdout(Stdio::from(stdout))
+                    .stderr(Stdio::from(stderr))
+                    .spawn()
+                    .expect("spawn owned publication child"),
+            )
+        }
+
+        fn child_logs(root: &std::path::Path, role: &str) -> String {
+            let stdout = std::fs::read_to_string(root.join(format!("{role}.stdout"))).expect("child stdout");
+            let stderr = std::fs::read_to_string(root.join(format!("{role}.stderr"))).expect("child stderr");
+            format!("{stdout} {stderr}")
+        }
+
+        fn file_info(data_dir: Uuid) -> FileInfo {
+            let mut fi = test_file_info(OBJECT, VERSION, Some(data_dir), None);
+            fi.size = 8;
+            fi.parts[0].size = 8;
+            fi.parts[0].actual_size = 8;
+            fi
+        }
+
+        if let Some(role) = std::env::var_os(ROLE) {
+            let role = role.to_str().expect("child role is UTF-8");
+            assert!(matches!(role, "writer-before" | "writer-after" | "verify-before" | "verify-after"));
+            let root = std::path::PathBuf::from(std::env::var_os(ROOT).expect("private child root"));
+            let nonce = std::env::var(NONCE).expect("private child nonce");
+            Uuid::parse_str(&nonce).expect("nonce is a UUID");
+            assert_eq!(root.file_name().and_then(|name| name.to_str()), Some(nonce.as_str()));
+            assert_eq!(std::fs::read_to_string(root.join("control-nonce")).expect("bound root nonce"), nonce);
+            let endpoint = Endpoint::try_from(root.to_str().expect("private root UTF-8")).expect("private endpoint");
+            let _strict = durability_mode_override::set(DurabilityMode::Strict);
+            let disk = LocalDisk::new(&endpoint, false).await.expect("child opens real disk");
+            let object_dir = root.join(BUCKET).join(OBJECT);
+            if role.starts_with("verify-") {
+                let fi = disk
+                    .read_version("", BUCKET, OBJECT, &VERSION.to_string(), &ReadOptions::default())
+                    .await
+                    .expect("fresh verifier reads actual version");
+                let (expected_dir, expected_bytes) = if role == "verify-before" {
+                    assert_eq!(
+                        fs::read(object_dir.join(STORAGE_FORMAT_FILE))
+                            .await
+                            .expect("read old metadata"),
+                        fs::read(root.join("old-meta")).await.expect("read old fixture"),
+                        "prepublication metadata must remain byte-for-byte old"
+                    );
+                    (OLD, b"old-data")
+                } else {
+                    (NEW, b"new-data")
+                };
+                assert_eq!(fi.data_dir, Some(expected_dir), "{role}: actual reopened version data_dir");
+                let part = object_dir
+                    .join(fi.data_dir.expect("referenced data dir").to_string())
+                    .join("part.1");
+                assert_eq!(fs::read(part).await.expect("read referenced shard"), expected_bytes);
+                return;
+            }
+
+            let staged = root.join(RUSTFS_META_TMP_BUCKET).join(STAGED).join(NEW.to_string());
+            fs::create_dir_all(&staged).await.expect("staged shard directory");
+            fs::write(staged.join("part.1"), b"new-data")
+                .await
+                .expect("stage new shard bytes");
+            let ready = root.join(format!("{role}.ready"));
+            let pause = move || {
+                let pending = ready.with_extension("pending");
+                std::fs::write(&pending, &nonce).expect("write actual writer cut marker");
+                std::fs::rename(pending, &ready).expect("announce complete writer cut marker");
+                let (_sender, receiver) = std::sync::mpsc::channel::<()>();
+                receiver.recv().expect("writer remains paused until SIGKILL");
+            };
+            let _hook = if role == "writer-before" {
+                // Use the executor's exact path spelling, including canonical
+                // macOS roots and Linux mount-pinned I/O roots.
+                let destination = disk
+                    .io_get_object_path(BUCKET, &format!("{OBJECT}/{STORAGE_FORMAT_FILE}"))
+                    .expect("actual metadata publication path");
+                assert_eq!(
+                    std::fs::canonicalize(&destination).expect("canonical publication destination"),
+                    disk.root.join(BUCKET).join(OBJECT).join(STORAGE_FORMAT_FILE),
+                    "hook destination must identify this disk's seeded metadata"
+                );
+                Some(os::prepared_publication_test_hooks::install_at(
+                    os::prepared_publication_test_hooks::Stage::PreparedRename,
+                    &destination,
+                    pause,
+                ))
+            } else {
+                disk.rename_data(RUSTFS_META_TMP_BUCKET, STAGED, file_info(NEW), BUCKET, OBJECT)
+                    .await
+                    .expect("Strict real rename_data completes before cut");
+                pause();
+                return;
+            };
+            disk.rename_data(RUSTFS_META_TMP_BUCKET, STAGED, file_info(NEW), BUCKET, OBJECT)
+                .await
+                .expect("before writer must remain in physical hook");
+            panic!("prepublication hook unexpectedly returned");
+        }
+
+        let temp = tempfile::tempdir().expect("parent owns private disk roots");
+        for phase in ["before", "after"] {
+            let nonce = Uuid::new_v4().to_string();
+            let root = temp.path().join(&nonce);
+            fs::create_dir(&root).await.expect("private scenario root");
+            fs::write(root.join("control-nonce"), &nonce)
+                .await
+                .expect("bind private root");
+            let endpoint = Endpoint::try_from(root.to_str().expect("private root UTF-8")).expect("private endpoint");
+            let disk = LocalDisk::new(&endpoint, false).await.expect("seed real disk");
+            ensure_test_volume(&disk, BUCKET).await;
+            ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
+            let object_dir = root.join(BUCKET).join(OBJECT);
+            fs::create_dir_all(object_dir.join(OLD.to_string()))
+                .await
+                .expect("old shard directory");
+            fs::write(object_dir.join(OLD.to_string()).join("part.1"), b"old-data")
+                .await
+                .expect("seed old shard bytes");
+            let old_meta = test_meta(file_info(OLD));
+            fs::write(object_dir.join(STORAGE_FORMAT_FILE), &old_meta)
+                .await
+                .expect("seed old metadata");
+            fs::write(root.join("old-meta"), &old_meta)
+                .await
+                .expect("retain exact old fixture");
+            drop(disk);
+
+            let writer_role = format!("writer-{phase}");
+            let mut writer = spawn(&root, &nonce, &writer_role);
+            let deadline = Instant::now() + Duration::from_secs(60);
+            let ready = root.join(format!("{writer_role}.ready"));
+            loop {
+                if let Some(status) = writer.0.try_wait().expect("poll writer") {
+                    panic!(
+                        "writer exited before cut: {writer_role} status={status}: {}",
+                        child_logs(&root, &writer_role)
+                    );
+                }
+                match fs::read_to_string(&ready).await {
+                    Ok(marker) => {
+                        assert_eq!(marker, nonce, "cut marker belongs to this private writer");
+                        break;
+                    }
+                    Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+                    Err(error) => panic!("read cut marker: {error}"),
+                }
+                if Instant::now() >= deadline {
+                    panic!("writer cut deadline: {writer_role}: {}", child_logs(&root, &writer_role));
+                }
+                tokio::time::sleep(Duration::from_millis(20)).await;
+            }
+            let pid = writer.0.id();
+            writer.0.kill().expect("SIGKILL owned writer at verified cut");
+            let status = wait_exit(&mut writer).await;
+            assert_eq!(status.signal(), Some(9), "owned writer must actually die by SIGKILL");
+            writeln!(std::io::stdout(), "{writer_role} pid={pid} signal={:?}", status.signal())
+                .expect("record actual writer wait status");
+
+            let verifier_role = format!("verify-{phase}");
+            let mut verifier = spawn(&root, &nonce, &verifier_role);
+            let status = wait_exit(&mut verifier).await;
+            let logs = child_logs(&root, &verifier_role);
+            assert!(status.success(), "{verifier_role} status={status}: {logs}");
+            let stdout = std::fs::read_to_string(root.join(format!("{verifier_role}.stdout"))).expect("verifier stdout");
+            assert!(
+                stdout.contains(TEST) && stdout.contains("1 passed"),
+                "verifier must execute exact test: {logs}"
+            );
+            writeln!(std::io::stdout(), "{verifier_role} status={status}").expect("record actual verifier exit");
+        }
+    }
+
    /// Crash-consistency harness for the rename_data commit sequence
    /// (rustfs/backlog#935 HP-14, test plan rustfs/backlog#896; hard rule from
    /// rustfs/backlog#878: "After partial commit, the object can only be an old or new version; it cannot be mixed").
@@ -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>;
@@ -442,12 +442,19 @@ impl WritePlan {
        }

        if let Some(encryption) = self.encryption {
+            let etag_nonce = match encryption.mode {
+                WriteEncryptionMode::Singlepart { base_nonce } | WriteEncryptionMode::MultipartLegacy { base_nonce, .. } => {
+                    base_nonce
+                }
+                WriteEncryptionMode::SinglepartObjectKey | WriteEncryptionMode::MultipartObjectKey { .. } => [0u8; 12],
+            };
+            let protected_reader = rustfs_rio::EtagReader::new_encrypted(reader, encryption.key_bytes, etag_nonce)?;
            reader = match encryption.mode {
                WriteEncryptionMode::SinglepartObjectKey => HashReader::from_reader(
                    #[cfg(feature = "rio-v2")]
-                    EncryptReader::new_with_object_key(reader, encryption.key_bytes),
+                    EncryptReader::new_with_object_key(protected_reader, encryption.key_bytes),
                    #[cfg(not(feature = "rio-v2"))]
-                    EncryptReader::new(reader, encryption.key_bytes, [0u8; 12]),
+                    EncryptReader::new(protected_reader, encryption.key_bytes, [0u8; 12]),
                    HashReader::SIZE_PRESERVE_LAYER,
                    actual_size,
                    None,
@@ -457,12 +464,12 @@ impl WritePlan {
                WriteEncryptionMode::Singlepart { base_nonce } => {
                    #[cfg(not(feature = "rio-v2"))]
                    let encrypt_reader = if encryption_frame_v2_enabled() {
-                        EncryptReader::new_v2(reader, encryption.key_bytes, base_nonce)
+                        EncryptReader::new_v2(protected_reader, encryption.key_bytes, base_nonce)
                    } else {
-                        EncryptReader::new(reader, encryption.key_bytes, base_nonce)
+                        EncryptReader::new(protected_reader, encryption.key_bytes, base_nonce)
                    };
                    #[cfg(feature = "rio-v2")]
-                    let encrypt_reader = EncryptReader::new(reader, encryption.key_bytes, base_nonce);
+                    let encrypt_reader = EncryptReader::new(protected_reader, encryption.key_bytes, base_nonce);
                    HashReader::from_reader(encrypt_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)?
                }
                WriteEncryptionMode::MultipartLegacy {
@@ -471,20 +478,25 @@ impl WritePlan {
                } => {
                    #[cfg(not(feature = "rio-v2"))]
                    let encrypt_reader = if encryption_frame_v2_enabled() {
-                        EncryptReader::new_multipart_v2(reader, encryption.key_bytes, base_nonce, multipart_part_number)
+                        EncryptReader::new_multipart_v2(protected_reader, encryption.key_bytes, base_nonce, multipart_part_number)
                    } else {
-                        EncryptReader::new_multipart(reader, encryption.key_bytes, base_nonce, multipart_part_number)
+                        EncryptReader::new_multipart(protected_reader, encryption.key_bytes, base_nonce, multipart_part_number)
                    };
                    #[cfg(feature = "rio-v2")]
                    let encrypt_reader =
-                        EncryptReader::new_multipart(reader, encryption.key_bytes, base_nonce, multipart_part_number);
+                        EncryptReader::new_multipart(protected_reader, encryption.key_bytes, base_nonce, multipart_part_number);
                    HashReader::from_reader(encrypt_reader, HashReader::SIZE_PRESERVE_LAYER, actual_size, None, None, false)?
                }
                WriteEncryptionMode::MultipartObjectKey { multipart_part_number } => HashReader::from_reader(
                    #[cfg(feature = "rio-v2")]
-                    EncryptReader::new_multipart_with_object_key(reader, encryption.key_bytes, multipart_part_number),
+                    EncryptReader::new_multipart_with_object_key(protected_reader, encryption.key_bytes, multipart_part_number),
                    #[cfg(not(feature = "rio-v2"))]
-                    EncryptReader::new_multipart(reader, encryption.key_bytes, [0u8; 12], multipart_part_number as usize),
+                    EncryptReader::new_multipart(
+                        protected_reader,
+                        encryption.key_bytes,
+                        [0u8; 12],
+                        multipart_part_number as usize,
+                    ),
                    HashReader::SIZE_PRESERVE_LAYER,
                    actual_size,
                    None,
@@ -497,6 +509,7 @@ impl WritePlan {
        // `ignore_value` deliberately avoids a second hasher over compressed or
        // encrypted bytes. The inner reader still validates the plaintext request
        // checksum while this outer reader exposes the request checksum context.
+        reader.encrypted_etag = encrypted;
        reader.add_non_trailing_checksum(checksum, true)?;
        reader.set_trailer(trailer);

@@ -513,6 +526,36 @@ mod tests {
    use std::io::Cursor;
    use tokio::io::AsyncReadExt;

+    #[tokio::test]
+    async fn write_plan_encryption_persists_protected_etags_for_every_mode() {
+        let data = b"Salary 2026: 85000 EUR";
+        let size = i64::try_from(data.len()).expect("test data size");
+        let mut control =
+            HashReader::from_stream(Cursor::new(data), size, size, None, None, false).expect("create control reader");
+        control.read_to_end(&mut Vec::new()).await.expect("read control plaintext");
+        let md5 = control.try_resolve_etag().expect("control MD5");
+        for encryption in [
+            WriteEncryption::singlepart([1; 32], [1; 12]),
+            WriteEncryption::singlepart_object_key([1; 32]),
+            WriteEncryption::multipart([1; 32], [1; 12], 1),
+            WriteEncryption::multipart_object_key([1; 32], 1),
+        ] {
+            let plaintext = HashReader::from_stream(Cursor::new(data), size, size, Some(md5.clone()), None, false)
+                .expect("create checked reader");
+            let mut reader = WritePlan::new()
+                .with_encryption(encryption)
+                .apply(plaintext, size)
+                .expect("apply encryption");
+            let mut ciphertext = Vec::new();
+            reader.read_to_end(&mut ciphertext).await.expect("stream ciphertext");
+            assert_ne!(ciphertext, data);
+            let tag = reader.try_resolve_etag().expect("resolve persisted encrypted ETag");
+            assert_ne!(tag, md5, "stored ETag must not reveal the plaintext MD5");
+            assert_eq!(tag.len(), 32, "multipart ETags require a 16-byte hexadecimal part tag");
+            assert_eq!(reader.try_resolve_etag().as_ref(), Some(&tag));
+        }
+    }
+
    async fn assert_non_trailing_checksum_survives(plan: WritePlan) {
        let plaintext = b"checksum-context-through-write-plan".repeat(256);
        let actual_size = plaintext.len() as i64;
@@ -29,7 +29,7 @@ use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions};
 use crate::storage_api_contracts::range::HTTPRangeSpec;
 use rustfs_filemeta::ObjectPartInfo;
 use rustfs_rio::HashReader;
-use rustfs_s3_types::S3ErrorCode;
+use s3s::S3ErrorCode;

 //#[derive(Clone)]
 pub struct PutObjReader {
@@ -231,45 +231,6 @@ mod tests {
        assert_eq!(result.2, 1024 * 1024);
    }

-    #[test]
-    fn test_new_getobjectreader_rejects_a_range_past_the_object_as_invalid_range() {
-        let oi = multipart_object_info();
-        let past_the_end = Some(HTTPRangeSpec {
-            is_suffix_length: false,
-            start: oi.size,
-            end: oi.size + 10,
-        });
-
-        let err = new_getobjectreader(&past_the_end, &oi, &ObjectOptions::default(), &HeaderMap::new())
-            .err()
-            .expect("a range starting at the object size must be rejected");
-
-        assert_eq!(err.code, S3ErrorCode::InvalidRange);
-        assert_eq!(err.code.as_str(), "InvalidRange");
-        assert_eq!(err.key, None);
-        assert_eq!(err.bucket_name, None);
-    }
-
-    #[test]
-    fn test_new_getobjectreader_rejects_an_unknown_size_without_a_range_as_invalid_range() {
-        let oi = ObjectInfo {
-            bucket: "bucket".to_string(),
-            name: "object".to_string(),
-            size: -1,
-            ..Default::default()
-        };
-
-        let err = new_getobjectreader(&None, &oi, &ObjectOptions::default(), &HeaderMap::new())
-            .err()
-            .expect("an object of unknown size cannot be read without a range");
-
-        assert_eq!(err.code, S3ErrorCode::InvalidRange);
-        assert_eq!(err.code.as_str(), "InvalidRange");
-        assert_eq!(err.message, "Invalid range");
-        assert_eq!(err.key.as_deref(), Some("object"));
-        assert_eq!(err.bucket_name.as_deref(), Some("bucket"));
-    }
-
    #[test]
    fn test_to_s3s_etag() {
        // Test unquoted ETag - should become strong etag
@@ -82,9 +82,8 @@ use crate::{
 use rustfs_filemeta::FileInfo;
 use rustfs_rio::HashReader;
 use rustfs_s3_client::{admin_handler_utils::AdminError, provider_versions::ProviderVersionCapabilities};
-use rustfs_s3_types::S3ErrorCode;
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join};
-use s3s::dto::BucketLifecycleConfiguration;
+use s3s::{S3ErrorCode, dto::BucketLifecycleConfiguration};

 use super::{
    tier_handlers::{ERR_TIER_BUCKET_NOT_FOUND, ERR_TIER_CONNECT_ERR, ERR_TIER_INVALID_CREDENTIALS, ERR_TIER_PERM_ERR},
@@ -3004,6 +3004,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
        // etag
        let etag = resolve_complete_etag(opts, &uploaded_parts);

+        // An upload can contain parts from an older writer or a peer. Until
+        // every part's format is attested, keep its ETag out of keyless lists.
+        rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG);
+
        fi.metadata.insert("etag".to_owned(), etag);

        let persist_encryption_original_size = should_persist_encryption_original_size(&fi.metadata);
@@ -3847,6 +3847,16 @@ impl SetDisks {
                .map(crate::io_support::rio::compression_index_storage_bytes);

            let mut etag = data.stream.try_resolve_etag().unwrap_or_default();
+            if data.stream.encrypted_etag {
+                if opts.preserve_etag.as_ref().is_none_or(|preserved| preserved == &etag) {
+                    insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG, "v1".to_owned());
+                } else {
+                    rustfs_utils::http::metadata_compat::remove_str(
+                        &mut user_defined,
+                        rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
+                    );
+                }
+            }
            if let Some(ref tag) = opts.preserve_etag {
                etag = tag.clone();
            }
@@ -10610,6 +10620,65 @@ mod object_encryption_resolver_wiring_tests {
    use std::sync::Mutex;
    use std::sync::atomic::{AtomicUsize, Ordering};

+    #[tokio::test]
+    async fn encrypted_put_attests_only_etags_produced_by_its_encryption_reader() {
+        use super::hermetic_set_disks_support::hermetic_set_disks_isolated;
+        use crate::io_support::rio::{WriteEncryption, WritePlan};
+        use crate::storage_api_contracts::object::ObjectIO as _;
+        use tokio::io::AsyncReadExt;
+
+        let (_temp_dirs, disks, set_disks) = hermetic_set_disks_isolated(4).await;
+        let bucket = "encrypted-etag-proof";
+        for disk in &disks {
+            disk.make_volume(bucket).await.expect("create bucket volume");
+        }
+        let plaintext = b"Salary 2026: 85000 EUR";
+        let size = plaintext.len() as i64;
+        let mut control = HashReader::from_stream(Cursor::new(plaintext), size, size, None, None, false).expect("control reader");
+        control.read_to_end(&mut Vec::new()).await.expect("read control plaintext");
+        let plaintext_md5 = control.try_resolve_etag().expect("control MD5");
+        for (object, preserved) in [("new", None), ("preserved", Some(plaintext_md5.clone()))] {
+            let reader =
+                HashReader::from_stream(Cursor::new(plaintext), size, size, None, None, false).expect("plaintext reader");
+            let reader = WritePlan::new()
+                .with_encryption(WriteEncryption::singlepart([1; 32], [2; 12]))
+                .apply(reader, size)
+                .expect("encrypted reader");
+            let mut opts = ObjectOptions {
+                preserve_etag: preserved.clone(),
+                user_defined: HashMap::from([
+                    ("x-amz-server-side-encryption-customer-algorithm".to_owned(), "AES256".to_owned()),
+                    (rustfs_utils::http::SSEC_ORIGINAL_SIZE_HEADER.to_owned(), size.to_string()),
+                ]),
+                ..Default::default()
+            };
+            // A copied producer marker cannot attest to a preserved plaintext tag.
+            insert_str(&mut opts.user_defined, rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG, "v1".to_owned());
+            set_disks
+                .put_object(bucket, object, &mut PutObjReader::new(reader), &opts)
+                .await
+                .expect("store encrypted bytes");
+            let stored = set_disks
+                .get_object_info(bucket, object, &ObjectOptions::default())
+                .await
+                .expect("read persisted metadata");
+            if let Some(preserved) = preserved {
+                assert_eq!(stored.etag.as_deref(), Some(preserved.as_str()));
+                assert!(!rustfs_utils::http::etag_can_be_listed(&stored.user_defined));
+            } else {
+                assert_ne!(stored.etag.as_ref(), Some(&plaintext_md5));
+                assert_eq!(
+                    rustfs_utils::http::get_consistent_str(
+                        &stored.user_defined,
+                        rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG
+                    ),
+                    Some("v1")
+                );
+                assert!(rustfs_utils::http::etag_can_be_listed(&stored.user_defined));
+            }
+        }
+    }
+
    #[derive(Clone, Default)]
    struct CapturedLogs(Arc<Mutex<Vec<u8>>>);

@@ -901,6 +901,7 @@ impl ECStore {
            }
        };
        let confirmed_missing = existing_bucket_info.is_none();
+        let mut pending_creation_metadata = None;
        if confirmed_missing
            && !opts.no_lock
            && !is_meta_bucketname(bucket)
@@ -921,13 +922,25 @@ impl ECStore {
                    ),
                    other => other,
                })?;
-            return Err(StorageError::InvalidArgument(
-                "CreateBucket".to_string(),
-                "bucket".to_string(),
-                format!("orphaned bucket generation {old_incarnation} requires administrator recovery before recreation"),
-            ));
+            let (metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
+            let metadata_store = metadata_sys::object_store_in(&self.ctx).await?;
+            let pending_intent = persisted
+                && metadata.bucket_incarnation_id == old_incarnation
+                && metadata.needs_bucket_creation_commit()
+                && !crate::bucket::retirement::is_retired(metadata_store, bucket, old_incarnation).await?;
+            if pending_intent {
+                pending_creation_metadata = Some(metadata);
+            } else {
+                return Err(StorageError::InvalidArgument(
+                    "CreateBucket".to_string(),
+                    "bucket".to_string(),
+                    format!("orphaned bucket generation {old_incarnation} requires administrator recovery before recreation"),
+                ));
+            }
        }
-        let existing_metadata = if opts.force_create && !confirmed_missing && !is_meta_bucketname(bucket) {
+        let existing_metadata = if let Some(metadata) = pending_creation_metadata {
+            Some(metadata)
+        } else if opts.force_create && !confirmed_missing && !is_meta_bucketname(bucket) {
            let (mut metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?;
            if !persisted {
                metadata = BucketMetadata::new(bucket);
@@ -1347,6 +1360,23 @@ impl ECStore {
        } else {
            None
        };
+        // Match metadata migration: lifecycle -> metadata transaction ->
+        // bucket namespace. Holding the transaction while waiting for the
+        // namespace prevents a migration writer and force delete from
+        // acquiring these two locks in opposite orders.
+        let object_lock_metadata_transaction = if opts.force && !is_meta_bucketname(bucket) {
+            Some(
+                await_bucket_namespace_operation(
+                    bucket_lifecycle_guard.as_ref(),
+                    bucket,
+                    "Object Lock metadata transaction acquisition",
+                    crate::bucket::metadata_sys::acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket),
+                )
+                .await?,
+            )
+        } else {
+            None
+        };
        let ns_guard = if !opts.no_lock {
            let ns_lock = self.new_ns_lock(bucket, bucket).await?;
            Some(
@@ -1402,6 +1432,29 @@ impl ECStore {
            }
        }

+        // Keep configuration changes fenced until data deletion finishes;
+        // a force delete cannot bypass any Object Lock protection.
+        let object_lock_metadata_guard = if bucket_exists && let Some(guard) = object_lock_metadata_transaction {
+            let (state, incarnation, _) =
+                crate::bucket::metadata_sys::get_object_lock_config_and_incarnation_from_disk_with_guard_in(
+                    &self.ctx,
+                    bucket,
+                    Some(ns_guard.as_ref().unwrap_or(&guard)),
+                )
+                .await?;
+            match state {
+                crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent => Some((guard, incarnation)),
+                crate::bucket::metadata_sys::ObjectLockConfigState::Configured { .. } => {
+                    return Err(StorageError::PrefixAccessDenied(bucket.to_owned(), String::new()));
+                }
+                crate::bucket::metadata_sys::ObjectLockConfigState::Fabricated => {
+                    return Err(StorageError::FileCorrupt);
+                }
+            }
+        } else {
+            None
+        };
+
        if sr_delete && !bucket_exists {
            delete_opts.force_if_empty = true;
        }
@@ -1415,7 +1468,12 @@ impl ECStore {
                ns_guard.as_ref(),
                bucket,
                "bucket delete marker creation",
-                self.mark_bucket_deleted(bucket),
+                await_bucket_namespace_operation(
+                    object_lock_metadata_guard.as_ref().map(|(guard, _)| guard),
+                    bucket,
+                    "Object Lock configuration fence during bucket retirement",
+                    self.mark_bucket_deleted(bucket),
+                ),
            )
            .await?;
        }
@@ -1424,9 +1482,13 @@ impl ECStore {
        // Legacy buckets without a stamp cannot produce retirement authority.
        let retirement = if bucket_exists && bucket_lifecycle_guard.is_some() && !is_meta_bucketname(bucket) {
            if let Some(store) = metadata_sys::object_store_if_initialized_in(&self.ctx).await {
-                crate::bucket::metadata::load_bucket_incarnation(store.clone(), bucket)
-                    .await?
-                    .map(|incarnation| (store, incarnation))
+                if let Some((_, incarnation)) = object_lock_metadata_guard.as_ref() {
+                    Some((store, *incarnation))
+                } else {
+                    crate::bucket::metadata::load_bucket_incarnation(store.clone(), bucket)
+                        .await?
+                        .map(|incarnation| (store, incarnation))
+                }
            } else {
                None
            }
@@ -1438,13 +1500,19 @@ impl ECStore {
            bucket_lifecycle_guard.as_ref(),
            bucket,
            "physical bucket deletion",
-            run_physical_bucket_deletion(ns_guard.as_ref(), bucket, async {
-                self.delete_bucket_on_sets(bucket, &delete_opts)
-                    .await
-                    .map_err(|err| to_object_err(err, vec![bucket]))
-            }),
+            await_bucket_namespace_operation(
+                object_lock_metadata_guard.as_ref().map(|(guard, _)| guard),
+                bucket,
+                "Object Lock configuration fence during bucket deletion",
+                run_physical_bucket_deletion(ns_guard.as_ref(), bucket, async {
+                    self.delete_bucket_on_sets(bucket, &delete_opts)
+                        .await
+                        .map_err(|err| to_object_err(err, vec![bucket]))
+                }),
+            ),
        )
        .await;
+        drop(object_lock_metadata_guard);
        if let Err(err) = delete_result
            && (!sr_delete || !is_err_strict_volume_not_found(&err))
        {
@@ -3689,6 +3757,116 @@ mod tests {
        assert!(!any_disk_path_exists(&disk_paths, bucket_deleted_marker_volume(&bucket)).await);
    }

+    #[tokio::test]
+    #[serial]
+    async fn force_bucket_delete_preserves_object_lock_versions() {
+        let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4], Some(2)).await;
+        metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = format!("force-delete-worm-{}", Uuid::new_v4().simple());
+        store
+            .make_bucket(
+                &bucket,
+                &MakeBucketOptions {
+                    lock_enabled: true,
+                    versioning_enabled: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("create Object Lock bucket");
+        let until = (OffsetDateTime::now_utc() + time::Duration::days(365))
+            .format(&time::format_description::well_known::Rfc3339)
+            .expect("format future retention date");
+        let written = store
+            .put_object(
+                &bucket,
+                "retained",
+                &mut PutObjReader::from_vec(b"retained bytes".to_vec()),
+                &ObjectOptions {
+                    versioned: true,
+                    user_defined: HashMap::from([
+                        ("x-amz-object-lock-mode".to_owned(), "COMPLIANCE".to_owned()),
+                        ("x-amz-object-lock-retain-until-date".to_owned(), until),
+                    ]),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("write retained object");
+        for operation in [SRBucketDeleteOp::NoOp, SRBucketDeleteOp::MarkDelete, SRBucketDeleteOp::Purge] {
+            let err = store
+                .delete_bucket(
+                    &bucket,
+                    &DeleteBucketOptions {
+                        force: true,
+                        srdelete_op: operation,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect_err("force deletion must not bypass Object Lock");
+            assert!(matches!(err, StorageError::PrefixAccessDenied(_, _)), "unexpected denial: {err}");
+            let info = store
+                .get_object_info(
+                    &bucket,
+                    "retained",
+                    &ObjectOptions {
+                        version_id: written.version_id.map(|version| version.to_string()),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("retained version must survive the rejected bucket deletion");
+            assert_eq!(info.etag, written.etag);
+            assert_eq!(info.size, written.size);
+            assert_eq!(info.version_id, written.version_id);
+        }
+        metadata_sys::inject_object_lock_disk_read_error_in(&store.ctx, &bucket)
+            .await
+            .expect("inject authoritative metadata read failure");
+        let error = store
+            .delete_bucket(
+                &bucket,
+                &DeleteBucketOptions {
+                    force: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("an unreadable Object Lock configuration must fail closed");
+        assert!(error.to_string().contains("injected Object Lock metadata disk read failure"));
+        let info = store
+            .get_object_info(&bucket, "retained", &ObjectOptions::default())
+            .await
+            .expect("metadata read failure must leave the retained object intact");
+        assert_eq!(info.version_id, written.version_id);
+
+        let unlocked = format!("force-delete-unlocked-{}", Uuid::new_v4().simple());
+        store
+            .make_bucket(&unlocked, &MakeBucketOptions::default())
+            .await
+            .expect("create unlocked control bucket");
+        store
+            .put_object(
+                &unlocked,
+                "ordinary",
+                &mut PutObjReader::from_vec(b"ordinary bytes".to_vec()),
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("write ordinary control object");
+        store
+            .delete_bucket(
+                &unlocked,
+                &DeleteBucketOptions {
+                    force: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("force deletion must still work for an unlocked bucket");
+    }
+
    #[tokio::test]
    #[serial]
    async fn bucket_delete_mark_delete_rejects_hidden_object_paths_without_force() {
@@ -18318,7 +18318,7 @@ mod tests {
        let bucket = format!("lock-create-intent-retry-{}", uuid::Uuid::new_v4());
        let mut intent = crate::bucket::metadata::BucketMetadata::new(&bucket);
        intent.lock_enabled = true;
-        crate::bucket::metadata_sys::set_new_bucket_metadata_in(&ctx, intent)
+        crate::bucket::metadata_sys::set_new_bucket_metadata_intent_in(&ctx, intent)
            .await
            .expect("persist pre-visibility lock intent");
        assert!(
@@ -448,8 +448,8 @@ const MAX_LIST_OBJECTS_METADATA_FAST_STALENESS_MS: u64 = 60_000;
 const LIST_OBJECTS_INDEX_PROVIDER_WALKER_KEY_ONLY: &str = "walker_key_only";
 const LIST_OBJECTS_INDEX_PROVIDER_PERSISTENT_KEY_ONLY: &str = "persistent_key_only";
 const LIST_OBJECTS_INDEX_PROVIDER_PERSISTENT_KEY_ONLY_DEFAULT_GENERATION: &str = "persistent-key-only";
-const PERSISTENT_KEY_ONLY_INDEX_FORMAT_VERSION: u8 = 2;
-const PERSISTENT_KEY_ONLY_INDEX_HEADER: &str = "# rustfs-listobjects-key-only-v2";
+const PERSISTENT_KEY_ONLY_INDEX_FORMAT_VERSION: u8 = 3;
+const PERSISTENT_KEY_ONLY_INDEX_HEADER: &str = "# rustfs-listobjects-key-only-v3";
 const PERSISTENT_KEY_ONLY_INDEX_BUCKET_HEADER: &str = "# bucket=";
 const PERSISTENT_KEY_ONLY_INDEX_GENERATION_HEADER: &str = "# generation=";
 const PERSISTENT_KEY_ONLY_INDEX_CHECKPOINT_HEADER: &str = "# checkpoint_high_water_mark=";
@@ -656,7 +656,10 @@ impl PersistentListMetadataObject {
            name: object.name.clone(),
            size: object.size,
            mod_time: object.mod_time,
-            etag: object.etag.clone(),
+            etag: object
+                .etag
+                .clone()
+                .filter(|_| rustfs_utils::http::etag_can_be_listed(&object.user_defined)),
            storage_class: object.storage_class.clone(),
        }
    }
@@ -2361,23 +2364,43 @@ fn list_objects_paginate(
    (objects, prefixes, is_truncated, next_marker, next_version_idmarker)
 }

-fn list_objects_paginate_versions<'a>(
-    mut get_objects: Vec<ObjectInfo>,
-    delimiter: &Option<String>,
+struct VersionedListPagination<'a, I> {
+    delimiter: &'a Option<String>,
    max_keys: i32,
    disk_has_more: bool,
-    cache_id: Option<&str>,
-    marker: Option<&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 =
-        versioned_prefix_page_boundary_key(&get_objects, raw_keys, delimiter.as_deref(), max_keys, disk_has_more);
+    cache_id: Option<&'a str>,
+    marker: Option<&'a str>,
+    last_scanned_key: Option<&'a str>,
+    raw_keys: I,
+}

-    let (objects, prefixes, is_truncated, mut next_marker, mut next_version_idmarker) =
-        list_objects_paginate(get_objects, delimiter, max_keys, disk_has_more, cache_id, true, last_scanned_key);
+fn list_objects_paginate_versions<'a, I>(
+    mut get_objects: Vec<ObjectInfo>,
+    pagination: VersionedListPagination<'a, I>,
+) -> (Vec<ObjectInfo>, Vec<String>, bool, Option<String>, Option<String>)
+where
+    I: Iterator<Item = &'a str>,
+{
+    filter_versions_common_prefixes_after_marker(&mut get_objects, pagination.delimiter.as_deref(), pagination.marker);
+    let prefix_at_page_boundary = versioned_prefix_page_boundary_key(
+        &get_objects,
+        pagination.raw_keys,
+        pagination.delimiter.as_deref(),
+        pagination.max_keys,
+        pagination.disk_has_more,
+    );
+
+    let (objects, prefixes, is_truncated, mut next_marker, mut next_version_idmarker) = list_objects_paginate(
+        get_objects,
+        pagination.delimiter,
+        pagination.max_keys,
+        pagination.disk_has_more,
+        pagination.cache_id,
+        true,
+        pagination.last_scanned_key,
+    );
    if is_truncated && let Some(raw_key) = prefix_at_page_boundary {
-        next_marker = Some(append_list_cache_id_to_marker(raw_key, cache_id));
+        next_marker = Some(append_list_cache_id_to_marker(raw_key, pagination.cache_id));
        next_version_idmarker = None;
    }

@@ -4325,12 +4348,15 @@ impl ECStore {

        let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions(
            get_objects,
-            &delimiter,
-            max_keys,
-            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())),
+            VersionedListPagination {
+                delimiter: &delimiter,
+                max_keys,
+                disk_has_more,
+                cache_id: next_cache_id.as_deref(),
+                marker: opts.marker.as_deref(),
+                last_scanned_key: last_scanned_key.as_deref(),
+                raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()),
+            },
        );

        Ok(ListObjectVersionsInfo {
@@ -5752,12 +5778,15 @@ impl Sets {

        let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions(
            get_objects,
-            &delimiter,
-            max_keys,
-            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())),
+            VersionedListPagination {
+                delimiter: &delimiter,
+                max_keys,
+                disk_has_more,
+                cache_id: next_cache_id.as_deref(),
+                marker: opts.marker.as_deref(),
+                last_scanned_key: last_scanned_key.as_deref(),
+                raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()),
+            },
        );

        Ok(ListObjectVersionsInfo {
@@ -6565,12 +6594,15 @@ impl SetDisks {

        let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions(
            get_objects,
-            &delimiter,
-            max_keys,
-            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())),
+            VersionedListPagination {
+                delimiter: &delimiter,
+                max_keys,
+                disk_has_more,
+                cache_id: next_cache_id.as_deref(),
+                marker: opts.marker.as_deref(),
+                last_scanned_key: last_scanned_key.as_deref(),
+                raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()),
+            },
        );

        Ok(ListObjectVersionsInfo {
@@ -8844,12 +8876,15 @@ mod test {
            let get_objects = fold_delimiter_page(&window, "", "-");
            let (objects, prefixes, is_truncated, next_marker, _v) = super::list_objects_paginate_versions(
                get_objects,
-                &delimiter,
-                max_keys,
-                disk_has_more,
-                None,
-                marker.as_deref(),
-                (last_scanned.as_deref(), window.iter().map(String::as_str)),
+                super::VersionedListPagination {
+                    delimiter: &delimiter,
+                    max_keys,
+                    disk_has_more,
+                    cache_id: None,
+                    marker: marker.as_deref(),
+                    last_scanned_key: last_scanned.as_deref(),
+                    raw_keys: window.iter().map(String::as_str),
+                },
            );

            assert!(
@@ -8922,12 +8957,15 @@ mod test {
            let get_objects = fold_delimiter_page(&window, "", "-");
            let (objects, prefixes, is_truncated, next_marker, _) = super::list_objects_paginate_versions(
                get_objects,
-                &delimiter,
-                max_keys,
-                disk_has_more,
-                None,
-                marker.as_deref(),
-                (last_scanned.as_deref(), window.iter().map(String::as_str)),
+                super::VersionedListPagination {
+                    delimiter: &delimiter,
+                    max_keys,
+                    disk_has_more,
+                    cache_id: None,
+                    marker: marker.as_deref(),
+                    last_scanned_key: last_scanned.as_deref(),
+                    raw_keys: window.iter().map(String::as_str),
+                },
            );

            assert_eq!(objects.len() + prefixes.len(), 1, "a page must respect max-keys");
@@ -9170,6 +9208,38 @@ mod test {
        assert_eq!(parsed, object);
    }

+    #[test]
+    fn persistent_listing_snapshots_cannot_restore_legacy_encrypted_fingerprints() {
+        let mut object = ObjectInfo {
+            name: "encrypted".to_owned(),
+            etag: Some("fbfcbcc6d2f035411e6268c8b119593e".to_owned()),
+            user_defined: Arc::new(HashMap::from([(
+                "x-amz-server-side-encryption-customer-algorithm".to_owned(),
+                "AES256".to_owned(),
+            )])),
+            ..Default::default()
+        };
+        let snapshot = PersistentListMetadataObject::from_object_info(&object);
+        assert!(snapshot.to_object_info("bucket").etag.is_none());
+        let mut old_snapshot = snapshot.clone();
+        old_snapshot.etag = object.etag.clone();
+        let old_contents = format!(
+            "# rustfs-listobjects-key-only-v2\n{}\n",
+            encode_persistent_list_metadata_object(&old_snapshot)
+        );
+        let old_index = parse_persistent_key_only_index(&old_contents);
+        assert_ne!(old_index.format_version, PERSISTENT_KEY_ONLY_INDEX_FORMAT_VERSION);
+
+        rustfs_utils::http::metadata_compat::insert_str(
+            Arc::make_mut(&mut object.user_defined),
+            rustfs_utils::http::SUFFIX_OPAQUE_ENCRYPTED_ETAG,
+            "v1".to_owned(),
+        );
+        assert_eq!(PersistentListMetadataObject::from_object_info(&object).etag, object.etag);
+        object.user_defined = Arc::new(HashMap::new());
+        assert_eq!(PersistentListMetadataObject::from_object_info(&object).etag, object.etag);
+    }
+
    #[test]
    fn persistent_key_only_index_parses_legacy_keys_and_metadata_snapshots() {
        let object = PersistentListMetadataObject {
@@ -5351,13 +5351,41 @@ impl ECStore {
        tier_journal_api: Option<Arc<ECStore>>,
    ) -> (Vec<DeletedObject>, Vec<Option<Error>>, Vec<Option<DeleteAccounting>>) {
        Box::pin(async move {
+            let count = objects.len();
+            let mut deleted = vec![DeletedObject::default(); count];
+            let mut errors: Vec<Option<Error>> = (0..count).map(|_| None).collect();
+            let mut accounting = vec![None; count];
+            let mut valid_objects = Vec::with_capacity(count);
+            let mut valid_indices = Vec::with_capacity(count);
+            // Validate the literal key before encoding it or taking locks. A
+            // dot-segment key must never reach filesystem path resolution.
+            for (index, object) in objects.into_iter().enumerate() {
+                match check_del_obj_args(bucket, &object.object_name) {
+                    Ok(()) => {
+                        valid_indices.push(index);
+                        valid_objects.push(object);
+                    }
+                    Err(err) => errors[index] = Some(err),
+                }
+            }
+            if valid_objects.is_empty() {
+                return (deleted, errors, accounting);
+            }
            let mut opts = opts;
            let receipt_sink = install_tier_free_version_receipt_sink(&mut opts);
-            let result = self
-                .handle_delete_objects_with_journal_and_accounting_inner(bucket, objects, opts, tier_journal_api)
+            let (valid_deleted, valid_errors, valid_accounting) = self
+                .handle_delete_objects_with_journal_and_accounting_inner(bucket, valid_objects, opts, tier_journal_api)
                .await;
            enqueue_recorded_tier_free_versions(self, receipt_sink).await;
-            result
+            for (index, ((object, error), receipt)) in valid_indices
+                .into_iter()
+                .zip(valid_deleted.into_iter().zip(valid_errors).zip(valid_accounting))
+            {
+                deleted[index] = object;
+                errors[index] = error;
+                accounting[index] = receipt;
+            }
+            (deleted, errors, accounting)
        })
        .await
    }
@@ -9094,6 +9122,58 @@ mod tests {
        (dirs, store)
    }

+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn batch_delete_rejects_dot_segments_and_keeps_valid_result_positions() {
+        let bucket = "batch-delete-key-validation";
+        let (_dirs, store) = multipool_version_test_store(bucket).await;
+        let options = ObjectOptions {
+            versioned: true,
+            ..Default::default()
+        };
+        let victim = store
+            .put_object(bucket, "victim/object", &mut PutObjReader::from_vec(b"victim".to_vec()), &options)
+            .await
+            .expect("write protected-prefix control object");
+        store
+            .put_object(bucket, "allowed/object", &mut PutObjReader::from_vec(b"allowed".to_vec()), &options)
+            .await
+            .expect("write valid deletion control object");
+        let keys = [
+            "allowed/../victim/object",
+            "allowed/object",
+            "a/./b",
+            "../other-bucket/object",
+        ];
+        let objects = keys
+            .iter()
+            .map(|key| ObjectToDelete {
+                object_name: (*key).to_owned(),
+                ..Default::default()
+            })
+            .collect();
+        let (deleted, errors) = store.handle_delete_objects(bucket, objects, options).await;
+        assert_eq!(deleted.len(), keys.len());
+        assert_eq!(errors.len(), keys.len());
+        for index in [0, 2, 3] {
+            assert!(
+                matches!(&errors[index], Some(StorageError::ObjectNameInvalid(_, _))),
+                "invalid key {}: {:?}",
+                keys[index],
+                errors[index]
+            );
+            assert!(!deleted[index].delete_marker);
+        }
+        assert!(errors[1].is_none(), "valid deletion must still succeed: {:?}", errors[1]);
+        assert!(deleted[1].delete_marker);
+        assert_eq!(deleted[1].object_name, "allowed/object");
+        let after = store
+            .get_object_info(bucket, "victim/object", &ObjectOptions::default())
+            .await
+            .expect("the resolved victim path must remain visible");
+        assert_eq!(after.version_id, victim.version_id);
+        assert_eq!(after.etag, victim.etag);
+    }
+
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn retired_marker_converges_after_bucket_recreation_on_sixteen_disks() {
        let bucket = "retired-marker-c11";
@@ -1,497 +0,0 @@
-// Copyright 2026 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Byte-level goldens for the 13 persisted bucket-configuration XML families
-//! (rustfs/backlog#2744).
-//!
-//! Responsible for: pinning the exact bytes that `deserialize` followed by
-//! `serialize` produces for every family in its three source shapes (AWS
-//! documentation shape, MinIO-written shape, RustFS-written shape), and the
-//! exact bytes RustFS writes for a typed value. Only bytes are compared; a
-//! structural comparison would let a codec replacement change the persisted
-//! encoding without failing here.
-//!
-//! Not responsible for: validating configurations, touching the bucket
-//! metadata store, or choosing a canonical shape.
-//!
-//! Upstream: the bucket-config `deserialize`/`serialize` pair, reached through the
-//! integration-test facade `tests/storage_api.rs`, and the s3s DTOs they are generic over. Downstream: the DTO/codec
-//! replacement (T1.4) and the persisted-bytes comparison (T3.7), which must
-//! keep every golden green without regenerating it. Regeneration is reserved
-//! to `scripts/gen_bucket_config_goldens.sh`, which sets
-//! `BUCKET_CONFIG_GOLDENS_WRITE`: in that mode every test writes its files
-//! and then fails on purpose, so a write run never prints a green line.
-
-mod storage_api;
-
-use s3s::dto::{
-    AbortIncompleteMultipartUpload, AccelerateConfiguration, BucketAccelerateStatus, BucketLifecycleConfiguration,
-    BucketLoggingStatus, BucketVersioningStatus, CORSConfiguration, CORSRule, Condition, DefaultRetention, DelMarkerExpiration,
-    DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination,
-    ErrorDocument, Event, ExcludedPrefix, ExistingObjectReplication, ExistingObjectReplicationStatus, ExpirationStatus,
-    FilterRule, FilterRuleName, IndexDocument, LifecycleExpiration, LifecycleRule, LifecycleRuleAndOperator, LifecycleRuleFilter,
-    LoggingEnabled, NoncurrentVersionExpiration, NotificationConfiguration, NotificationConfigurationFilter,
-    ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRetentionMode, ObjectLockRule, Payer, PublicAccessBlockConfiguration,
-    QueueConfiguration, Redirect, ReplicaModifications, ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule,
-    ReplicationRuleAndOperator, ReplicationRuleFilter, ReplicationRuleStatus, RequestPaymentConfiguration, RoutingRule,
-    S3KeyFilter, ServerSideEncryption, ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration,
-    ServerSideEncryptionRule, SourceSelectionCriteria, StorageClass, Tag, Tagging, Transition, TransitionStorageClass,
-    VersioningConfiguration, WebsiteConfiguration,
-};
-use s3s::xml;
-use std::path::{Path, PathBuf};
-use storage_api::bucket_config_codec::{deserialize, serialize};
-
-/// Set only by `scripts/gen_bucket_config_goldens.sh`. Every pair test then
-/// writes its `.out.xml` (and, for the `rustfs` shape, its `.in.xml`) and
-/// fails on purpose, so a write run can never be mistaken for a verification.
-const WRITE_ENV: &str = "BUCKET_CONFIG_GOLDENS_WRITE";
-
-fn goldens_dir() -> PathBuf {
-    Path::new(env!("CARGO_MANIFEST_DIR"))
-        .join("tests")
-        .join("fixtures")
-        .join("bucket-config-goldens")
-}
-
-fn pair_path(family: &str, shape: &str, side: &str) -> PathBuf {
-    goldens_dir().join(family).join(format!("{shape}.{side}.xml"))
-}
-
-fn read(path: &Path) -> Vec<u8> {
-    std::fs::read(path).unwrap_or_else(|err| panic!("read {}: {err}", path.display()))
-}
-
-fn write(path: &Path, bytes: &[u8]) {
-    std::fs::write(path, bytes).unwrap_or_else(|err| panic!("write {}: {err}", path.display()));
-}
-
-fn contains(haystack: &[u8], needle: &[u8]) -> bool {
-    haystack.windows(needle.len()).any(|window| window == needle)
-}
-
-fn round_trip<T>(input: &[u8], label: &str) -> Vec<u8>
-where
-    T: for<'xml> xml::Deserialize<'xml> + xml::Serialize,
-{
-    let parsed: T = deserialize(input).unwrap_or_else(|err| panic!("{label}: deserialize failed: {err}"));
-    serialize(&parsed).unwrap_or_else(|err| panic!("{label}: serialize failed: {err}"))
-}
-
-/// Byte comparison with a readable report. Parsed values are never compared.
-fn assert_bytes_eq(actual: &[u8], expected: &[u8], context: &str) {
-    if actual != expected {
-        panic!(
-            "{context}\n--- expected ({} bytes) ---\n{}\n--- actual ({} bytes) ---\n{}\n",
-            expected.len(),
-            String::from_utf8_lossy(expected),
-            actual.len(),
-            String::from_utf8_lossy(actual),
-        );
-    }
-}
-
-/// Checks one `<family>/<shape>` pair. `typed` is the value RustFS writes for
-/// the `rustfs` shape; the `aws` and `minio` shapes pass `None`.
-fn check_pair<T>(family: &str, shape: &str, typed: Option<&T>)
-where
-    T: for<'xml> xml::Deserialize<'xml> + xml::Serialize,
-{
-    let in_path = pair_path(family, shape, "in");
-    let out_path = pair_path(family, shape, "out");
-    let label = format!("{family}/{shape}");
-
-    if std::env::var_os(WRITE_ENV).is_some() {
-        if let Some(value) = typed {
-            let written = serialize(value).unwrap_or_else(|err| panic!("{label}: typed value failed to serialize: {err}"));
-            write(&in_path, &written);
-        }
-        let input = read(&in_path);
-        write(&out_path, &round_trip::<T>(&input, &label));
-        panic!(
-            "{WRITE_ENV} is set: wrote {} without verifying anything; rerun without {WRITE_ENV}",
-            out_path.display()
-        );
-    }
-
-    let input = read(&in_path);
-    let expected = read(&out_path);
-    if let Some(value) = typed {
-        let written = serialize(value).unwrap_or_else(|err| panic!("{label}: typed value failed to serialize: {err}"));
-        assert_bytes_eq(
-            &written,
-            &input,
-            &format!(
-                "{label}: RustFS no longer writes the pinned bytes for the typed value ({})",
-                in_path.display()
-            ),
-        );
-    }
-    let actual = round_trip::<T>(&input, &label);
-    assert_bytes_eq(
-        &actual,
-        &expected,
-        &format!("{label}: deserialize->serialize bytes drifted from {}", out_path.display()),
-    );
-    let again = round_trip::<T>(&expected, &label);
-    assert_bytes_eq(
-        &again,
-        &expected,
-        &format!("{label}: the golden is not a fixed point of deserialize->serialize"),
-    );
-}
-
-/// The typed values behind every `rustfs.in.xml`: what RustFS writes today
-/// through `serialize` when it persists a configuration it built itself.
-/// MinIO-only members (`DelMarkerExpiration`, `ExcludeFolders`,
-/// `ExcludedPrefixes`, `DeleteReplication`) are included where RustFS accepts
-/// them, so the `rustfs` shape is the one that pins their encoding.
-mod rustfs_values {
-    use super::*;
-
-    fn tag(key: &str, value: &str) -> Tag {
-        Tag {
-            key: Some(key.to_owned()),
-            value: Some(value.to_owned()),
-        }
-    }
-
-    pub(super) fn notification() -> NotificationConfiguration {
-        NotificationConfiguration {
-            queue_configurations: Some(vec![QueueConfiguration {
-                events: vec![
-                    Event::from("s3:ObjectCreated:*".to_owned()),
-                    Event::from("s3:ObjectRemoved:*".to_owned()),
-                ],
-                filter: Some(NotificationConfigurationFilter {
-                    key: Some(S3KeyFilter {
-                        filter_rules: Some(vec![FilterRule {
-                            name: Some(FilterRuleName::from_static(FilterRuleName::PREFIX)),
-                            value: Some("uploads/".to_owned()),
-                        }]),
-                    }),
-                }),
-                id: Some("rustfs-webhook".to_owned()),
-                queue_arn: "arn:rustfs:sqs::primary:webhook".to_owned(),
-            }]),
-            ..Default::default()
-        }
-    }
-
-    pub(super) fn lifecycle() -> BucketLifecycleConfiguration {
-        BucketLifecycleConfiguration {
-            rules: vec![
-                LifecycleRule {
-                    abort_incomplete_multipart_upload: Some(AbortIncompleteMultipartUpload {
-                        days_after_initiation: Some(3),
-                    }),
-                    del_marker_expiration: Some(DelMarkerExpiration { days: Some(7) }),
-                    expiration: Some(LifecycleExpiration {
-                        days: Some(30),
-                        ..Default::default()
-                    }),
-                    filter: Some(LifecycleRuleFilter {
-                        prefix: Some("tmp/".to_owned()),
-                        ..Default::default()
-                    }),
-                    id: Some("expire-tmp".to_owned()),
-                    noncurrent_version_expiration: Some(NoncurrentVersionExpiration {
-                        newer_noncurrent_versions: Some(2),
-                        noncurrent_days: Some(7),
-                    }),
-                    noncurrent_version_transitions: None,
-                    prefix: None,
-                    status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
-                    transitions: None,
-                },
-                LifecycleRule {
-                    abort_incomplete_multipart_upload: None,
-                    del_marker_expiration: None,
-                    expiration: None,
-                    filter: Some(LifecycleRuleFilter {
-                        and: Some(LifecycleRuleAndOperator {
-                            prefix: Some("archive/".to_owned()),
-                            tags: Some(vec![tag("tier", "cold")]),
-                            ..Default::default()
-                        }),
-                        ..Default::default()
-                    }),
-                    id: Some("tier-cold".to_owned()),
-                    noncurrent_version_expiration: None,
-                    noncurrent_version_transitions: None,
-                    prefix: None,
-                    status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
-                    transitions: Some(vec![Transition {
-                        days: Some(90),
-                        storage_class: Some(TransitionStorageClass::from_static("COLDTIER")),
-                        ..Default::default()
-                    }]),
-                },
-            ],
-            ..Default::default()
-        }
-    }
-
-    pub(super) fn object_lock() -> ObjectLockConfiguration {
-        ObjectLockConfiguration {
-            object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)),
-            rule: Some(ObjectLockRule {
-                default_retention: Some(DefaultRetention {
-                    days: Some(30),
-                    mode: Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::GOVERNANCE)),
-                    years: None,
-                }),
-            }),
-        }
-    }
-
-    pub(super) fn versioning() -> VersioningConfiguration {
-        VersioningConfiguration {
-            exclude_folders: Some(true),
-            excluded_prefixes: Some(vec![
-                ExcludedPrefix {
-                    prefix: Some("tmp/".to_owned()),
-                },
-                ExcludedPrefix {
-                    prefix: Some("cache/".to_owned()),
-                },
-            ]),
-            mfa_delete: None,
-            status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
-        }
-    }
-
-    pub(super) fn encryption() -> ServerSideEncryptionConfiguration {
-        ServerSideEncryptionConfiguration {
-            rules: vec![ServerSideEncryptionRule {
-                apply_server_side_encryption_by_default: Some(ServerSideEncryptionByDefault {
-                    kms_master_key_id: Some("rustfs-bucket-key".to_owned()),
-                    sse_algorithm: ServerSideEncryption::from_static(ServerSideEncryption::AWS_KMS),
-                }),
-                blocked_encryption_types: None,
-                bucket_key_enabled: Some(false),
-            }],
-        }
-    }
-
-    pub(super) fn tagging() -> Tagging {
-        Tagging {
-            tag_set: vec![tag("team", "storage"), tag("env", "staging")],
-        }
-    }
-
-    pub(super) fn replication() -> ReplicationConfiguration {
-        ReplicationConfiguration {
-            role: String::new(),
-            rules: vec![ReplicationRule {
-                delete_marker_replication: Some(DeleteMarkerReplication {
-                    status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)),
-                }),
-                delete_replication: Some(DeleteReplication {
-                    status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED),
-                }),
-                destination: Destination {
-                    bucket: "arn:rustfs:replication::0b6f3c2a-5e6a-4d2b-9c1e-7a8f9d0e1b2c:interop-dr".to_owned(),
-                    storage_class: Some(StorageClass::from_static(StorageClass::STANDARD)),
-                    ..Default::default()
-                },
-                existing_object_replication: Some(ExistingObjectReplication {
-                    status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED),
-                }),
-                filter: Some(ReplicationRuleFilter {
-                    and: Some(ReplicationRuleAndOperator {
-                        prefix: Some("docs/".to_owned()),
-                        tags: Some(vec![tag("replicate", "yes")]),
-                    }),
-                    ..Default::default()
-                }),
-                id: Some("rustfs-dr".to_owned()),
-                prefix: None,
-                priority: Some(1),
-                source_selection_criteria: Some(SourceSelectionCriteria {
-                    replica_modifications: Some(ReplicaModifications {
-                        status: ReplicaModificationsStatus::from_static(ReplicaModificationsStatus::ENABLED),
-                    }),
-                    sse_kms_encrypted_objects: None,
-                }),
-                status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
-            }],
-        }
-    }
-
-    pub(super) fn cors() -> CORSConfiguration {
-        CORSConfiguration {
-            cors_rules: vec![CORSRule {
-                allowed_headers: Some(vec!["Authorization".to_owned(), "Content-Type".to_owned()]),
-                allowed_methods: vec!["GET".to_owned(), "PUT".to_owned()],
-                allowed_origins: vec!["https://app.example.internal".to_owned()],
-                expose_headers: Some(vec!["ETag".to_owned(), "x-amz-request-id".to_owned()]),
-                id: Some("rustfs-app".to_owned()),
-                max_age_seconds: Some(3600),
-            }],
-        }
-    }
-
-    pub(super) fn logging() -> BucketLoggingStatus {
-        BucketLoggingStatus {
-            logging_enabled: Some(LoggingEnabled {
-                target_bucket: "rustfs-access-logs".to_owned(),
-                target_grants: None,
-                target_object_key_format: None,
-                target_prefix: "buckets/interop/".to_owned(),
-            }),
-        }
-    }
-
-    pub(super) fn website() -> WebsiteConfiguration {
-        WebsiteConfiguration {
-            error_document: Some(ErrorDocument {
-                key: "404.html".to_owned(),
-            }),
-            index_document: Some(IndexDocument {
-                suffix: "index.html".to_owned(),
-            }),
-            redirect_all_requests_to: None,
-            routing_rules: Some(vec![RoutingRule {
-                condition: Some(Condition {
-                    http_error_code_returned_equals: None,
-                    key_prefix_equals: Some("legacy/".to_owned()),
-                }),
-                redirect: Redirect {
-                    host_name: None,
-                    http_redirect_code: Some("301".to_owned()),
-                    protocol: None,
-                    replace_key_prefix_with: Some("current/".to_owned()),
-                    replace_key_with: None,
-                },
-            }]),
-        }
-    }
-
-    pub(super) fn accelerate() -> AccelerateConfiguration {
-        AccelerateConfiguration {
-            status: Some(BucketAccelerateStatus::from_static(BucketAccelerateStatus::SUSPENDED)),
-        }
-    }
-
-    pub(super) fn request_payment() -> RequestPaymentConfiguration {
-        RequestPaymentConfiguration {
-            payer: Payer::from_static(Payer::BUCKET_OWNER),
-        }
-    }
-
-    pub(super) fn public_access_block() -> PublicAccessBlockConfiguration {
-        PublicAccessBlockConfiguration {
-            block_public_acls: Some(true),
-            block_public_policy: Some(false),
-            ignore_public_acls: Some(true),
-            restrict_public_buckets: Some(false),
-        }
-    }
-}
-
-/// One test per `<family>/<shape>` pair: 13 families x 3 shapes = 39.
-mod bucket_config_goldens {
-    use super::*;
-
-    macro_rules! golden {
-        ($name:ident, $family:literal, $shape:literal, $ty:ty) => {
-            #[test]
-            fn $name() {
-                check_pair::<$ty>($family, $shape, None);
-            }
-        };
-        ($name:ident, $family:literal, rustfs = $value:expr) => {
-            #[test]
-            fn $name() {
-                check_pair($family, "rustfs", Some(&$value));
-            }
-        };
-    }
-
-    golden!(notification_aws, "notification", "aws", NotificationConfiguration);
-    golden!(notification_minio, "notification", "minio", NotificationConfiguration);
-    golden!(notification_rustfs, "notification", rustfs = rustfs_values::notification());
-
-    golden!(lifecycle_aws, "lifecycle", "aws", BucketLifecycleConfiguration);
-    golden!(lifecycle_minio, "lifecycle", "minio", BucketLifecycleConfiguration);
-    golden!(lifecycle_rustfs, "lifecycle", rustfs = rustfs_values::lifecycle());
-
-    golden!(object_lock_aws, "object_lock", "aws", ObjectLockConfiguration);
-    golden!(object_lock_minio, "object_lock", "minio", ObjectLockConfiguration);
-    golden!(object_lock_rustfs, "object_lock", rustfs = rustfs_values::object_lock());
-
-    golden!(versioning_aws, "versioning", "aws", VersioningConfiguration);
-    golden!(versioning_minio, "versioning", "minio", VersioningConfiguration);
-    golden!(versioning_rustfs, "versioning", rustfs = rustfs_values::versioning());
-
-    golden!(encryption_aws, "encryption", "aws", ServerSideEncryptionConfiguration);
-    golden!(encryption_minio, "encryption", "minio", ServerSideEncryptionConfiguration);
-    golden!(encryption_rustfs, "encryption", rustfs = rustfs_values::encryption());
-
-    golden!(tagging_aws, "tagging", "aws", Tagging);
-    golden!(tagging_minio, "tagging", "minio", Tagging);
-    golden!(tagging_rustfs, "tagging", rustfs = rustfs_values::tagging());
-
-    /// `replication/aws.in.xml` is the unknown-element sample from
-    /// `crates/replication/src/config.rs`
-    /// (`s3_xml_parser_discards_unknown_replication_elements_before_validation`).
-    /// Beyond the byte golden, the pair proves the root-level `FutureTopLevel`
-    /// element is skipped on read and never re-emitted on write.
-    #[test]
-    fn replication_aws() {
-        check_pair::<ReplicationConfiguration>("replication", "aws", None);
-        let input = read(&pair_path("replication", "aws", "in"));
-        let output = read(&pair_path("replication", "aws", "out"));
-        assert!(
-            contains(&input, b"<FutureTopLevel>future</FutureTopLevel>"),
-            "the sample must still carry the unknown element"
-        );
-        assert!(
-            !contains(&output, b"FutureTopLevel"),
-            "an unknown element must not be re-emitted: {}",
-            String::from_utf8_lossy(&output)
-        );
-    }
-    golden!(replication_minio, "replication", "minio", ReplicationConfiguration);
-    golden!(replication_rustfs, "replication", rustfs = rustfs_values::replication());
-
-    golden!(cors_aws, "cors", "aws", CORSConfiguration);
-    golden!(cors_minio, "cors", "minio", CORSConfiguration);
-    golden!(cors_rustfs, "cors", rustfs = rustfs_values::cors());
-
-    golden!(logging_aws, "logging", "aws", BucketLoggingStatus);
-    golden!(logging_minio, "logging", "minio", BucketLoggingStatus);
-    golden!(logging_rustfs, "logging", rustfs = rustfs_values::logging());
-
-    golden!(website_aws, "website", "aws", WebsiteConfiguration);
-    golden!(website_minio, "website", "minio", WebsiteConfiguration);
-    golden!(website_rustfs, "website", rustfs = rustfs_values::website());
-
-    golden!(accelerate_aws, "accelerate", "aws", AccelerateConfiguration);
-    golden!(accelerate_minio, "accelerate", "minio", AccelerateConfiguration);
-    golden!(accelerate_rustfs, "accelerate", rustfs = rustfs_values::accelerate());
-
-    golden!(request_payment_aws, "request_payment", "aws", RequestPaymentConfiguration);
-    golden!(request_payment_minio, "request_payment", "minio", RequestPaymentConfiguration);
-    golden!(request_payment_rustfs, "request_payment", rustfs = rustfs_values::request_payment());
-
-    golden!(public_access_block_aws, "public_access_block", "aws", PublicAccessBlockConfiguration);
-    golden!(public_access_block_minio, "public_access_block", "minio", PublicAccessBlockConfiguration);
-    golden!(
-        public_access_block_rustfs,
-        "public_access_block",
-        rustfs = rustfs_values::public_access_block()
-    );
-}
@@ -1,58 +0,0 @@
-# Persisted bucket-configuration XML goldens (rustfs/backlog#2744)
-
-Byte-level goldens for the 13 XML families that `BucketMetadata` persists as
-`*_config_xml`. They pin what `rustfs_ecstore::api::bucket::utils::deserialize`
-followed by `serialize` emits today, so that the DTO/codec replacement (T1.4)
-and the persisted-bytes comparison (T3.7) can be proven byte-identical: an
-older binary must keep reading what a newer one writes, and the reverse.
-
-Generated at commit: `6b1554003ebf8f2037ffb7da9c9b906527e758da` (2026-10-07T17:40:56Z)
-
-Checked by `crates/ecstore/tests/bucket_config_goldens.rs`
-(`cargo nextest run -p rustfs-ecstore bucket_config_goldens`, 39 tests).
-Generated once by `scripts/gen_bucket_config_goldens.sh`, which records the
-commit above and refuses to run again unless `--force` is passed. Do not
-regenerate after a DTO or codec replacement: at that point a diff here is the
-finding, not noise.
-
-## Layout
-
-`<family>/<shape>.in.xml` is an input; `<family>/<shape>.out.xml` is the exact
-bytes of `serialize(deserialize(in))` on the generating commit. The test
-asserts `bytes == out` (never structural equality), that `out` is a fixed point
-of the same round trip, and for the `rustfs` shape that `serialize(typed
-value)` still equals `rustfs.in.xml`.
-
-| Shape | Source |
-| --- | --- |
-| `aws` | Written for this corpus from the public S3 XML schema in the shape of the AWS documentation examples: XML declaration, `xmlns` on the root, indented. No AWS prose is copied. |
-| `minio` | The bytes a MinIO server persists. Seven families are carved verbatim out of `../minio/bucket_metadata.blob.hex` (MinIO `RELEASE.2025-07-23T15-54-02Z`); the rest are hand-shaped as compact Go `encoding/xml` output with the S3 namespace on the root, because MinIO does not persist those families. |
-| `rustfs` | The bytes RustFS writes today through `serialize` for a typed value (`rustfs_values` in the test). This shape carries the MinIO-only members RustFS accepts (`DelMarkerExpiration`, `ExcludeFolders`, `ExcludedPrefixes`, `DeleteReplication`). |
-
-| Family | Root element | `minio.in.xml` provenance |
-| --- | --- | --- |
-| `notification` | `NotificationConfiguration` | MinIO fixture |
-| `lifecycle` | `LifecycleConfiguration` | MinIO fixture (carries the MinIO `ExpiryUpdatedAt` extension) |
-| `object_lock` | `ObjectLockConfiguration` | MinIO fixture |
-| `versioning` | `VersioningConfiguration` | MinIO fixture |
-| `encryption` | `ServerSideEncryptionConfiguration` | MinIO fixture |
-| `tagging` | `Tagging` | MinIO fixture |
-| `replication` | `ReplicationConfiguration` | MinIO fixture |
-| `cors` | `CORSConfiguration` | hand-shaped |
-| `logging` | `BucketLoggingStatus` | hand-shaped |
-| `website` | `WebsiteConfiguration` | hand-shaped |
-| `accelerate` | `AccelerateConfiguration` | hand-shaped |
-| `request_payment` | `RequestPaymentConfiguration` | hand-shaped |
-| `public_access_block` | `PublicAccessBlockConfiguration` | hand-shaped |
-
-## Unknown elements
-
-`replication/aws.in.xml` is the unknown-element sample from
-`crates/replication/src/config.rs`
-(`s3_xml_parser_discards_unknown_replication_elements_before_validation`),
-re-indented to two spaces. Its golden proves the root-level `FutureTopLevel`
-element is skipped on read and never re-emitted on write, which is the
-persisted-configuration exception recorded in the root `AGENTS.md`. On the
-generating commit every one of the 13 root elements skips unknown children;
-nested members (`Rule`, `Filter`, `Destination`, ...) reject them. A
-replacement codec must keep that boundary where it is.
@@ -1,4 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<AccelerateConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
-  <Status>Enabled</Status>
-</AccelerateConfiguration>
@@ -1 +0,0 @@
-<AccelerateConfiguration><Status>Enabled</Status></AccelerateConfiguration>
@@ -1 +0,0 @@
-<AccelerateConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>Suspended</Status></AccelerateConfiguration>
@@ -1 +0,0 @@
-<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>
@@ -1 +0,0 @@
-<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>
@@ -1 +0,0 @@
-<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>
@@ -1,17 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<CORSConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
-  <CORSRule>
-    <ID>AllowWebApp</ID>
-    <AllowedOrigin>https://www.example.com</AllowedOrigin>
-    <AllowedMethod>PUT</AllowedMethod>
-    <AllowedMethod>POST</AllowedMethod>
-    <AllowedMethod>DELETE</AllowedMethod>
-    <AllowedHeader>*</AllowedHeader>
-    <ExposeHeader>x-amz-server-side-encryption</ExposeHeader>
-    <MaxAgeSeconds>3000</MaxAgeSeconds>
-  </CORSRule>
-  <CORSRule>
-    <AllowedOrigin>*</AllowedOrigin>
-    <AllowedMethod>GET</AllowedMethod>
-  </CORSRule>
-</CORSConfiguration>
@@ -1 +0,0 @@
-<CORSConfiguration><CORSRule><AllowedHeader>*</AllowedHeader><AllowedMethod>PUT</AllowedMethod><AllowedMethod>POST</AllowedMethod><AllowedMethod>DELETE</AllowedMethod><AllowedOrigin>https://www.example.com</AllowedOrigin><ExposeHeader>x-amz-server-side-encryption</ExposeHeader><ID>AllowWebApp</ID><MaxAgeSeconds>3000</MaxAgeSeconds></CORSRule><CORSRule><AllowedMethod>GET</AllowedMethod><AllowedOrigin>*</AllowedOrigin></CORSRule></CORSConfiguration>
@@ -1 +0,0 @@
-<CORSConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><CORSRule><ID>console</ID><AllowedOrigin>https://console.example.internal</AllowedOrigin><AllowedMethod>GET</AllowedMethod><AllowedMethod>HEAD</AllowedMethod><AllowedHeader>Authorization</AllowedHeader><AllowedHeader>Content-Type</AllowedHeader><ExposeHeader>ETag</ExposeHeader><MaxAgeSeconds>600</MaxAgeSeconds></CORSRule></CORSConfiguration>
@@ -1 +0,0 @@
-<CORSConfiguration><CORSRule><AllowedHeader>Authorization</AllowedHeader><AllowedHeader>Content-Type</AllowedHeader><AllowedMethod>GET</AllowedMethod><AllowedMethod>HEAD</AllowedMethod><AllowedOrigin>https://console.example.internal</AllowedOrigin><ExposeHeader>ETag</ExposeHeader><ID>console</ID><MaxAgeSeconds>600</MaxAgeSeconds></CORSRule></CORSConfiguration>
@@ -1 +0,0 @@
-<CORSConfiguration><CORSRule><AllowedHeader>Authorization</AllowedHeader><AllowedHeader>Content-Type</AllowedHeader><AllowedMethod>GET</AllowedMethod><AllowedMethod>PUT</AllowedMethod><AllowedOrigin>https://app.example.internal</AllowedOrigin><ExposeHeader>ETag</ExposeHeader><ExposeHeader>x-amz-request-id</ExposeHeader><ID>rustfs-app</ID><MaxAgeSeconds>3600</MaxAgeSeconds></CORSRule></CORSConfiguration>
@@ -1 +0,0 @@
-<CORSConfiguration><CORSRule><AllowedHeader>Authorization</AllowedHeader><AllowedHeader>Content-Type</AllowedHeader><AllowedMethod>GET</AllowedMethod><AllowedMethod>PUT</AllowedMethod><AllowedOrigin>https://app.example.internal</AllowedOrigin><ExposeHeader>ETag</ExposeHeader><ExposeHeader>x-amz-request-id</ExposeHeader><ID>rustfs-app</ID><MaxAgeSeconds>3600</MaxAgeSeconds></CORSRule></CORSConfiguration>
@@ -1,10 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<ServerSideEncryptionConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
-  <Rule>
-    <ApplyServerSideEncryptionByDefault>
-      <SSEAlgorithm>aws:kms</SSEAlgorithm>
-      <KMSMasterKeyID>arn:aws:kms:us-east-1:123456789012:key/1234abcd-12ab-34cd-56ef-1234567890ab</KMSMasterKeyID>
-    </ApplyServerSideEncryptionByDefault>
-    <BucketKeyEnabled>true</BucketKeyEnabled>
-  </Rule>
-</ServerSideEncryptionConfiguration>
@@ -1 +0,0 @@
-<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><KMSMasterKeyID>arn:aws:kms:us-east-1:123456789012:key/1234abcd-12ab-34cd-56ef-1234567890ab</KMSMasterKeyID><SSEAlgorithm>aws:kms</SSEAlgorithm></ApplyServerSideEncryptionByDefault><BucketKeyEnabled>true</BucketKeyEnabled></Rule></ServerSideEncryptionConfiguration>
@@ -1 +0,0 @@
-<ServerSideEncryptionConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm></ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>
@@ -1 +0,0 @@
-<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm></ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>
@@ -1 +0,0 @@
-<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><KMSMasterKeyID>rustfs-bucket-key</KMSMasterKeyID><SSEAlgorithm>aws:kms</SSEAlgorithm></ApplyServerSideEncryptionByDefault><BucketKeyEnabled>false</BucketKeyEnabled></Rule></ServerSideEncryptionConfiguration>
@@ -1 +0,0 @@
-<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><KMSMasterKeyID>rustfs-bucket-key</KMSMasterKeyID><SSEAlgorithm>aws:kms</SSEAlgorithm></ApplyServerSideEncryptionByDefault><BucketKeyEnabled>false</BucketKeyEnabled></Rule></ServerSideEncryptionConfiguration>
@@ -1,49 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<LifecycleConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
-  <Rule>
-    <ID>Archive and then delete rule</ID>
-    <Filter>
-      <And>
-        <Prefix>projectdocs/</Prefix>
-        <Tag>
-          <Key>department</Key>
-          <Value>finance</Value>
-        </Tag>
-        <ObjectSizeGreaterThan>1024</ObjectSizeGreaterThan>
-      </And>
-    </Filter>
-    <Status>Enabled</Status>
-    <Transition>
-      <Days>30</Days>
-      <StorageClass>STANDARD_IA</StorageClass>
-    </Transition>
-    <Transition>
-      <Days>365</Days>
-      <StorageClass>GLACIER</StorageClass>
-    </Transition>
-    <Expiration>
-      <Days>3650</Days>
-    </Expiration>
-    <NoncurrentVersionTransition>
-      <NoncurrentDays>90</NoncurrentDays>
-      <StorageClass>GLACIER</StorageClass>
-    </NoncurrentVersionTransition>
-    <NoncurrentVersionExpiration>
-      <NoncurrentDays>400</NoncurrentDays>
-      <NewerNoncurrentVersions>3</NewerNoncurrentVersions>
-    </NoncurrentVersionExpiration>
-    <AbortIncompleteMultipartUpload>
-      <DaysAfterInitiation>7</DaysAfterInitiation>
-    </AbortIncompleteMultipartUpload>
-  </Rule>
-  <Rule>
-    <ID>Expire delete markers</ID>
-    <Filter>
-      <Prefix>logs/</Prefix>
-    </Filter>
-    <Status>Disabled</Status>
-    <Expiration>
-      <ExpiredObjectDeleteMarker>true</ExpiredObjectDeleteMarker>
-    </Expiration>
-  </Rule>
-</LifecycleConfiguration>
@@ -1 +0,0 @@
-<LifecycleConfiguration><Rule><AbortIncompleteMultipartUpload><DaysAfterInitiation>7</DaysAfterInitiation></AbortIncompleteMultipartUpload><Expiration><Days>3650</Days></Expiration><Filter><And><ObjectSizeGreaterThan>1024</ObjectSizeGreaterThan><Prefix>projectdocs/</Prefix><Tag><Key>department</Key><Value>finance</Value></Tag></And></Filter><ID>Archive and then delete rule</ID><NoncurrentVersionExpiration><NewerNoncurrentVersions>3</NewerNoncurrentVersions><NoncurrentDays>400</NoncurrentDays></NoncurrentVersionExpiration><NoncurrentVersionTransition><NoncurrentDays>90</NoncurrentDays><StorageClass>GLACIER</StorageClass></NoncurrentVersionTransition><Status>Enabled</Status><Transition><Days>30</Days><StorageClass>STANDARD_IA</StorageClass></Transition><Transition><Days>365</Days><StorageClass>GLACIER</StorageClass></Transition></Rule><Rule><Expiration><ExpiredObjectDeleteMarker>true</ExpiredObjectDeleteMarker></Expiration><Filter><Prefix>logs/</Prefix></Filter><ID>Expire delete markers</ID><Status>Disabled</Status></Rule></LifecycleConfiguration>
@@ -1 +0,0 @@
-<LifecycleConfiguration><Rule><ID>d96i4g89k8h26a95st60</ID><Status>Enabled</Status><Filter><Prefix></Prefix></Filter><Expiration><Days>30</Days></Expiration><NoncurrentVersionExpiration><NoncurrentDays>7</NoncurrentDays></NoncurrentVersionExpiration></Rule><ExpiryUpdatedAt>2026-07-07T15:58:57.337315Z</ExpiryUpdatedAt></LifecycleConfiguration>
@@ -1 +0,0 @@
-<LifecycleConfiguration><ExpiryUpdatedAt>2026-07-07T15:58:57.337Z</ExpiryUpdatedAt><Rule><Expiration><Days>30</Days></Expiration><Filter><Prefix></Prefix></Filter><ID>d96i4g89k8h26a95st60</ID><NoncurrentVersionExpiration><NoncurrentDays>7</NoncurrentDays></NoncurrentVersionExpiration><Status>Enabled</Status></Rule></LifecycleConfiguration>
@@ -1 +0,0 @@
-<LifecycleConfiguration><Rule><AbortIncompleteMultipartUpload><DaysAfterInitiation>3</DaysAfterInitiation></AbortIncompleteMultipartUpload><DelMarkerExpiration><Days>7</Days></DelMarkerExpiration><Expiration><Days>30</Days></Expiration><Filter><Prefix>tmp/</Prefix></Filter><ID>expire-tmp</ID><NoncurrentVersionExpiration><NewerNoncurrentVersions>2</NewerNoncurrentVersions><NoncurrentDays>7</NoncurrentDays></NoncurrentVersionExpiration><Status>Enabled</Status></Rule><Rule><Filter><And><Prefix>archive/</Prefix><Tag><Key>tier</Key><Value>cold</Value></Tag></And></Filter><ID>tier-cold</ID><Status>Enabled</Status><Transition><Days>90</Days><StorageClass>COLDTIER</StorageClass></Transition></Rule></LifecycleConfiguration>
@@ -1 +0,0 @@
-<LifecycleConfiguration><Rule><AbortIncompleteMultipartUpload><DaysAfterInitiation>3</DaysAfterInitiation></AbortIncompleteMultipartUpload><DelMarkerExpiration><Days>7</Days></DelMarkerExpiration><Expiration><Days>30</Days></Expiration><Filter><Prefix>tmp/</Prefix></Filter><ID>expire-tmp</ID><NoncurrentVersionExpiration><NewerNoncurrentVersions>2</NewerNoncurrentVersions><NoncurrentDays>7</NoncurrentDays></NoncurrentVersionExpiration><Status>Enabled</Status></Rule><Rule><Filter><And><Prefix>archive/</Prefix><Tag><Key>tier</Key><Value>cold</Value></Tag></And></Filter><ID>tier-cold</ID><Status>Enabled</Status><Transition><Days>90</Days><StorageClass>COLDTIER</StorageClass></Transition></Rule></LifecycleConfiguration>
@@ -1,20 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<BucketLoggingStatus xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
-  <LoggingEnabled>
-    <TargetBucket>amzn-s3-demo-logging-bucket</TargetBucket>
-    <TargetPrefix>logs/</TargetPrefix>
-    <TargetGrants>
-      <Grant>
-        <Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="AmazonCustomerByEmail">
-          <EmailAddress>user@example.com</EmailAddress>
-        </Grantee>
-        <Permission>READ</Permission>
-      </Grant>
-    </TargetGrants>
-    <TargetObjectKeyFormat>
-      <PartitionedPrefix>
-        <PartitionDateSource>EventTime</PartitionDateSource>
-      </PartitionedPrefix>
-    </TargetObjectKeyFormat>
-  </LoggingEnabled>
-</BucketLoggingStatus>
@@ -1 +0,0 @@
-<BucketLoggingStatus><LoggingEnabled><TargetBucket>amzn-s3-demo-logging-bucket</TargetBucket><TargetGrants><Grant><Grantee xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:type="AmazonCustomerByEmail"><EmailAddress>user@example.com</EmailAddress></Grantee><Permission>READ</Permission></Grant></TargetGrants><TargetObjectKeyFormat><PartitionedPrefix><PartitionDateSource>EventTime</PartitionDateSource></PartitionedPrefix></TargetObjectKeyFormat><TargetPrefix>logs/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -1 +0,0 @@
-<BucketLoggingStatus xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><LoggingEnabled><TargetBucket>interop-logs</TargetBucket><TargetPrefix>interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -1 +0,0 @@
-<BucketLoggingStatus><LoggingEnabled><TargetBucket>interop-logs</TargetBucket><TargetPrefix>interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -1 +0,0 @@
-<BucketLoggingStatus><LoggingEnabled><TargetBucket>rustfs-access-logs</TargetBucket><TargetPrefix>buckets/interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -1 +0,0 @@
-<BucketLoggingStatus><LoggingEnabled><TargetBucket>rustfs-access-logs</TargetBucket><TargetPrefix>buckets/interop/</TargetPrefix></LoggingEnabled></BucketLoggingStatus>
@@ -1,30 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<NotificationConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
-  <TopicConfiguration>
-    <Id>ImageUploads</Id>
-    <Topic>arn:aws:sns:us-east-1:123456789012:s3-events</Topic>
-    <Event>s3:ObjectCreated:*</Event>
-    <Filter>
-      <S3Key>
-        <FilterRule>
-          <Name>prefix</Name>
-          <Value>images/</Value>
-        </FilterRule>
-        <FilterRule>
-          <Name>suffix</Name>
-          <Value>.jpg</Value>
-        </FilterRule>
-      </S3Key>
-    </Filter>
-  </TopicConfiguration>
-  <QueueConfiguration>
-    <Id>Deletions</Id>
-    <Queue>arn:aws:sqs:us-east-1:123456789012:s3-deletes</Queue>
-    <Event>s3:ObjectRemoved:*</Event>
-  </QueueConfiguration>
-  <CloudFunctionConfiguration>
-    <Id>Thumbnails</Id>
-    <CloudFunction>arn:aws:lambda:us-east-1:123456789012:function:thumbnail</CloudFunction>
-    <Event>s3:ObjectCreated:Put</Event>
-  </CloudFunctionConfiguration>
-</NotificationConfiguration>
@@ -1 +0,0 @@
-<NotificationConfiguration><CloudFunctionConfiguration><Event>s3:ObjectCreated:Put</Event><Id>Thumbnails</Id><CloudFunction>arn:aws:lambda:us-east-1:123456789012:function:thumbnail</CloudFunction></CloudFunctionConfiguration><QueueConfiguration><Event>s3:ObjectRemoved:*</Event><Id>Deletions</Id><Queue>arn:aws:sqs:us-east-1:123456789012:s3-deletes</Queue></QueueConfiguration><TopicConfiguration><Event>s3:ObjectCreated:*</Event><Filter><S3Key><FilterRule><Name>prefix</Name><Value>images/</Value></FilterRule><FilterRule><Name>suffix</Name><Value>.jpg</Value></FilterRule></S3Key></Filter><Id>ImageUploads</Id><Topic>arn:aws:sns:us-east-1:123456789012:s3-events</Topic></TopicConfiguration></NotificationConfiguration>
@@ -1 +0,0 @@
-<NotificationConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><QueueConfiguration><Id></Id><Event>s3:ObjectCreated:*</Event><Queue>arn:minio:sqs::primary:webhook</Queue></QueueConfiguration></NotificationConfiguration>
@@ -1 +0,0 @@
-<NotificationConfiguration><QueueConfiguration><Event>s3:ObjectCreated:*</Event><Id></Id><Queue>arn:minio:sqs::primary:webhook</Queue></QueueConfiguration></NotificationConfiguration>
@@ -1 +0,0 @@
-<NotificationConfiguration><QueueConfiguration><Event>s3:ObjectCreated:*</Event><Event>s3:ObjectRemoved:*</Event><Filter><S3Key><FilterRule><Name>prefix</Name><Value>uploads/</Value></FilterRule></S3Key></Filter><Id>rustfs-webhook</Id><Queue>arn:rustfs:sqs::primary:webhook</Queue></QueueConfiguration></NotificationConfiguration>
@@ -1 +0,0 @@
-<NotificationConfiguration><QueueConfiguration><Event>s3:ObjectCreated:*</Event><Event>s3:ObjectRemoved:*</Event><Filter><S3Key><FilterRule><Name>prefix</Name><Value>uploads/</Value></FilterRule></S3Key></Filter><Id>rustfs-webhook</Id><Queue>arn:rustfs:sqs::primary:webhook</Queue></QueueConfiguration></NotificationConfiguration>
@@ -1,10 +0,0 @@
-<?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>
@@ -1 +0,0 @@
-<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Mode>COMPLIANCE</Mode><Years>1</Years></DefaultRetention></Rule></ObjectLockConfiguration>
@@ -1 +0,0 @@
-<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Mode>GOVERNANCE</Mode><Days>7</Days></DefaultRetention></Rule></ObjectLockConfiguration>
@@ -1 +0,0 @@
-<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Days>7</Days><Mode>GOVERNANCE</Mode></DefaultRetention></Rule></ObjectLockConfiguration>
@@ -1 +0,0 @@
-<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Days>30</Days><Mode>GOVERNANCE</Mode></DefaultRetention></Rule></ObjectLockConfiguration>
@@ -1 +0,0 @@
-<ObjectLockConfiguration><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Days>30</Days><Mode>GOVERNANCE</Mode></DefaultRetention></Rule></ObjectLockConfiguration>
@@ -1,7 +0,0 @@
-<?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>
@@ -1 +0,0 @@
-<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><BlockPublicPolicy>true</BlockPublicPolicy><IgnorePublicAcls>true</IgnorePublicAcls><RestrictPublicBuckets>true</RestrictPublicBuckets></PublicAccessBlockConfiguration>
@@ -1 +0,0 @@
-<PublicAccessBlockConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><BlockPublicAcls>true</BlockPublicAcls><IgnorePublicAcls>false</IgnorePublicAcls><BlockPublicPolicy>true</BlockPublicPolicy><RestrictPublicBuckets>false</RestrictPublicBuckets></PublicAccessBlockConfiguration>
@@ -1 +0,0 @@
-<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><BlockPublicPolicy>true</BlockPublicPolicy><IgnorePublicAcls>false</IgnorePublicAcls><RestrictPublicBuckets>false</RestrictPublicBuckets></PublicAccessBlockConfiguration>
@@ -1 +0,0 @@
-<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><BlockPublicPolicy>false</BlockPublicPolicy><IgnorePublicAcls>true</IgnorePublicAcls><RestrictPublicBuckets>false</RestrictPublicBuckets></PublicAccessBlockConfiguration>
@@ -1 +0,0 @@
-<PublicAccessBlockConfiguration><BlockPublicAcls>true</BlockPublicAcls><BlockPublicPolicy>false</BlockPublicPolicy><IgnorePublicAcls>true</IgnorePublicAcls><RestrictPublicBuckets>false</RestrictPublicBuckets></PublicAccessBlockConfiguration>
@@ -1,12 +0,0 @@
-
-<ReplicationConfiguration>
-  <Role></Role>
-  <FutureTopLevel>future</FutureTopLevel>
-  <Rule>
-    <ID>unknown</ID>
-    <Status>Enabled</Status>
-    <Destination>
-      <Bucket>arn:aws:s3:::destination</Bucket>
-    </Destination>
-  </Rule>
-</ReplicationConfiguration>
@@ -1 +0,0 @@
-<ReplicationConfiguration><Role></Role><Rule><Destination><Bucket>arn:aws:s3:::destination</Bucket></Destination><ID>unknown</ID><Status>Enabled</Status></Rule></ReplicationConfiguration>
@@ -1 +0,0 @@
-<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>
@@ -1 +0,0 @@
-<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>
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
overtrue	5983594da9	feat(capture): add explicit schema-bound checkpoint capabilities	2026-10-11 21:28:05 +08:00
overtrue	dfa5d50565	feat(capture): persist first-decision provenance	2026-10-11 18:58:12 +08:00
overtrue	a902886737	feat(capture): atomically migrate legacy Created history	2026-10-11 16:39:10 +08:00
overtrue	8e53b87421	test(capture): specify atomic populated native migration	2026-10-11 16:39:10 +08:00
overtrue	f7631e91b1	feat(capture): stream an owned coherent snapshot view	2026-10-11 16:30:12 +08:00
overtrue	f9bc4e921f	test(capture): specify coherent snapshot view ownership	2026-10-11 16:30:12 +08:00
overtrue	781584307a	feat(capture): apply committed native entries on the ordered worker	2026-10-11 15:35:02 +08:00
overtrue	59d641a5fe	docs(capture): define recoverable checkpoint and purge contracts	2026-10-11 15:00:38 +08:00
overtrue	c618b57870	feat(capture): stream snapshot rows with a permanent failure fence	2026-10-11 14:44:05 +08:00
overtrue	bbe1d54098	feat(capture): add bounded streaming snapshot codec	2026-10-11 14:21:30 +08:00
overtrue	9e8ea4dcdb	feat(capture-authority): serialize native async writes	2026-10-11 13:35:15 +08:00
overtrue	63cf8e2f2f	feat(capture-authority): add native OpenRaft log reader	2026-10-11 13:01:43 +08:00
overtrue	b7fe40f4b3	feat(capture): support complete native log range bounds	2026-10-11 12:28:23 +08:00
overtrue	5b872b8e56	feat(capture): apply native entries atomically with durable replay	2026-10-11 11:50:32 +08:00
overtrue	ca5be1ef95	feat(capture): persist native Raft entries in explicitly enabled stores	2026-10-11 05:49:17 +08:00
overtrue	9373e8fa3e	Bind capture commands and responses to native Raft entries	2026-10-11 03:45:17 +08:00
overtrue	cb32965522	feat(capture): preserve exact raft values and membership codec	2026-10-11 02:04:47 +08:00
overtrue	fb6cf49a67	feat(capture): validate peer metadata and CNOD codec	2026-10-11 01:25:21 +08:00
overtrue	8c97049e68	build(capture): pin OpenRaft and guard resolved features	2026-10-11 00:56:30 +08:00
overtrue	17d4fe27cc	feat(capture): apply committed Created decisions atomically	2026-10-10 23:35:52 +08:00
overtrue	fb0c7972b2	feat(capture): persist committed log boundary	2026-10-10 23:10:28 +08:00
overtrue	eba7967755	feat(capture): persist atomic log suffix truncation	2026-10-10 22:59:26 +08:00
overtrue	451c1c31ad	feat(capture): add durable local authority storage	2026-10-10 22:33:44 +08:00
overtrue	8780dd670e	test(ecstore): verify publication across process kills	2026-10-10 20:19:46 +08:00
overtrue	f29bf2c902	fix(iam): preserve empty identity read failures	2026-10-10 11:37:07 +08:00
overtrue	fd03436090	feat(ecstore): expose unbounded preserve-empty config reads	2026-10-10 11:10:27 +08:00
overtrue	1f1e7b1c97	fix(iam): preserve identity decryption errors	2026-10-10 10:37:15 +08:00
overtrue	45a952609e	test(iam): add real identity storage fixtures	2026-10-10 10:01:03 +08:00
overtrue	323190a95a	test(test-utils): support caller-owned instance contexts	2026-10-10 09:43:01 +08:00
overtrue	38cdda9292	fix(ecstore): expose config reader without private bounds	2026-10-10 09:11:47 +08:00
overtrue	85c84fa73d	feat(policy): add dormant principal incarnation records	2026-10-10 08:12:14 +08:00
overtrue	51402087c5	feat(ecstore): expose bounded config snapshot reader	2026-10-10 08:03:27 +08:00
overtrue	a994984fe2	build(policy): prepare identity incarnation UUID dependency	2026-10-10 06:57:51 +08:00
overtrue	48eb0e402c	test(iam): use identity constructor in existing fixtures	2026-10-10 06:52:04 +08:00
GatewayJ	446479791e	refactor(iam): remove legacy OIDC federation paths (#8433 ) Co-authored-by: Hauser <housemecn@gmail.com>	2026-10-09 07:11:50 -07:00
Chris	99af59a914	Merge commit from fork	2026-10-09 21:43:53 +08:00
Chris	b77bd26874	Merge commit from fork	2026-10-09 21:43:08 +08:00
Chris	956c1cf027	Merge commit from fork	2026-10-09 21:40:59 +08:00
Chris	8b9d44110e	Merge commit from fork	2026-10-09 21:40:08 +08:00
Chris	44ce8d114f	Merge commit from fork	2026-10-09 21:39:18 +08:00
Chris	769b0d7578	Merge commit from fork	2026-10-09 21:28:19 +08:00
Chris	03e775947d	Merge commit from fork	2026-10-09 21:18:27 +08:00
Hauser	25e73d3bd6	fix(zip): statically link liblzma on macOS (#8434 ) * fix(zip): statically link liblzma on macOS Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(zip): pin macOS liblzma minimum to 0.4.8 Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(admin): route bucket recovery errors through facade Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(ecstore): group version pagination markers Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(heal): avoid cloning copy purge identity Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-10-09 16:25:24 +08:00
Hauser	a464b45367	fix(rustfs): keep recovery handler within S3 gateway (#8429 ) * fix(rustfs): route recovery errors through S3 gateway Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(ecstore): satisfy strict Clippy pagination checks Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(heal): avoid cloning Copy purge identity Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test: restore one-second drive request budgets * fix(ecstore): resume pending lock bucket creation Allow retries to reuse an uncommitted Object Lock bucket intent when the same incarnation remains unretired and all physical volumes are absent. Isolate the recovery entrypoint test from the shared production leader lock. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test: provision stack for deep catalog test futures Apply the existing 4 MiB test-thread stack setup to table-catalog handler/store tests and the relocated-pool GET regression in both local and CI nextest profiles. These cases overflowed libtest's 2 MiB default stack on Linux. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: zhi22915 <qiuzgang@gmail.com> Co-authored-by: overtrue <anzhengchao@gmail.com>	2026-10-09 14:08:12 +08:00
				`@@ -1 +0,0 @@`
				`<AccelerateConfiguration><Status>Enabled</Status></AccelerateConfiguration>`
				`@@ -1 +0,0 @@`
				`<AccelerateConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>Suspended</Status></AccelerateConfiguration>`
				`@@ -1 +0,0 @@`
				`<AccelerateConfiguration><Status>Suspended</Status></AccelerateConfiguration>`
				`@@ -1 +0,0 @@`
				`<CORSConfiguration><CORSRule><AllowedHeader></AllowedHeader><AllowedMethod>PUT</AllowedMethod><AllowedMethod>POST</AllowedMethod><AllowedMethod>DELETE</AllowedMethod><AllowedOrigin>https://www.example.com</AllowedOrigin><ExposeHeader>x-amz-server-side-encryption</ExposeHeader><ID>AllowWebApp</ID><MaxAgeSeconds>3000</MaxAgeSeconds></CORSRule><CORSRule><AllowedMethod>GET</AllowedMethod><AllowedOrigin></AllowedOrigin></CORSRule></CORSConfiguration>`
				`@@ -1 +0,0 @@`
				`<CORSConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><CORSRule><ID>console</ID><AllowedOrigin>https://console.example.internal</AllowedOrigin><AllowedMethod>GET</AllowedMethod><AllowedMethod>HEAD</AllowedMethod><AllowedHeader>Authorization</AllowedHeader><AllowedHeader>Content-Type</AllowedHeader><ExposeHeader>ETag</ExposeHeader><MaxAgeSeconds>600</MaxAgeSeconds></CORSRule></CORSConfiguration>`
				`@@ -1 +0,0 @@`
				`<ServerSideEncryptionConfiguration><Rule><ApplyServerSideEncryptionByDefault><KMSMasterKeyID>arn:aws:kms:us-east-1:123456789012:key/1234abcd-12ab-34cd-56ef-1234567890ab</KMSMasterKeyID><SSEAlgorithm>aws:kms</SSEAlgorithm></ApplyServerSideEncryptionByDefault><BucketKeyEnabled>true</BucketKeyEnabled></Rule></ServerSideEncryptionConfiguration>`
				`@@ -1 +0,0 @@`
				`<ServerSideEncryptionConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm></ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>`
				`@@ -1 +0,0 @@`
				<LifecycleConfiguration><Rule><AbortIncompleteMultipartUpload><DaysAfterInitiation>7</DaysAfterInitiation></AbortIncompleteMultipartUpload><Expiration><Days>3650</Days></Expiration><Filter><And><ObjectSizeGreaterThan>1024</ObjectSizeGreaterThan><Prefix>projectdocs/</Prefix><Tag><Key>department</Key><Value>finance</Value></Tag></And></Filter><ID>Archive and then delete rule</ID><NoncurrentVersionExpiration><NewerNoncurrentVersions>3</NewerNoncurrentVersions><NoncurrentDays>400</NoncurrentDays></NoncurrentVersionExpiration><NoncurrentVersionTransition><NoncurrentDays>90</NoncurrentDays><StorageClass>GLACIER</StorageClass></NoncurrentVersionTransition><Status>Enabled</Status><Transition><Days>30</Days><StorageClass>STANDARD_IA</StorageClass></Transition><Transition><Days>365</Days><StorageClass>GLACIER</StorageClass></Transition></Rule><Rule><Expiration><ExpiredObjectDeleteMarker>true</ExpiredObjectDeleteMarker></Expiration><Filter><Prefix>logs/</Prefix></Filter><ID>Expire delete markers</ID><Status>Disabled</Status></Rule></LifecycleConfiguration>
				`@@ -1 +0,0 @@`
				`<LifecycleConfiguration><Rule><ID>d96i4g89k8h26a95st60</ID><Status>Enabled</Status><Filter><Prefix></Prefix></Filter><Expiration><Days>30</Days></Expiration><NoncurrentVersionExpiration><NoncurrentDays>7</NoncurrentDays></NoncurrentVersionExpiration></Rule><ExpiryUpdatedAt>2026-07-07T15:58:57.337315Z</ExpiryUpdatedAt></LifecycleConfiguration>`