fix(lifecycle): restore #[serial] on 2 tests that read env vars without temp_env

eval_inner_expires_latest_object_after_days_due and eval_inner_does_not_panic_on_many_equal_due_events call eval_inner() which reads ENV_ILM_PROCESS_TIME via std::env::var(). Without #[serial] they race with other tests that set these vars via temp_env.
cleanup: remove #[serial] annotations from e2e_test, scanner, lifecycle, and object-capacity crates
2026-08-21 20:06:37 +00:00 · 2026-08-22 01:22:54 +08:00 · 2026-08-22 00:28:25 +08:00 · 2026-08-22 00:04:12 +08:00
68 changed files with 598 additions and 4399 deletions
@@ -1,2 +0,0 @@
-sha256-darwin=b4ae71aa894e5c7795ae3eb8116f1777a7601d0f5db3898be2e48faf3329bd9b
-sha256-linux=433debd9d9defa832986269abdf0f1d131597b2d7a417ce930e17c1fd47d85ba
@@ -1 +0,0 @@
-sha256=9b9bc336b43b70d0e06e0adb5455bf035bb18945d85d60936eb6fe4d48e0e680
@@ -1,2 +0,0 @@
-sha256-darwin=55534a97fbd376f64c8f6c341d319017d11ff77cad6da8629a1a7f6a874e0315
-sha256-linux=c06fb8c19aed6f388b9dc61cb8251b7a44f8561a9bf764ad2b9e635598f8dc17
@@ -1 +0,0 @@
-sha256=655a3f3c1d042e694339d15caba7580518320322d1bac0f09450b37e6c09e2e7
@@ -1 +0,0 @@
-sha256=ec27cde6ce6400723c4b372bfbd2ac61709c744294e4810af765e8a808d8e31d
@@ -75,11 +75,6 @@ embedded-secrets-check: ## Check no private key material or credential literal i
 	@echo "🔑 Checking embedded secret material guard..."
 	./scripts/check_embedded_secrets.sh

-.PHONY: test-wiring-check
-test-wiring-check: ## Check tests stay registered and selected by their intended runners
-	@echo "🧪 Checking test wiring..."
-	python3 ./scripts/check_test_wiring.py
-
 .PHONY: log-analyzer-rules-check
 log-analyzer-rules-check: core-deps ## Check log-analyzer rule anchors still exist verbatim in source
 	@echo "🩺 Checking log-analyzer rule anchors..."
@@ -19,13 +19,13 @@ 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 tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check embedded-secrets-check test-wiring-check doc-paths-check planning-docs-check quick-check ## Run fast pre-commit checks without clippy/full tests
+pre-commit: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check embedded-secrets-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 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 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 tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check embedded-secrets-check doc-paths-check planning-docs-check log-analyzer-rules-check clippy-check test ## Run full pre-PR checks with clippy and tests
 	@echo "✅ All pre-PR checks passed!"

 .PHONY: dev-check
-dev-check: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-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 local development checks
+dev-check: fmt-check unsafe-code-check architecture-migration-check logging-guardrails-check tokio-io-uring-check extension-schema-check body-cache-whitelist-check s3s-footprint-check fips-wording-check embedded-secrets-check doc-paths-check planning-docs-check quick-check ## Run fast local development checks
 	@echo "✅ Fast development checks passed!"
@@ -35,8 +35,6 @@ script-tests: ## Run shell script tests
 	./scripts/test_pinned_paired_abba_bench.sh
 	./scripts/test_manual_transition_runbooks.sh
 	./scripts/check_embedded_secrets.sh --self-test
-	python3 ./scripts/check_test_wiring.py --self-test
-	python3 ./scripts/s3-tests/test_report_compat.py
 	bash -n ./scripts/validate_object_data_cache_cold_stampede.sh
 	python3 ./scripts/check_object_data_cache_follower_samples.py --self-test
 	./scripts/validate_object_data_cache_cold_stampede.sh --self-test
@@ -38,11 +38,10 @@ e2e-vault = { max-threads = 1 }
 # replacement_privileged_e2e_test when explicitly run as root on Linux). They
 # are correct in isolation but resource-heavy; serialize them under nextest's
 # process boundary (serial_test's #[serial] does not cross it) so several 4-disk
-# servers never run at once. The e2e-full merge/main lane picks these up;
+# servers never run at once. ci-7's nightly picks these up via the e2e suite;
 # they are deliberately NOT in the fast PR `e2e-smoke` filter.
 e2e-reliability = { max-threads = 1 }
 e2e-inline-boundaries = { max-threads = 1 }
-e2e-cluster-nightly = { max-threads = 1 }

 # --- default profile (local): serialize the flaky groups, never retry --------
 [[profile.default.overrides]]
@@ -162,7 +161,7 @@ retries = 2
 # Serialize the 4-disk reliability / degraded-read e2e tests under the ci
 # profile too (see the e2e-reliability test-group note near the top). Not a
 # quarantine: no retries, just single-threaded so several 4-disk servers never
-# run concurrently when e2e-full runs the suite.
+# run concurrently when ci-7's nightly runs the full e2e suite.
 [[profile.ci.overrides]]
 filter = 'package(e2e_test) & test(/^(reliability_disk_fault|degraded_read_eof_regression|replacement_privileged_e2e)_test::/)'
 test-group = 'e2e-reliability'
@@ -231,8 +230,8 @@ test-group = 'ecstore-serial-flaky'
 # the nightly profile derives its set as "the replication module MINUS this
 # allowlist", so any new replication test lands in nightly by default (never
 # silently unrun) until it is explicitly blessed as fast here. Keep the two
-# regexes byte-identical. The committed profile selection digests make changes
-# visible in CI; current counts live in docs/testing/e2e-suite-inventory.md.
+# regexes byte-identical. Count invariant: 20 here + 49 nightly = 69 total
+# (authority: `cargo nextest list`; docs/testing/e2e-suite-inventory.md).
 # HISTORY (2026-07-11): the 20 fast tests were briefly pulled out of this lane
 # (#4724) because they set a loopback (127.0.0.1) replication target that the
 # SSRF egress guard rejected on every PR after repl-1 (#4712). That is fixed —
@@ -328,8 +327,9 @@ slow-timeout = { period = "60s", terminate-after = 2, grace-period = "10s" }
 # the STS dual-node test actually exercises its path (it skips gracefully with
 # a visible log line when awscurl is absent), and routes scheduled failures
 # through .github/actions/schedule-failure-issue (ci-8). Explicit division of
-# labor with e2e-full: these tests run only in the consolidated nightly
-# workflow, not in the merge/main lane.
+# labor with ci-5's future e2e-full merge gate: these tests run ONLY here, not
+# double-run there. TODO(ci-7): fold this interim repl-owned lane into the ci
+# domain's consolidated scheduled e2e workflow once it exists.
 [profile.e2e-repl-nightly]
 default-filter = """
    package(e2e_test)
@@ -343,60 +343,26 @@ fail-fast = false
 # workflow as the failure-triage artifact.
 path = "junit.xml"

-# ---------------------------------------------------------------------------
-# e2e-nightly profile — destructive multi-process cluster fault domains
-# ---------------------------------------------------------------------------
-# These seven modules are deliberately outside e2e-full's merge budget. Each
-# starts a real multi-process or multi-disk topology and exercises node/disk
-# loss, quorum, cleanup, notification fan-in, or admin-timeout behavior. The
-# consolidated nightly workflow runs them serially to avoid resource
-# starvation; failures are never retried.
-[profile.e2e-nightly]
-default-filter = """
-    package(e2e_test)
-    & test(/^(admin_timeout_regression_test|cluster_concurrency_test|cluster_multidrive_pool_test|heal_erasure_disk_rebuild_test|namespace_lock_quorum_test|object_lambda_test|stale_multipart_cleanup_cluster_test)::/)
-"""
-fail-fast = false
-
-[profile.e2e-nightly.junit]
-path = "junit.xml"
-
-[[profile.e2e-nightly.overrides]]
-filter = 'package(e2e_test)'
-test-group = 'e2e-cluster-nightly'
-
-# ---------------------------------------------------------------------------
-# e2e-protocols profile — serial protocol lane
-# ---------------------------------------------------------------------------
-# The suite owns fixed ports, so the nightly workflow runs this exact profile
-# with one nextest worker.
-[profile.e2e-protocols]
-default-filter = 'package(e2e_test) & test(/^protocols::/)'
-fail-fast = false
-
-[profile.e2e-protocols.junit]
-path = "junit.xml"
-
 # ---------------------------------------------------------------------------
 # e2e-full profile — merge-gate full single-node e2e lane (backlog#1149 ci-5)
 # ---------------------------------------------------------------------------
 # The merge gate (ci.yml `e2e-full` job: push main + merge_group +
-# workflow_dispatch). Runs the user-visible KMS, object-lock, multipart-auth,
-# quota, checksum, encryption,
+# workflow_dispatch). Runs the never-automated user-visible suites — KMS (40),
+# object_lock (33), multipart_auth (109), quota, checksum, encryption,
 # security-boundary, ... — that the fast PR `e2e-smoke` subset deliberately
 # skips. Budget <= 45 min; authority for the suite count is `cargo nextest list
 # --profile e2e-full` (see docs/testing/e2e-suite-inventory.md).
 #
 # The filter is "the whole e2e_test crate MINUS the sets owned by other lanes":
-#   * protocols:: — FTPS/SFTP/WebDAV, run from the dedicated protocol profile
-#     with one worker because the suite owns fixed ports.
+#   * protocols:: — FTPS/SFTP/WebDAV, still pinned to --test-threads=1 by fixed
+#     ports; they join a scheduled lane once ci-6 randomises the ports (ci-7).
 #   * the 7 cluster suites that spin up a RustFSTestClusterEnvironment
 #     (cluster_concurrency, cluster_multidrive_pool, stale_multipart_cleanup_cluster,
 #     namespace_lock_quorum, heal_erasure_disk_rebuild, admin_timeout_regression,
-#     object_lambda) — too heavy for the merge budget; they run in the
-#     e2e-nightly serial cluster-fault lane.
+#     object_lambda) — too heavy for the merge budget; they run in ci-7's
+#     nightly 4-node lane.
 #   * replication_extension_test — repl-1 already splits it into the PR
-#     `e2e-smoke` (20 fast) and `e2e-repl-nightly` (55 slow) lanes and reserves
+#     `e2e-smoke` (20 fast) and `e2e-repl-nightly` (49 slow) lanes and reserves
 #     it for those, so e2e-full does not double-run it.
 #   * #[ignore]d tests — nextest skips them by default (no --run-ignored); the
 #     manual-localhost:9000 reliant/policy tests are ci-13's migration.
@@ -46,9 +46,10 @@ lists when upstream changes.
  the PR.
 - **Weekly + manual**: `.github/workflows/e2e-s3tests.yml` runs the full
  upstream suite (`TEST_SCOPE=all`) against a Docker deployment (single node
-  or a 4-node distributed cluster behind HAProxy). The canonical gate policy
-  and compatibility-report behavior are documented in
-  [`scripts/s3-tests/README.md`](../../scripts/s3-tests/README.md).
+  or a 4-node distributed cluster behind HAProxy). It fails only on
+  regressions in the implemented whitelist and publishes a classification
+  report (`compat-report.md`, also shown in the job summary) listing promotion
+  candidates and unclassified tests.

 ## Running Tests Locally

@@ -125,9 +125,6 @@ jobs:
      - name: Check no embedded secret material
        run: ./scripts/check_embedded_secrets.sh

-      - name: Check test wiring
-        run: python3 ./scripts/check_test_wiring.py
-
      - name: Check no planning docs committed
        run: ./scripts/check_no_planning_docs.sh

@@ -160,9 +160,6 @@ jobs:
      - name: Check no embedded secret material
        run: ./scripts/check_embedded_secrets.sh

-      - name: Check test wiring
-        run: python3 ./scripts/check_test_wiring.py
-
      - name: Check no planning docs committed
        run: ./scripts/check_no_planning_docs.sh

@@ -689,9 +686,9 @@ jobs:
      - name: Make binary executable
        run: chmod +x ./target/debug/rustfs

-      # Build the e2e test graph once. The archive is reused by the smoke
-      # selection guard, security exact-count check, and run below, avoiding a
-      # second compile of the same e2e_test target on cold runners (backlog#1645).
+      # Build the e2e test graph once. The archive is reused by the security
+      # count-floor check and the smoke run below, avoiding a second compile of
+      # the same e2e_test target on cold runners (backlog#1645).
      - name: Archive e2e smoke test binaries
        env:
          NEXTEST_ARCHIVE: ${{ runner.temp }}/rustfs-e2e-smoke.tar.zst
@@ -699,7 +696,6 @@ jobs:
        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}"

      # PR smoke subset of the in-repo e2e suite (backlog#1149 ci-4). The
@@ -764,7 +760,7 @@ jobs:
    # suites — KMS, object_lock, multipart_auth, quota, checksum, encryption,
    # security-boundary, ... — via the e2e-full nextest profile. Too heavy for
    # every PR, so it is gated to main pushes, the merge queue, and manual
-    # dispatch. protocols / the 7 cluster suites / replication / #[ignore] are
+    # dispatch. protocols / the 6 cluster suites / replication / #[ignore] are
    # owned by other lanes (see .config/nextest.toml profile.e2e-full).
    if: >-
      github.event_name == 'workflow_dispatch' ||
@@ -824,13 +820,6 @@ jobs:
      - name: Make binary executable
        run: chmod +x ./target/debug/rustfs

-      - name: Verify e2e full membership
-        env:
-          NEXTEST_LISTING: ${{ runner.temp }}/rustfs-e2e-full-list.json
-        run: |
-          cargo nextest list --profile e2e-full -p e2e_test --message-format json > "${NEXTEST_LISTING}"
-          python3 ./scripts/check_test_wiring.py --check-profile e2e-full "${NEXTEST_LISTING}"
-
      # Full single-node e2e lane (backlog#1149 ci-5). The e2e-full
      # default-filter in .config/nextest.toml is the single wiring mechanism —
      # extend that filter, never add ad-hoc e2e jobs here. Reuses the downloaded
@@ -843,9 +832,7 @@ jobs:
        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
        with:
          name: e2e-full-junit-${{ github.run_number }}
-          path: |
-            target/nextest/e2e-full/junit.xml
-            ${{ runner.temp }}/rustfs-e2e-full-list.json
+          path: target/nextest/e2e-full/junit.xml
          retention-days: 7

  e2e-tests-rio-v2:
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

-# Consolidated nightly e2e lane for replication, cluster faults, and protocols.
+# Nightly full replication e2e lane (backlog#1147 repl-1, deps: ci-4).
 #
 # The per-PR gate (ci.yml `e2e-tests` job, `--profile e2e-smoke`) runs the
 # FAST replication tests. This scheduled lane runs the remaining heavier
@@ -28,12 +28,15 @@
 # add ad-hoc cargo-test steps here; change the filterset instead. The
 # authoritative membership and count come from
 # `cargo nextest list -p e2e_test --profile e2e-repl-nightly`; the PR/nightly
-# selection digest is committed under .config/.
+# count invariant is maintained next to the filtersets in .config/nextest.toml
+# (deliberately not duplicated here).
 #
-# Explicit division of labor: these subsets run only here and never double-run
-# in the e2e-full merge gate.
+# Explicit division of labor: the nightly subset runs ONLY here, never double-run
+# in ci-5's future e2e-full merge gate. TODO(ci-7): once the ci domain's
+# consolidated scheduled e2e workflow exists, fold this interim repl-owned lane
+# into it rather than growing a second scheduled entrypoint.

-name: e2e-nightly
+name: e2e-replication-nightly

 on:
  workflow_dispatch:
@@ -47,10 +50,6 @@ on:
 permissions:
  contents: read

-concurrency:
-  group: ${{ github.workflow }}-${{ github.ref }}
-  cancel-in-progress: false
-
 jobs:
  repl-nightly:
    name: Replication e2e (nightly)
@@ -98,20 +97,9 @@ jobs:
      # demand otherwise, but a single explicit build avoids several parallel
      # nextest test processes racing to build it at once.
      - name: Build rustfs binary
-        run: |
-          cargo build -p rustfs --bins
-          : > target/debug/rustfs.features
-
-      - name: Verify replication e2e membership
-        env:
-          NEXTEST_LISTING: ${{ runner.temp }}/rustfs-e2e-repl-nightly-list.json
-        run: |
-          cargo nextest list --profile e2e-repl-nightly -p e2e_test --message-format json > "${NEXTEST_LISTING}"
-          python3 ./scripts/check_test_wiring.py --check-profile e2e-repl-nightly "${NEXTEST_LISTING}"
+        run: cargo build -p rustfs --bins

      - name: Run replication e2e nightly suite
-        env:
-          RUSTFS_E2E_LOG_DIR: ${{ runner.temp }}/rustfs-e2e-repl-nightly-logs
        run: cargo nextest run --profile e2e-repl-nightly -p e2e_test

      - name: Upload nextest junit report
@@ -119,112 +107,13 @@ jobs:
        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
        with:
          name: e2e-replication-nightly-junit-${{ github.run_number }}
-          path: |
-            target/nextest/e2e-repl-nightly/junit.xml
-            ${{ runner.temp }}/rustfs-e2e-repl-nightly-list.json
-            ${{ runner.temp }}/rustfs-e2e-repl-nightly-logs/
+          path: target/nextest/e2e-repl-nightly/junit.xml
          retention-days: 7
          if-no-files-found: ignore

-  cluster-nightly:
-    name: Cluster fault e2e (nightly)
-    runs-on: sm-standard-4
-    timeout-minutes: 90
-    env:
-      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
-        with:
-          persist-credentials: false
-
-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
-        with:
-          rust-version: stable
-          cache-shared-key: ci-e2e-nightly
-          cache-save-if: 'false'
-          install-build-packaging-tools: 'false'
-
-      - name: Build rustfs binary
-        run: |
-          cargo build -p rustfs --bins --features e2e-test-hooks
-          : > target/debug/rustfs.features
-
-      - name: Verify cluster fault e2e membership
-        env:
-          NEXTEST_LISTING: ${{ runner.temp }}/rustfs-e2e-nightly-list.json
-        run: |
-          cargo nextest list --profile e2e-nightly -p e2e_test --message-format json > "${NEXTEST_LISTING}"
-          python3 ./scripts/check_test_wiring.py --check-profile e2e-nightly "${NEXTEST_LISTING}"
-
-      - name: Run cluster fault e2e nightly suite
-        env:
-          RUSTFS_E2E_LOG_DIR: ${{ runner.temp }}/rustfs-e2e-nightly-logs
-        run: cargo nextest run --profile e2e-nightly -p e2e_test
-
-      - name: Upload cluster fault diagnostics
-        if: always()
-        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
-        with:
-          name: e2e-cluster-nightly-${{ github.run_number }}
-          path: |
-            target/nextest/e2e-nightly/junit.xml
-            ${{ runner.temp }}/rustfs-e2e-nightly-list.json
-            ${{ runner.temp }}/rustfs-e2e-nightly-logs/
-          retention-days: 7
-          if-no-files-found: warn
-
-  protocols-nightly:
-    name: Protocol e2e (nightly)
-    runs-on: sm-standard-4
-    timeout-minutes: 90
-    env:
-      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
-      RUSTFS_BUILD_FEATURES: ftps,webdav,sftp
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
-        with:
-          persist-credentials: false
-
-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
-        with:
-          rust-version: stable
-          cache-shared-key: ci-e2e-protocols
-          cache-save-if: 'false'
-          install-build-packaging-tools: 'false'
-
-      # The suite owns fixed protocol ports and serializes its internal cases.
-      - name: Verify protocol e2e membership
-        env:
-          NEXTEST_LISTING: ${{ runner.temp }}/rustfs-e2e-protocols-list.json
-        run: |
-          cargo nextest list --profile e2e-protocols -p e2e_test --message-format json > "${NEXTEST_LISTING}"
-          python3 ./scripts/check_test_wiring.py --check-profile e2e-protocols "${NEXTEST_LISTING}"
-
-      - name: Run protocol e2e nightly suite
-        env:
-          RUSTFS_E2E_LOG_DIR: ${{ runner.temp }}/rustfs-protocol-e2e-logs
-        run: >-
-          cargo nextest run -j 1 --profile e2e-protocols -p e2e_test --no-capture
-
-      - name: Upload protocol diagnostics
-        if: always()
-        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
-        with:
-          name: e2e-protocol-nightly-${{ github.run_number }}
-          path: |
-            target/nextest/e2e-protocols/junit.xml
-            ${{ runner.temp }}/rustfs-e2e-protocols-list.json
-            ${{ runner.temp }}/rustfs-protocol-e2e-logs/
-          retention-days: 7
-          if-no-files-found: warn
-
  alert-on-failure:
    name: Alert on scheduled failure
-    needs: [repl-nightly, cluster-nightly, protocols-nightly]
+    needs: [repl-nightly]
    # Only scheduled runs open/append the tracking issue (backlog#1149 ci-8);
    # manual workflow_dispatch runs stay quiet so a debugging run never files a
    # spurious alert.
@@ -18,9 +18,10 @@
 # runs only the implemented_tests.txt whitelist. This workflow complements it:
 #
 #   - Scheduled weekly full sweep (TEST_SCOPE=all): runs the ENTIRE upstream
-#     suite and reports promotion candidates. Regressions, unclassified tests,
-#     incomplete execution, and infrastructure errors fail the job; classified
-#     failures for not-yet-implemented features remain informational.
+#     suite and reports promotion candidates (tests that newly pass) and
+#     unclassified tests. The job fails only on regressions in the implemented
+#     whitelist or on infrastructure errors — expected failures from
+#     not-yet-implemented features do not turn the run red.
 #   - Manual runs (workflow_dispatch): same, with configurable mode/scope.
 #
 # All test execution is delegated to scripts/s3-tests/run.sh (single source of
@@ -44,6 +45,13 @@
 # The PR gate (ci.yml s3-implemented-tests) is unaffected: it avoids Docker
 # via DEPLOY_MODE=binary and defers all pip setup to run.sh's self-bootstrap.

+# DISABLED. This workflow is switched off in the repository's Actions settings
+# (state: disabled_manually) and does not run on any trigger, including its cron
+# and workflow_dispatch. That state lives in GitHub's UI and is invisible when
+# reading this file, which has already misled at least one audit — hence this
+# banner. Re-enabling is a UI action; anyone doing so should first check that the
+# workflow still matches the current CI layout. See rustfs/backlog#1603.
+#
 name: e2e-s3tests

 on:
@@ -73,19 +81,6 @@ on:
        description: "Stop after N failures. '0' to run everything."
        required: false
        default: "0"
-      shard-count:
-        description: "Exact-node-ID shard count for a targeted manual run"
-        required: false
-        default: "1"
-        type: choice
-        options:
-          - "1"
-          - "2"
-          - "4"
-      shard-index:
-        description: "Zero-based shard index for a targeted manual run"
-        required: false
-        default: "0"
      markexpr:
        description: "Optional pytest -m expression"
        required: false
@@ -116,8 +111,6 @@ env:
  XDIST: ${{ github.event.inputs.xdist || '4' }}
  MAXFAIL: ${{ github.event.inputs.maxfail || '0' }}
  MARKEXPR: ${{ github.event.inputs.markexpr || '' }}
-  S3_SHARD_COUNT: ${{ github.event_name == 'schedule' && '4' || github.event.inputs.shard-count || '1' }}
-  TEST_TIMEOUT: "300"

 concurrency:
  group: ${{ github.workflow }}-${{ github.ref }}-${{ github.event.inputs['test-mode'] || 'single' }}
@@ -134,22 +127,19 @@ defaults:

 jobs:
  s3tests:
-    name: s3tests (${{ matrix.test-mode }}, shard ${{ matrix.shard-index }})
    # GitHub-hosted: reliably provides Docker + docker compose + python3/pip.
    # See the header note (ci-1) for why the self-hosted sm-standard-4 label
-    # was abandoned. Scheduled failures are handled by alert-on-failure below.
+    # was abandoned. TODO(ci-8): scheduled-failure alerting (auto-open issue)
+    # is added by the ci-8 composite action; do not implement it here.
    runs-on: ubuntu-latest
    timeout-minutes: 180
    strategy:
      fail-fast: false
-      max-parallel: 2
      matrix:
        # Scheduled sweeps cover both topologies; manual runs use the input.
        test-mode: ${{ github.event_name == 'schedule' && fromJSON('["single", "multi"]') || fromJSON(format('["{0}"]', github.event.inputs.test-mode || 'single')) }}
-        shard-index: ${{ github.event_name == 'schedule' && fromJSON('[0, 1, 2, 3]') || fromJSON(format('[{0}]', github.event.inputs.shard-index || '0')) }}
    env:
      TEST_MODE: ${{ matrix.test-mode }}
-      S3_SHARD_INDEX: ${{ matrix.shard-index }}
    steps:
      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
        with:
@@ -191,7 +181,6 @@ jobs:
      - name: Start single RustFS
        if: env.TEST_MODE == 'single'
        run: |
-          SSE_KEY="$(head -c 32 /dev/zero | base64 -w0)"
          docker network inspect rustfs-net >/dev/null 2>&1 || docker network create rustfs-net
          docker rm -f rustfs-single >/dev/null 2>&1 || true
          # The four disks share one physical device on the runner (a single
@@ -204,7 +193,6 @@ jobs:
            -e RUSTFS_ADDRESS=0.0.0.0:9000 \
            -e RUSTFS_ACCESS_KEY="${S3_ACCESS_KEY}" \
            -e RUSTFS_SECRET_KEY="${S3_SECRET_KEY}" \
-            -e RUSTFS_SSE_S3_MASTER_KEY="${SSE_KEY}" \
            -e RUSTFS_VOLUMES="/data/rustfs{0...3}" \
            -e RUSTFS_UNSAFE_BYPASS_DISK_CHECK=true \
            -v /tmp/rustfs-single:/data \
@@ -213,7 +201,6 @@ jobs:
      - name: Start 4-node distributed cluster
        if: env.TEST_MODE == 'multi'
        run: |
-          SSE_KEY="$(head -c 32 /dev/zero | base64 -w0)"
          # A real distributed deployment: every node lists all endpoints in
          # RUSTFS_VOLUMES so data is erasure-coded ACROSS nodes. Do not use
          # node-local volume paths here — that would create four independent
@@ -226,7 +213,6 @@ jobs:
              RUSTFS_ADDRESS: "0.0.0.0:9000"
              RUSTFS_ACCESS_KEY: ${S3_ACCESS_KEY}
              RUSTFS_SECRET_KEY: ${S3_SECRET_KEY}
-              RUSTFS_SSE_S3_MASTER_KEY: "${SSE_KEY}"
              RUSTFS_VOLUMES: "http://rustfs{1...4}:9000/data/rustfs{0...3}"
              # Each node's four disks share one physical device inside its
              # container, so bypass the local physical-disk-independence guard
@@ -308,6 +294,7 @@ jobs:

      - name: Run ceph s3-tests
        run: |
+          set +e
          DEPLOY_MODE=existing \
          TEST_MODE="${TEST_MODE}" \
          TEST_SCOPE="${TEST_SCOPE}" \
@@ -315,6 +302,26 @@ jobs:
          MAXFAIL="${MAXFAIL}" \
          MARKEXPR="${MARKEXPR}" \
          ./scripts/s3-tests/run.sh
+          RC=$?
+          set -e
+
+          if [ "${TEST_SCOPE}" = "implemented" ]; then
+            # Whitelist run: every failure is a regression.
+            exit "${RC}"
+          fi
+
+          # Full sweep: failures outside the implemented whitelist are
+          # inventory (promotion candidates / unimplemented features), not a
+          # gate. Fail only on whitelist regressions or infrastructure errors.
+          JUNIT="artifacts/s3tests-${TEST_MODE}/junit.xml"
+          if [ ! -f "${JUNIT}" ]; then
+            echo "No junit.xml produced — infrastructure failure (exit ${RC})" >&2
+            exit "${RC}"
+          fi
+          python3 scripts/s3-tests/report_compat.py \
+            --junit "${JUNIT}" \
+            --lists-dir scripts/s3-tests \
+            --fail-on-regression

      - name: Publish compatibility report
        if: always()
@@ -339,7 +346,7 @@ jobs:
        if: always() && env.ACT != 'true'
        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
        with:
-          name: s3tests-${{ env.TEST_MODE }}-shard-${{ matrix.shard-index }}
+          name: s3tests-${{ env.TEST_MODE }}
          path: artifacts/**

  alert-on-failure:
@@ -12,22 +12,27 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

+# DISABLED. This workflow is switched off in the repository's Actions settings
+# (state: disabled_manually) and does not run on any trigger, including its cron
+# and workflow_dispatch. That state lives in GitHub's UI and is invisible when
+# reading this file, which has already misled at least one audit — hence this
+# banner. Re-enabling is a UI action; anyone doing so should first check that the
+# workflow still matches the current CI layout. See rustfs/backlog#1603.
+#
 name: Fuzz

 on:
  pull_request:
    types: [ opened, synchronize, reopened, closed ]
-    # Run when the harness or any directly fuzzed production crate changes.
+    # PR trigger is intentionally narrow: only changes to the fuzz harness
+    # itself gate a PR. Broad crate paths (ecstore/filemeta/utils/policy/…)
+    # are covered by the nightly `schedule` run below, which fuzzes against
+    # whatever landed on main. Widening these paths previously queued a
+    # ~45min fuzz-build on nearly every PR and is why this workflow was
+    # disabled; do not re-add crate paths here.
    paths:
      - "fuzz/**"
      - "scripts/fuzz/**"
-      - "crates/ecstore/**"
-      - "crates/filemeta/**"
-      - "crates/policy/**"
-      - "crates/security-governance/**"
-      - "crates/utils/**"
-      - "Cargo.toml"
-      - "Cargo.lock"
      - ".github/workflows/fuzz.yml"
  schedule:
    - cron: "0 2 * * *"
@@ -76,7 +81,7 @@ jobs:
      github.event_name == 'schedule' ||
      github.event_name == 'workflow_dispatch'
    runs-on: sm-standard-4
-    timeout-minutes: 60
+    timeout-minutes: 45
    env:
      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
    steps:
@@ -116,7 +121,12 @@ jobs:
        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
        with:
          name: fuzz-prebuilt-binaries-${{ github.run_number }}
-          path: fuzz/prebuilt/${{ env.CARGO_BUILD_TARGET }}/release/
+          path: |
+            fuzz/prebuilt/${{ env.CARGO_BUILD_TARGET }}/release/archive_extract
+            fuzz/prebuilt/${{ env.CARGO_BUILD_TARGET }}/release/bucket_validation
+            fuzz/prebuilt/${{ env.CARGO_BUILD_TARGET }}/release/local_metadata
+            fuzz/prebuilt/${{ env.CARGO_BUILD_TARGET }}/release/path_containment
+            fuzz/prebuilt/${{ env.CARGO_BUILD_TARGET }}/release/policy_ingress
          if-no-files-found: error
          retention-days: 1
          compression-level: 0
@@ -182,7 +192,10 @@ jobs:
  nightly-fuzz-corpus:
    name: "Nightly / ${{ matrix.target }}"
    needs: fuzz-build
-    # Scheduled failures are handled by alert-on-failure below.
+    # TODO(ci-8): when the schedule-failure-issue composite action lands,
+    # add a step here (or a dependent job) that opens/updates a GitHub issue
+    # on nightly failure. ci-8 is the single alerting mechanism for all
+    # scheduled workflows; do not self-roll alerting in this workflow.
    if: >
      github.event_name == 'schedule' ||
      (github.event_name == 'workflow_dispatch' &&
@@ -3843,7 +3843,6 @@ dependencies = [
 "s3s",
 "serde",
 "serde_json",
- "serial_test",
 "sha2 0.11.0",
 "suppaftp",
 "time",
@@ -9804,7 +9803,6 @@ dependencies = [
 "rustfs-replication",
 "rustfs-storage-api",
 "s3s",
- "serial_test",
 "temp-env",
 "time",
 "tokio",
@@ -9920,7 +9918,6 @@ dependencies = [
 "rustfs-config",
 "rustfs-io-metrics",
 "rustfs-utils",
- "serial_test",
 "temp-env",
 "tempfile",
 "tokio",
@@ -10293,7 +10290,6 @@ dependencies = [
 "s3s",
 "serde",
 "serde_json",
- "serial_test",
 "sha2 0.11.0",
 "temp-env",
 "tempfile",
@@ -96,7 +96,6 @@ tokio-stream = { workspace = true }
 rustfs-madmin.workspace = true
 rustfs-filemeta.workspace = true
 bytes = { workspace = true, features = ["serde"] }
-serial_test = { workspace = true }
 aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] }
 aws-sdk-sts = { workspace = true, default-features = false, features = ["default-https-client", "rt-tokio"] }
 aws-config = { workspace = true }
@@ -48,14 +48,16 @@ cargo nextest run --profile e2e-smoke -p e2e_test
 cargo nextest run -j1 --run-ignored ignored-only -p rustfs-scanner -p rustfs \
  -E 'binary(lifecycle_integration_test) or (package(rustfs) and test(lifecycle_transition_api_test))'

+# Protocols suite — fixed ports, MUST be single-threaded, gated by build features
+RUSTFS_BUILD_FEATURES=ftps,webdav,sftp \
+  cargo test -p e2e_test test_protocol_core_suite -- --test-threads=1 --nocapture
 ```

 The protocols suite has its own contract (fixed bind ports 9022–9301,
-single-worker execution, feature-gated scheduling) documented in
+`--test-threads=1`, feature-gated scheduling) documented in
 [`src/protocols/README.md`](src/protocols/README.md). `RUSTFS_BUILD_FEATURES`
 selects which features the spawned binary is built with; leave it unset to run
-every protocol entry. Use the exact profile command under
-[Troubleshooting](#troubleshooting) for CI-equivalent execution.
+every protocol entry.

 ### `#[ignore]` semantics

@@ -157,26 +159,27 @@ construction (random port + isolated temp dir) and need no serialization.
 ## CI map

 `e2e_test` is **excluded** from the main `cargo nextest run --profile ci --all`
-pass (`--exclude e2e_test`) — the whole crate is too slow to gate every PR.
-Subsets join CI through nextest profiles; the fixed-port protocol suite uses
-the same profile for membership and execution with one nightly worker.
+pass ([`.github/workflows/ci.yml`](../../.github/workflows/ci.yml) line 158,
+`--exclude e2e_test`) — the whole crate is too slow to gate every PR. Subsets
+join CI through the nextest profile system only (never as ad-hoc jobs):

 | Suite | Runs where | Status |
 | --- | --- | --- |
 | Smoke subset (`e2e-smoke` profile) | `e2e-tests` job, every PR | **Active** (backlog#1149 ci-4) |
-| Full single-node suite (`e2e-full` profile) | `e2e-full` job, merge queue + main | **Active** (backlog#1149 ci-5) |
 | `s3s-e2e` black-box | `e2e-tests` + `e2e-tests-rio-v2` jobs | **Active** (external conformance tool) |
 | ILM / lifecycle (ignored) | `test-ilm-integration-serial` lane, `-j1` | **Active** (backlog#1148 ilm-1) |
-| KMS suite | `e2e-full` job, merge queue + main | **Active** |
-| Cluster faults (`e2e-nightly` profile) | consolidated nightly workflow | **Active** (backlog#1149 ci-7) |
-| Protocols (FTPS/WebDAV/SFTP) | consolidated nightly workflow, serial | **Active** (backlog#1149 ci-7) |
+| KMS suite | — | Not in CI yet (backlog#1149 ci-5) |
+| Protocols (FTPS/WebDAV/SFTP) | — | Not in CI yet (backlog#1149 ci-7) |
 | Replication (fast subset) | `e2e-smoke` profile, `e2e-tests` job, every PR | **Active** (backlog#1147 repl-1) |
-| Replication (slow + multi-node) | `e2e-repl-nightly` profile, consolidated nightly workflow | **Active** (backlog#1147 repl-1) |
-| `reliant/*` | 19 tests in PR smoke; remaining default tests in `e2e-full` | **Active** except `#[ignore]` |
+| Replication (slow + dual-node) | `e2e-repl-nightly` profile, scheduled workflow | **Active** (backlog#1147 repl-1) |
+| `reliant/*` (pre-started server) | — | Manual only |

-The profile filters in [`.config/nextest.toml`](../../.config/nextest.toml) are
-the wiring source of truth. Committed test-ID digests under
-`.config/e2e-*-selection.txt` make every membership change explicit.
+Links: [`ci.yml`](../../.github/workflows/ci.yml) `e2e-tests` (line 347),
+`test-ilm-integration-serial` (line 196). The `e2e-smoke` `default-filter` in
+[`.config/nextest.toml`](../../.config/nextest.toml) is the **single wiring
+mechanism** — extend that filter (or add a sibling profile) to admit more
+tests; do not add e2e jobs to `ci.yml`. repl-1 / ilm-3 are landing in parallel
+and may add lanes; keep the table above easy to extend.

 ## Troubleshooting

@@ -185,15 +188,9 @@ the wiring source of truth. Committed test-ID digests under
 ```bash
 # Smoke (e2e-tests job) — includes the 20 fast replication tests
 cargo nextest run --profile e2e-smoke -p e2e_test
-# Full single-node merge/main lane
-cargo nextest run --profile e2e-full -p e2e_test
-# Cluster fault nightly lane
-cargo nextest run --profile e2e-nightly -p e2e_test
-# Replication nightly lane; install awscurl so STS paths do not skip
+# Replication nightly lane (16 slow + dual-node tests; install awscurl for the
+# STS dual-node test, else it skips gracefully)
 cargo nextest run --profile e2e-repl-nightly -p e2e_test
-# Fixed-port protocol nightly lane
-RUSTFS_BUILD_FEATURES=ftps,webdav,sftp \
-  cargo nextest run -j 1 --profile e2e-protocols -p e2e_test --no-capture
 # ILM serial lane
 cargo nextest run -j1 --run-ignored ignored-only -p rustfs-scanner -p rustfs \
  -E 'binary(lifecycle_integration_test) or (package(rustfs) and test(lifecycle_transition_api_test))'
@@ -276,6 +273,4 @@ current subset is.
 `docs/testing/e2e-suite-inventory.md` records the per-module test counts as
 listed by `cargo nextest list -p e2e_test`. Regenerate it when adding or
 moving e2e tests so acceptance numbers in the test-strategy issues
-(backlog#1147–#1155) stay auditable. When a profile membership change is
-intentional, review its JSON listing before updating the matching
-`.config/e2e-*-selection.txt` test-ID digest.
+(backlog#1147–#1155) stay auditable.
@@ -55,7 +55,6 @@ mod tests {
    use aws_sdk_s3::Client;
    use aws_sdk_s3::primitives::ByteStream;
    use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
-    use serial_test::serial;
    use sha2::{Digest, Sha256};
    use std::error::Error;
    use tokio::time::{Duration, timeout};
@@ -269,7 +268,6 @@ mod tests {
    /// stripes) and a multipart object (3 parts × 5 MiB) must GET back as a
    /// full, byte-identical body with the correct Content-Length. No early EOF.
    #[tokio::test]
-    #[serial]
    async fn degraded_read_large_objects_with_one_disk_offline_return_full_body() -> TestResult {
        init_logging();
        info!("dist-13 (a): large-object degraded read with one of four disks offline");
@@ -335,7 +333,6 @@ mod tests {
    /// mid-stream — the exact window the fixes had to reconstruct through rather
    /// than truncate.
    #[tokio::test]
-    #[serial]
    async fn degraded_read_reconstructs_through_midstream_bitrot_within_quorum() -> TestResult {
        init_logging();
        info!("dist-13 (b): mid-stream bitrot within quorum must reconstruct a full body");
@@ -393,7 +390,6 @@ mod tests {
    /// Content-Length. `get_checked` panics on that forbidden outcome, so this
    /// test fails loudly if the truncation bug ever returns.
    #[tokio::test]
-    #[serial]
    async fn beyond_quorum_degraded_read_never_silently_truncates() -> TestResult {
        init_logging();
        info!("dist-13 (c): beyond-quorum degraded read must fail, never 200+truncated");
@@ -51,7 +51,6 @@ mod tests {
    use aws_sdk_s3::Client;
    use aws_sdk_s3::primitives::ByteStream;
    use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
-    use serial_test::serial;
    use std::error::Error;
    use tokio::time::{Duration, timeout};
    use tracing::info;
@@ -129,7 +128,6 @@ mod tests {
    /// the body — and assert the server log names the object, at the log level a
    /// default deployment actually runs with.
    #[tokio::test]
-    #[serial]
    async fn midstream_get_failure_is_logged_with_the_object_at_default_log_level() -> TestResult {
        init_logging();
        info!("rustfs#4784: a mid-stream GET failure must name its object in the source log");
@@ -46,7 +46,6 @@ use prost::Message;
 use rustfs_signer::constants::UNSIGNED_PAYLOAD;
 use rustfs_signer::sign_v4;
 use s3s::Body;
-use serial_test::serial;
 use std::collections::BTreeMap;
 use std::convert::Infallible;
 use std::error::Error;
@@ -1695,7 +1694,6 @@ fn assert_storage_layout(
 }

 #[tokio::test]
-#[serial]
 async fn four_node_inline_storage_and_get_boundaries() -> TestResult {
    init_logging();

@@ -1767,7 +1765,6 @@ async fn four_node_inline_storage_and_get_boundaries() -> TestResult {
 }

 #[tokio::test]
-#[serial]
 async fn four_node_empty_legacy_volumes_start_as_fresh() -> TestResult {
    init_logging();

@@ -1805,7 +1802,6 @@ async fn four_node_empty_legacy_volumes_start_as_fresh() -> TestResult {
 }

 #[tokio::test]
-#[serial]
 async fn four_node_inline_fallback_controls() -> TestResult {
    init_logging();

@@ -1870,7 +1866,6 @@ async fn four_node_inline_fallback_controls() -> TestResult {
 }

 #[tokio::test]
-#[serial]
 async fn four_node_compressed_inline_fallback() -> TestResult {
    init_logging();

@@ -1905,7 +1900,6 @@ async fn four_node_compressed_inline_fallback() -> TestResult {
 /// Multipart disk compression is live again, so a compression-enabled cluster classifies multipart objects as compressed and the roundtrip (full GET plus partNumber GET) must still return the original bytes.
 /// Reverting the multipart compression fix must fail this test.
 #[tokio::test]
-#[serial]
 async fn four_node_multipart_disk_compression_roundtrip() -> TestResult {
    init_logging();

@@ -1952,7 +1946,6 @@ async fn four_node_multipart_disk_compression_roundtrip() -> TestResult {
 /// read costs on the order of the covering part's block size against a ~5 MiB
 /// object.
 #[tokio::test]
-#[serial]
 async fn four_node_compressed_multipart_tail_range_reads_are_bounded() -> TestResult {
    init_logging();

@@ -2019,7 +2012,6 @@ async fn four_node_compressed_multipart_tail_range_reads_are_bounded() -> TestRe
 }

 #[tokio::test]
-#[serial]
 async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> TestResult {
    init_logging();

@@ -2123,7 +2115,6 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls() -> Te
 }

 #[tokio::test]
-#[serial]
 async fn four_node_add_tier_converges() -> TestResult {
    init_logging();

@@ -2142,7 +2133,6 @@ async fn four_node_add_tier_converges() -> TestResult {
 }

 #[tokio::test]
-#[serial]
 async fn four_node_add_tier_converges_after_offline_node_restart_without_second_mutation() -> TestResult {
    init_logging();

@@ -2164,7 +2154,6 @@ async fn four_node_add_tier_converges_after_offline_node_restart_without_second_
 }

 #[tokio::test]
-#[serial]
 async fn four_node_manual_transition_job_status_survives_node_restart() -> TestResult {
    init_logging();

@@ -2239,7 +2228,6 @@ async fn four_node_manual_transition_job_status_survives_node_restart() -> TestR
 }

 #[tokio::test]
-#[serial]
 async fn four_node_manual_transition_distributed_admission_conflict_reports_status_and_backpressure() -> TestResult {
    init_logging();

@@ -2381,7 +2369,6 @@ async fn four_node_manual_transition_distributed_admission_conflict_reports_stat
 }

 #[tokio::test]
-#[serial]
 #[ignore = "manual #1508 evidence harness: starts a 4-node cluster, a remote tier, and an in-flight transition job"]
 async fn four_node_manual_transition_rollout_non_empty_restart_readback() -> TestResult {
    init_logging();
@@ -2486,7 +2473,6 @@ async fn four_node_manual_transition_rollout_non_empty_restart_readback() -> Tes
 }

 #[tokio::test]
-#[serial]
 async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_transition() -> TestResult {
    init_logging();

@@ -2598,7 +2584,6 @@ async fn four_node_mixed_msgpack_compat_mode_preserves_fallback_controls_during_
 }

 #[tokio::test]
-#[serial]
 async fn four_node_transitioned_inline_fallback() -> TestResult {
    init_logging();

@@ -19,7 +19,6 @@
 //! multipart upload behaviour.

 use crate::common::{TEST_BUCKET, init_logging};
-use serial_test::serial;
 use tokio::time::{Duration, sleep};
 use tracing::{error, info};

@@ -62,7 +61,6 @@ impl VaultKmsTestContext {
 }

 #[tokio::test]
-#[serial]
 async fn test_vault_kms_end_to_end() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    init_logging();
    if skip_if_kms_admin_tool_unavailable("test_vault_kms_end_to_end") {
@@ -118,7 +116,6 @@ async fn test_vault_kms_end_to_end() -> Result<(), Box<dyn std::error::Error + S
 }

 #[tokio::test]
-#[serial]
 async fn test_vault_kms_key_isolation() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    init_logging();
    if skip_if_kms_admin_tool_unavailable("test_vault_kms_key_isolation") {
@@ -205,7 +202,6 @@ async fn test_vault_kms_key_isolation() -> Result<(), Box<dyn std::error::Error
 }

 #[tokio::test]
-#[serial]
 async fn test_vault_kms_large_file() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    init_logging();
    if skip_if_kms_admin_tool_unavailable("test_vault_kms_large_file") {
@@ -270,7 +266,6 @@ async fn test_vault_kms_large_file() -> Result<(), Box<dyn std::error::Error + S
 }

 #[tokio::test]
-#[serial]
 async fn test_vault_kms_multipart_upload() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    init_logging();
    if skip_if_kms_admin_tool_unavailable("test_vault_kms_multipart_upload") {
@@ -301,7 +296,6 @@ async fn test_vault_kms_multipart_upload() -> Result<(), Box<dyn std::error::Err
 }

 #[tokio::test]
-#[serial]
 async fn test_vault_kms_key_operations() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    init_logging();
    if skip_if_kms_admin_tool_unavailable("test_vault_kms_key_operations") {
@@ -17,7 +17,6 @@
 use crate::common::{awscurl_delete, awscurl_put, init_logging};
 use crate::policy::test_env::PolicyTestEnvironment;
 use aws_sdk_s3::primitives::ByteStream;
-use serial_test::serial;
 use tracing::info;

 /// Helper function to create a regular user with given credentials
@@ -122,7 +121,6 @@ async fn cleanup_user_and_policy(env: &PolicyTestEnvironment, username: &str, po

 /// Test AWS policy variables with single-value scenarios
 #[tokio::test(flavor = "multi_thread")]
-#[serial]
 #[ignore = "Starts a rustfs server; enable when running full E2E"]
 pub async fn test_aws_policy_variables_single_value() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    test_aws_policy_variables_single_value_impl().await
@@ -275,7 +273,6 @@ pub async fn test_aws_policy_variables_single_value_impl_with_env(

 /// Test AWS policy variables with multi-value scenarios
 #[tokio::test(flavor = "multi_thread")]
-#[serial]
 #[ignore = "Starts a rustfs server; enable when running full E2E"]
 pub async fn test_aws_policy_variables_multi_value() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    test_aws_policy_variables_multi_value_impl().await
@@ -401,7 +398,6 @@ pub async fn test_aws_policy_variables_multi_value_impl_with_env(

 /// Test AWS policy variables with variable concatenation
 #[tokio::test(flavor = "multi_thread")]
-#[serial]
 #[ignore = "Starts a rustfs server; enable when running full E2E"]
 pub async fn test_aws_policy_variables_concatenation() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    test_aws_policy_variables_concatenation_impl().await
@@ -491,7 +487,6 @@ pub async fn test_aws_policy_variables_concatenation_impl_with_env(

 /// Test AWS policy variables with nested scenarios
 #[tokio::test(flavor = "multi_thread")]
-#[serial]
 #[ignore = "Starts a rustfs server; enable when running full E2E"]
 pub async fn test_aws_policy_variables_nested() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    test_aws_policy_variables_nested_impl().await
@@ -509,7 +504,6 @@ pub async fn test_aws_policy_variables_nested_impl() -> Result<(), Box<dyn std::

 /// Test AWS policy variables with STS temporary credentials
 #[tokio::test(flavor = "multi_thread")]
-#[serial]
 #[ignore = "Starts a rustfs server; enable when running full E2E"]
 pub async fn test_aws_policy_variables_sts() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    test_aws_policy_variables_sts_impl().await
@@ -705,7 +699,6 @@ pub async fn test_aws_policy_variables_sts_impl_with_env(

 /// Test AWS policy variables with deny scenarios
 #[tokio::test(flavor = "multi_thread")]
-#[serial]
 #[ignore = "Starts a rustfs server; enable when running full E2E"]
 pub async fn test_aws_policy_variables_deny() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    test_aws_policy_variables_deny_impl().await
@@ -14,7 +14,6 @@

 use crate::common::init_logging;
 use crate::policy::test_env::PolicyTestEnvironment;
-use serial_test::serial;
 use std::time::Instant;
 use tokio::time::{Duration, sleep};
 use tracing::{error, info};
@@ -213,7 +212,6 @@ impl PolicyTestSuite {

 /// Test suite
 #[tokio::test]
-#[serial]
 #[ignore = "Connects to existing rustfs server"]
 async fn test_policy_critical_suite() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let config = TestSuiteConfig {
@@ -11,17 +11,10 @@ test process directly.

 ## Running Tests

-Use the canonical CI-equivalent protocol command in the parent
-[`e2e_test` README](../../README.md#troubleshooting).
-
-For targeted debugging of the core suite only:
-
 ```bash
 RUSTFS_BUILD_FEATURES=ftps,webdav,sftp cargo test --package e2e_test test_protocol_core_suite -- --test-threads=1 --nocapture
 ```

-This targeted command does not cover the full `e2e-protocols` profile.
-
 `RUSTFS_BUILD_FEATURES` controls which features the test rustfs binary is
 built with. When this variable is set, the protocol test runner schedules
 only entries whose protocol is present in the requested feature list. Leave
@@ -140,3 +133,4 @@ property without consulting any external doc.
 Bind ports 9023 (SFTP) and 9100 (S3). Spawns rustfs with
 `RUSTFS_SFTP_IDLE_TIMEOUT=5`, sleeps 10 s past the timeout, then issues an
 SFTP request and asserts the server has closed the session.
+
@@ -41,7 +41,6 @@ use reqwest::Client;
 use rustfs_signer::constants::UNSIGNED_PAYLOAD;
 use rustfs_signer::sign_v4;
 use s3s::Body;
-use serial_test::serial;
 use tokio::process::Command;
 use tracing::info;

@@ -821,7 +820,6 @@ pub async fn test_webdav_core_operations() -> Result<()> {
 }

 #[tokio::test]
-#[serial]
 async fn test_webdav_core_operations_direct() -> Result<()> {
    test_webdav_core_operations().await
 }
@@ -27,7 +27,6 @@ mod tests {
    use aws_sdk_s3::Client;
    use aws_sdk_s3::primitives::ByteStream;
    use aws_sdk_s3::types::{BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, VersioningConfiguration};
-    use serial_test::serial;
    use sha2::{Digest, Sha256};
    use std::collections::HashSet;
    use std::error::Error;
@@ -157,7 +156,6 @@ mod tests {
    /// content, degraded writes must succeed, and everything must still
    /// verify after the disk returns.
    #[tokio::test]
-    #[serial]
    async fn test_degraded_read_write_with_one_disk_offline() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        init_logging();
        info!("Reliability: degraded read/write with one of four disks offline");
@@ -210,7 +208,6 @@ mod tests {
    /// bytes to a reader: per-shard bitrot checksums reject the bad shard and
    /// the object is reconstructed from the remaining shards.
    #[tokio::test]
-    #[serial]
    async fn test_bitrot_corrupted_shard_read_returns_correct_data() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        init_logging();
        info!("Reliability: GET must read through a bitrot-corrupted shard");
@@ -253,7 +250,6 @@ mod tests {
    /// heal, and require the replaced disk to be rebuilt and all content to
    /// verify against the sha256 manifest.
    #[tokio::test]
-    #[serial]
    async fn test_fresh_disk_replacement_heals_after_sigkill_restart() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        init_logging();
        info!("Reliability: fresh-disk replacement heals after SIGKILL restart");
@@ -327,7 +323,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_versioned_shard_census_selects_each_version_data_dir() -> Result<(), Box<dyn Error + Send + Sync>> {
        init_logging();
        info!("Reliability: physical shard census selects the requested object version");
@@ -29,7 +29,6 @@ mod tests {
    use aws_sdk_s3::primitives::ByteStream;
    use aws_sdk_s3::types::{BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, VersioningConfiguration};
    use http::Method;
-    use serial_test::serial;
    use sha2::{Digest, Sha256};
    use std::collections::BTreeSet;
    use std::error::Error;
@@ -1061,7 +1060,6 @@ mod tests {
    /// Linux mount namespaces are per-thread; keep mount setup and process
    /// spawning on one OS thread so child RustFS nodes inherit the test mounts.
    #[tokio::test(flavor = "current_thread")]
-    #[serial]
    #[ignore = "requires Linux root/CAP_SYS_ADMIN and RUSTFS_PRIVILEGED_REPLACEMENT_E2E=1"]
    async fn test_privileged_3x4_auto_replacement_rebuilds_ec8_plus_4_without_admin_heal()
    -> Result<(), Box<dyn Error + Send + Sync>> {
@@ -1075,7 +1073,6 @@ mod tests {
    /// Linux mount namespaces are per-thread; keep mount setup and process
    /// spawning on one OS thread so child RustFS nodes inherit the test mounts.
    #[tokio::test(flavor = "current_thread")]
-    #[serial]
    #[ignore = "requires Linux root/CAP_SYS_ADMIN and RUSTFS_PRIVILEGED_REPLACEMENT_E2E=1"]
    async fn test_privileged_3x4_auto_replacement_rebuilds_ec6_plus_6_without_admin_heal()
    -> Result<(), Box<dyn Error + Send + Sync>> {
@@ -922,10 +922,14 @@ mod prepared_get_object_metadata_tests {
            .expect("test should find an object whose initial fanout covers both data shards")
    }

-    fn bounded_initial_parity_disk_index(bucket: &str, object: &str) -> usize {
+    #[allow(
+        dead_code,
+        reason = "test fixture no assertion in this module uses today; the live namesake lives in io_primitives tests (backlog#1823)"
+    )]
+    fn bounded_spare_disk_index(bucket: &str, object: &str) -> usize {
        *bounded_metadata_fanout_order(bucket, object, 4, 2)
-            .get(2)
-            .expect("4-disk test geometry should schedule one parity disk initially")
+            .get(3)
+            .expect("4-disk test geometry should leave one bounded spare disk")
    }

    #[tokio::test]
@@ -1083,7 +1087,7 @@ mod prepared_get_object_metadata_tests {
                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>),
                    ],
                    async {
-                        let slow_parity_disk = bounded_initial_parity_disk_index(bucket, &object);
+                        let slow_parity_disk = bounded_spare_disk_index(bucket, &object);
                        let barrier =
                            rename_fanout_barrier::arm(&object, slow_parity_disk, rename_fanout_barrier::PHASE_READ_VERSION);
                        let calls = disk_call_counters::observe(&object);
@@ -110,10 +110,7 @@ impl HealStorageAPI for MockStorage {
        Ok(Vec::new())
    }

-    async fn get_bucket_info(&self, bucket: &str) -> Result<Option<BucketInfo>> {
-        if bucket == "panic" {
-            panic!("test-only panic payload must not escape the scheduler");
-        }
+    async fn get_bucket_info(&self, _bucket: &str) -> Result<Option<BucketInfo>> {
        Ok(None)
    }

@@ -1024,231 +1021,6 @@ async fn test_task_alias_is_removed_after_terminal_completion() {
    assert_eq!(manager.canonical_task_id(&duplicate_id).await, duplicate_id);
 }

-#[tokio::test]
-#[serial_test::serial]
-async fn scheduler_panic_releases_active_slot_and_allows_same_target_readmission() {
-    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
-    let manager = HealManager::new(storage, None);
-    let request = bucket_request("panic", HealPriority::Normal, HealRequestSource::Admin);
-    let task_id = request.id.clone();
-
-    assert_eq!(
-        manager
-            .submit_heal_request(request)
-            .await
-            .expect("panic request should be admitted"),
-        HealAdmissionResult::Accepted
-    );
-    let duplicate = bucket_request("panic", HealPriority::Normal, HealRequestSource::Admin);
-    let duplicate_id = duplicate.id.clone();
-    assert_eq!(
-        manager
-            .submit_heal_request(duplicate)
-            .await
-            .expect("same target should merge while active is queued"),
-        HealAdmissionResult::Merged
-    );
-    assert_eq!(manager.canonical_task_id(&duplicate_id).await, task_id);
-    process_manager_queue_once(&manager).await;
-
-    let status = tokio::time::timeout(Duration::from_secs(1), async {
-        loop {
-            if let Ok(status) = manager.get_task_status(&task_id).await
-                && matches!(status, HealTaskStatus::Failed { .. })
-            {
-                break status;
-            }
-            tokio::time::sleep(Duration::from_millis(10)).await;
-        }
-    })
-    .await
-    .expect("panic task should reach a terminal status");
-    assert_eq!(
-        status,
-        HealTaskStatus::Failed {
-            error: PANICKED_HEAL_TASK_ERROR.to_string()
-        }
-    );
-    assert_eq!(manager.get_active_task_count().await, 0);
-    assert_eq!(manager.get_queue_length().await, 0);
-    assert!(manager.retrying_heals.lock().await.is_empty());
-    assert!(manager.task_aliases.lock().await.is_empty());
-    assert!(manager.completed_heals.lock().await.contains_key(&task_id));
-    assert_eq!(manager.canonical_task_id(&duplicate_id).await, duplicate_id);
-
-    let readmitted = bucket_request("panic", HealPriority::Normal, HealRequestSource::Admin);
-    assert_eq!(
-        manager
-            .submit_heal_request(readmitted)
-            .await
-            .expect("same target should be re-admitted after a panic"),
-        HealAdmissionResult::Accepted
-    );
-}
-
-#[tokio::test]
-#[serial_test::serial]
-async fn retry_child_panic_finishes_parent_once() {
-    clear_scheduler_panic();
-    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
-    let manager = HealManager::new(storage, None);
-    let request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
-    let task_id = request.id.clone();
-    assert_eq!(
-        manager
-            .submit_heal_request(request)
-            .await
-            .expect("retry request should be admitted"),
-        HealAdmissionResult::Accepted
-    );
-    arm_scheduler_panic(SchedulerPanicPoint::RetryChild, &task_id);
-    process_manager_queue_once(&manager).await;
-
-    let status = tokio::time::timeout(Duration::from_secs(1), async {
-        loop {
-            if let Ok(status) = manager.get_task_status(&task_id).await
-                && matches!(status, HealTaskStatus::Failed { .. })
-            {
-                break status;
-            }
-            tokio::time::sleep(Duration::from_millis(10)).await;
-        }
-    })
-    .await
-    .expect("retry child panic should finish the parent");
-    clear_scheduler_panic();
-    assert_eq!(
-        status,
-        HealTaskStatus::Failed {
-            error: PANICKED_HEAL_TASK_ERROR.to_string()
-        }
-    );
-    assert_eq!(manager.get_active_task_count().await, 0);
-    assert_eq!(manager.get_queue_length().await, 0);
-    assert!(manager.retrying_heals.lock().await.is_empty());
-    assert!(manager.task_aliases.lock().await.is_empty());
-    assert_eq!(manager.completed_heals.lock().await.len(), 1);
-    assert_eq!(manager.get_statistics().await.failed_tasks, 1);
-}
-
-#[tokio::test]
-#[serial_test::serial]
-async fn cleanup_panic_is_supervised() {
-    clear_scheduler_panic();
-    let notice_bucket = "cleanup-panic-mrf";
-    let notice_object = "object";
-    let _ = rustfs_common::mrf_channel::take_mrf_repaired_events_for(notice_bucket);
-    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
-    let manager = HealManager::new(storage, None);
-    let mut request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::Normal);
-    request.source = HealRequestSource::Admin;
-    let task_id = request.id.clone();
-    assert_eq!(
-        manager
-            .submit_heal_request(request)
-            .await
-            .expect("cleanup request should be admitted"),
-        HealAdmissionResult::Accepted
-    );
-    manager
-        .mrf_repair_notice_targets
-        .lock()
-        .expect("mrf repair notice registry poisoned")
-        .insert(
-            task_id.clone(),
-            vec![MrfRepairNoticeTarget {
-                bucket: Arc::from(notice_bucket),
-                object: Arc::from(notice_object),
-                version_id: None,
-            }],
-        );
-    arm_scheduler_panic(SchedulerPanicPoint::Cleanup, &task_id);
-    process_manager_queue_once(&manager).await;
-
-    let status = tokio::time::timeout(Duration::from_secs(1), async {
-        loop {
-            if let Ok(status) = manager.get_task_status(&task_id).await
-                && matches!(status, HealTaskStatus::Completed)
-            {
-                break status;
-            }
-            tokio::time::sleep(Duration::from_millis(10)).await;
-        }
-    })
-    .await
-    .expect("cleanup panic should leave a terminal status");
-    clear_scheduler_panic();
-    assert_eq!(status, HealTaskStatus::Completed);
-    assert_eq!(manager.get_active_task_count().await, 0);
-    assert!(manager.task_aliases.lock().await.is_empty());
-    assert_eq!(manager.completed_heals.lock().await.len(), 1);
-    assert_eq!(manager.get_statistics().await.successful_tasks, 1);
-    let events = rustfs_common::mrf_channel::take_mrf_repaired_events_for(notice_bucket);
-    assert_eq!(events.len(), 1, "cleanup panic must preserve successful MRF notice delivery");
-    assert_eq!(events[0].object.as_ref(), notice_object);
-}
-
-#[tokio::test]
-#[serial_test::serial]
-async fn cancelled_retry_child_panic_does_not_rearchive_failed_status() {
-    let manager = HealManager::new(Arc::new(MockStorage), None);
-    let request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
-    let task_id = request.id.clone();
-    let retry_cancel_token = insert_retrying_request(&manager, request.clone()).await;
-
-    manager
-        .cancel_task(&task_id)
-        .await
-        .expect("retry cancellation should succeed");
-    assert!(retry_cancel_token.is_cancelled());
-
-    let state = PanicCleanupState {
-        active_heals: manager.active_heals.clone(),
-        heal_queue: manager.heal_queue.clone(),
-        completed_heals: manager.completed_heals.clone(),
-        task_aliases: manager.task_aliases.clone(),
-        retrying_heals: manager.retrying_heals.clone(),
-        mrf_repair_notice_targets: manager.mrf_repair_notice_targets.clone(),
-        replacement_recovery_anchors: manager.replacement_recovery_anchors.clone(),
-        statistics: manager.statistics.clone(),
-    };
-    finish_panicked_retry_child(task_id.clone(), request.heal_type, retry_cancel_token, state).await;
-
-    assert!(manager.retrying_heals.lock().await.is_empty());
-    assert!(manager.completed_heals.lock().await.is_empty());
-    assert_eq!(manager.get_statistics().await.failed_tasks, 0);
-}
-
-#[tokio::test]
-async fn active_cancel_wins_parent_panic_cleanup_without_completed_status() {
-    let manager = HealManager::new(Arc::new(MockStorage), None);
-    let request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::Normal);
-    let task_id = request.id.clone();
-    let task = Arc::new(HealTask::from_request(request, Arc::new(MockStorage)));
-    manager.active_heals.lock().await.insert(task_id.clone(), task.clone());
-
-    manager
-        .cancel_task(&task_id)
-        .await
-        .expect("active task cancellation should win");
-    assert_eq!(task.get_status().await, HealTaskStatus::Cancelled);
-
-    let state = PanicCleanupState {
-        active_heals: manager.active_heals.clone(),
-        heal_queue: manager.heal_queue.clone(),
-        completed_heals: manager.completed_heals.clone(),
-        task_aliases: manager.task_aliases.clone(),
-        retrying_heals: manager.retrying_heals.clone(),
-        mrf_repair_notice_targets: manager.mrf_repair_notice_targets.clone(),
-        replacement_recovery_anchors: manager.replacement_recovery_anchors.clone(),
-        statistics: manager.statistics.clone(),
-    };
-    finish_panicked_heal_task(task, task_id, state).await;
-
-    assert!(manager.completed_heals.lock().await.is_empty());
-    assert_eq!(manager.get_statistics().await.failed_tasks, 0);
-}
-
 #[tokio::test]
 async fn test_duplicate_admission_is_atomic_with_queue_to_active_transition() {
    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
@@ -69,7 +69,7 @@ uuid = { workspace = true, features = ["v4", "serde", "fast-rng", "macro-diagnos
 [dev-dependencies]
 metrics-util = { workspace = true, features = ["debugging"] }
 proptest = "1"
-serial_test.workspace = true
+serial_test = { workspace = true }
 temp-env.workspace = true
 tokio = { workspace = true, features = ["macros", "fs", "rt-multi-thread"] }

@@ -1564,7 +1564,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn abort_incomplete_multipart_upload_due_accepts_zero_days() {
        let initiated = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -1625,7 +1624,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn predict_expiration_selects_closest_expiry_for_put_object() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -1872,7 +1870,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn empty_transition_vectors_are_not_active_or_due() {
        let lc = BucketLifecycleConfiguration {
            expiry_updated_at: None,
@@ -1938,7 +1935,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn eval_inner_keeps_latest_object_before_days_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -1972,7 +1968,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn eval_inner_transitions_latest_object_after_days_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2010,7 +2005,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn eval_inner_transitions_latest_object_after_date_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let transition_date = base_time - Duration::days(1);
@@ -2050,7 +2044,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn eval_inner_selects_earliest_due_among_multiple_past_due_events() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        // Two enabled rules both yield a past-due DeleteAction and a third yields a
@@ -2164,7 +2157,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn eval_inner_expires_noncurrent_version_after_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2202,7 +2194,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn eval_inner_skips_noncurrent_expiration_without_successor() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).expect("valid fixed test timestamp");
        let lc = BucketLifecycleConfiguration {
@@ -2238,7 +2229,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn eval_inner_missing_successor_does_not_skip_noncurrent_transition() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).expect("valid fixed test timestamp");
        let lc = BucketLifecycleConfiguration {
@@ -2281,7 +2271,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn eval_inner_noncurrent_expiration_one_day_respects_due_boundary() {
        let successor_time = datetime!(2025-06-15 12:00:00 UTC);
        let due = expected_expiry_time(successor_time, 1);
@@ -2323,7 +2312,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn eval_inner_expires_noncurrent_version_immediately_when_zero_days() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2361,7 +2349,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn eval_inner_transitions_noncurrent_version_after_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2437,7 +2424,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn evaluator_honors_newer_noncurrent_versions_retention_count() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = Arc::new(BucketLifecycleConfiguration {
@@ -2726,7 +2712,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn expired_object_delete_marker_ignores_marker_with_noncurrent_versions_present() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2803,7 +2788,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn expired_object_delete_marker_deletes_only_delete_marker_immediately() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2881,7 +2865,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn expiration_days_deletes_only_expired_delete_marker_when_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -2932,7 +2915,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn expiration_days_uses_earliest_due_rule_for_expired_delete_marker() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let make_rule = |id: &str, days| LifecycleRule {
@@ -3263,7 +3245,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn del_marker_expiration_deletes_marker_and_older_versions_when_due() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).expect("fixed timestamp should be valid");
        let lc = BucketLifecycleConfiguration {
@@ -3303,7 +3284,6 @@ mod tests {
    // --- TASK-003 tests: Round up to next UTC processing boundary ---

    #[test]
-    #[serial]
    fn expected_expiry_time_rounds_up_to_next_midnight_utc() {
        with_default_ilm_process_time(|| {
            // Object created at 2025-01-15T10:30:45Z, expire in 30 days
@@ -3319,7 +3299,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn expected_expiry_time_immediate_expiry_returns_epoch() {
        with_default_ilm_process_time(|| {
            let mod_time = datetime!(2025-06-01 12:00:00 UTC);
@@ -3329,7 +3308,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn expected_expiry_time_preserves_exact_midnight_boundary() {
        with_default_ilm_process_time(|| {
            let mod_time = datetime!(2025-03-01 00:00:00 UTC);
@@ -3339,7 +3317,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn expected_expiry_time_rounds_end_of_day_to_following_midnight() {
        with_default_ilm_process_time(|| {
            let mod_time = datetime!(2025-06-15 23:59:59 UTC);
@@ -3349,7 +3326,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn expected_expiry_time_uses_canonical_process_time_boundary() {
        let mod_time = datetime!(2025-01-15 10:30:45 UTC);

@@ -3362,7 +3338,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn expected_expiry_time_uses_deprecated_process_time_alias() {
        let mod_time = datetime!(2025-01-15 10:30:45 UTC);

@@ -3375,7 +3350,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn expected_expiry_time_uses_default_boundary_when_process_time_is_zero_or_invalid() {
        let mod_time = datetime!(2025-01-15 10:30:45 UTC);

@@ -3398,7 +3372,6 @@ mod tests {

    // (a) Default path (env unset) is byte-identical: one day == 86400s.
    #[test]
-    #[serial]
    fn ilm_day_secs_defaults_to_86400_when_unset() {
        temp_env::with_var_unset(ENV_ILM_DEBUG_DAY_SECS, || {
            assert_eq!(ilm_day_secs(), DEFAULT_ILM_DAY_SECS);
@@ -3427,7 +3400,6 @@ mod tests {

    // (b) End-to-end env read scales the day length.
    #[test]
-    #[serial]
    fn ilm_day_secs_scales_when_env_set() {
        temp_env::with_var(ENV_ILM_DEBUG_DAY_SECS, Some("2"), || {
            assert_eq!(ilm_day_secs(), 2);
@@ -3436,7 +3408,6 @@ mod tests {

    // (c) Invalid env value falls back to 86400.
    #[test]
-    #[serial]
    fn ilm_day_secs_falls_back_on_invalid_env() {
        temp_env::with_var(ENV_ILM_DEBUG_DAY_SECS, Some("bogus"), || {
            assert_eq!(ilm_day_secs(), DEFAULT_ILM_DAY_SECS);
@@ -3449,7 +3420,6 @@ mod tests {
    // Deadline math scales: with a 1s day and PROCESS_TIME unset, a Days=1 rule is
    // due 1s after mod_time (rounded up to the next 1s boundary => same instant).
    #[test]
-    #[serial]
    fn expected_expiry_time_scales_with_debug_day_secs() {
        let mod_time = datetime!(2025-01-15 10:30:45 UTC);
        temp_env::with_var(ENV_ILM_DEBUG_DAY_SECS, Some("1"), || {
@@ -3465,7 +3435,6 @@ mod tests {

    // days == 0 still yields the immediate-expiry sentinel regardless of the switch.
    #[test]
-    #[serial]
    fn expected_expiry_time_zero_days_ignores_debug_day_secs() {
        let mod_time = datetime!(2025-06-01 12:00:00 UTC);
        temp_env::with_var(ENV_ILM_DEBUG_DAY_SECS, Some("2"), || {
@@ -3476,7 +3445,6 @@ mod tests {
    // (③) Interaction with an explicit RUSTFS_ILM_PROCESS_TIME: the deadline offset
    // uses the accelerated day length, but the rounding boundary honors PROCESS_TIME.
    #[test]
-    #[serial]
    fn expected_expiry_time_debug_day_secs_respects_explicit_process_time() {
        let mod_time = datetime!(2025-01-15 10:30:00 UTC);
        // day == 10s, but round up to the next 60s (PROCESS_TIME) boundary.
@@ -3493,7 +3461,6 @@ mod tests {

    // (③) With the switch unset, an explicit PROCESS_TIME behaves exactly as before.
    #[test]
-    #[serial]
    fn expected_expiry_time_unset_debug_day_secs_matches_legacy_process_time() {
        let mod_time = datetime!(2025-01-15 10:30:45 UTC);
        temp_env::with_var_unset(ENV_ILM_DEBUG_DAY_SECS, || {
@@ -3521,7 +3488,6 @@ mod tests {

    // The abort-incomplete-multipart deadline path also scales through the switch.
    #[test]
-    #[serial]
    fn abort_incomplete_multipart_due_scales_with_debug_day_secs() {
        use s3s::dto::AbortIncompleteMultipartUpload;
        let initiated = datetime!(2025-01-15 10:30:45 UTC);
@@ -3566,7 +3532,6 @@ mod tests {
    // (⑤ evaluator seam) A Days=1 rule fires under RUSTFS_ILM_DEBUG_DAY_SECS=1 once
    // `now` advances a few seconds past a mod_time only ~seconds in the past.
    #[test]
-    #[serial]
    fn eval_inner_expires_days_one_rule_under_debug_day_secs() {
        let lc = BucketLifecycleConfiguration {
            expiry_updated_at: None,
@@ -3615,7 +3580,6 @@ mod tests {

    // Absolute Date-based rules must NOT scale with the switch (regression guard).
    #[test]
-    #[serial]
    fn eval_inner_date_rule_ignores_debug_day_secs() {
        let expiry_date = datetime!(2025-06-01 00:00:00 UTC);
        let lc = BucketLifecycleConfiguration {
@@ -3873,7 +3837,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn eval_inner_triggers_delete_all_versions_when_expired_object_all_versions_set() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -3912,7 +3875,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn expired_object_all_versions_does_not_apply_to_current_delete_marker() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).expect("fixed timestamp should be valid");
        let lc = BucketLifecycleConfiguration {
@@ -3942,7 +3904,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn eval_inner_uses_delete_action_when_all_versions_not_set() {
        let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap();
        let lc = BucketLifecycleConfiguration {
@@ -4061,7 +4022,6 @@ mod tests {
        use super::*;
        use proptest::prelude::*;
        use s3s::dto::{NoncurrentVersionExpiration, Tag};
-        use serial_test::serial;

        const DAY_SECS: i64 = 86400;

@@ -4292,7 +4252,6 @@ mod tests {
            /// combination, and must be deterministic: the same input
            /// evaluated twice yields an identical event.
            #[test]
-            #[serial]
            fn eval_inner_never_panics_and_is_deterministic(
                rules in prop::collection::vec(arb_rule(), 0..4),
                obj in arb_object_opts(),
@@ -4432,7 +4391,6 @@ mod tests {
            /// candidate set — earliest due wins, ties prefer delete-class —
            /// and must be `NoneAction` exactly when that set is empty.
            #[test]
-            #[serial]
            fn eval_inner_winner_matches_selection_oracle(
                rules in prop::collection::vec(arb_selection_rule(), 0..5),
                mod_off in 0i64..(2 * DAY_SECS),
@@ -4486,7 +4444,6 @@ mod tests {
            /// non-decreasing in `days` (days == 0 maps to UNIX_EPOCH, below
            /// any post-1970 deadline).
            #[test]
-            #[serial]
            fn expected_expiry_time_is_monotonic_in_days(
                mod_off in 0i64..(3650 * DAY_SECS),
                d1 in 0i32..2000,
@@ -4508,7 +4465,6 @@ mod tests {
            /// to the next whole-day boundary: the result is day-aligned, not
            /// before `mod_time + days`, and less than one boundary beyond it.
            #[test]
-            #[serial]
            fn expected_expiry_time_lands_on_default_day_boundary(
                mod_off in 0i64..(3650 * DAY_SECS),
                days in 1i32..2000,
@@ -4526,7 +4482,6 @@ mod tests {
            /// to that boundary instead: aligned to it, never early, and less
            /// than one boundary late.
            #[test]
-            #[serial]
            fn expected_expiry_time_lands_on_explicit_process_boundary(
                mod_off in 0i64..(365 * DAY_SECS),
                days in 1i32..400,
@@ -73,7 +73,6 @@ walkdir = { workspace = true }

 [dev-dependencies]
 criterion = { workspace = true, features = ["html_reports"] }
-serial_test = { workspace = true }
 temp-env = { workspace = true, features = ["async_closure"] }
 tempfile = { workspace = true }
 tokio = { workspace = true, features = ["test-util", "macros", "fs", "rt-multi-thread"] }
@@ -1484,7 +1484,6 @@ mod tests {
        ENV_CAPACITY_SAMPLE_RATE, ENV_CAPACITY_STAT_TIMEOUT, ENV_CAPACITY_WRITE_FREQUENCY_THRESHOLD,
        ENV_CAPACITY_WRITE_TRIGGER_DELAY,
    };
-    use serial_test::serial;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

@@ -1669,7 +1668,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn test_config_getter_defaults() {
        for (env_var, getter, default, _, _) in config_getter_cases() {
            temp_env::with_var(env_var, None::<&str>, || {
@@ -1679,7 +1677,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn test_config_getter_env_overrides() {
        for (env_var, getter, _, override_value, expected) in config_getter_cases() {
            temp_env::with_var(env_var, Some(override_value), || {
@@ -1689,7 +1686,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn test_zero_env_values_clamp_to_defaults() {
        // A zero threshold makes small disks report 0 bytes; a zero timeout
        // (with dynamic timeout off) makes every scan fail. Both must fall
@@ -1709,7 +1705,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_update_capacity_preserves_retrieval_metadata() {
        let manager = HybridCapacityManager::from_env();

@@ -1725,7 +1720,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_record_write_operation() {
        let manager = HybridCapacityManager::from_env();

@@ -1736,7 +1730,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_write_frequency_window() {
        let manager = HybridCapacityManager::from_env();

@@ -1824,7 +1817,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn test_recent_write_count_ignores_future_buckets() {
        let record = WriteRecord::new();
        record.write_buckets[0].store(120, 3);
@@ -1838,7 +1830,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_needs_fast_update() {
        let manager = HybridCapacityManager::from_env();

@@ -1855,7 +1846,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_cache_age_tracking() {
        let manager = HybridCapacityManager::from_env();

@@ -1875,7 +1865,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_data_source_tracking() {
        let manager = HybridCapacityManager::from_env();

@@ -1891,7 +1880,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_needs_fast_update_waits_for_write_trigger_delay() {
        let manager = create_isolated_manager(HybridStrategyConfig {
            scheduled_update_interval: Duration::from_secs(60),
@@ -1922,7 +1910,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_needs_fast_update_respects_enable_write_trigger() {
        let manager = create_isolated_manager(HybridStrategyConfig {
            scheduled_update_interval: Duration::from_secs(60),
@@ -1949,7 +1936,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_concurrent_access() {
        let manager = Arc::new(HybridCapacityManager::from_env());
        let mut handles = Vec::new();
@@ -1976,7 +1962,6 @@ mod tests {
    // exact under heavy same-second contention or the frequency window (and the
    // write-trigger decision) would undercount.
    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
-    #[serial]
    async fn test_record_write_operation_lock_free_is_exact_under_contention() {
        let manager = Arc::new(HybridCapacityManager::from_env());
        let mut handles = Vec::new();
@@ -2001,7 +1986,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_performance_overhead() {
        let manager = Arc::new(HybridCapacityManager::from_env());
        let start = Instant::now();
@@ -2018,7 +2002,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_refresh_or_join_singleflight() {
        let manager = Arc::new(HybridCapacityManager::from_env());
        let calls = Arc::new(AtomicUsize::new(0));
@@ -2058,7 +2041,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_refresh_or_join_recovers_after_leader_cancellation() {
        let manager = Arc::new(HybridCapacityManager::from_env());

@@ -2087,7 +2069,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_refresh_or_join_cancelled_leader_unblocks_joiner() {
        let manager = Arc::new(HybridCapacityManager::from_env());

@@ -2115,7 +2096,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_spawn_refresh_if_needed_deduplicates_background_refresh() {
        let manager = Arc::new(HybridCapacityManager::from_env());
        let calls = Arc::new(AtomicUsize::new(0));
@@ -2153,7 +2133,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_record_write_operation_with_scope_token_marks_dirty_disks() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());
        let token = uuid::Uuid::new_v4();
@@ -2177,7 +2156,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_get_dirty_disks_drains_global_dirty_scope_registry() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());
        record_global_dirty_scope(CapacityScope {
@@ -2197,7 +2175,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_update_capacity_recomputes_total_from_disk_cache_for_subset_refresh() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());

@@ -2308,7 +2285,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_update_capacity_degraded_full_refresh_merges_cache_and_does_not_oscillate() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());

@@ -2354,7 +2330,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_update_capacity_degraded_with_empty_per_disk_serves_merged_cache() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());
        manager.update_capacity(full_two_disk_update(), DataSource::RealTime).await;
@@ -2384,7 +2359,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_update_capacity_degraded_without_complete_cache_keeps_partial_sum() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());

@@ -2425,7 +2399,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_commit_keeps_dirty_marks_recorded_after_scan_start() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());
        let disk = scope_disk("node-a", "/tmp/disk-a");
@@ -2456,7 +2429,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_commit_clears_dirty_marks_recorded_before_scan_start() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());
        let disk = scope_disk("node-a", "/tmp/disk-a");
@@ -2477,7 +2449,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_retain_dirty_disks_within_drops_ghost_entries() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());
        let local = scope_disk("node-a", "/tmp/disk-a");
@@ -2496,7 +2467,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_spawn_refresh_recovers_from_construction_panic() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());

@@ -2563,7 +2533,6 @@ mod tests {
    }

    #[tokio::test(start_paused = true)]
-    #[serial]
    async fn test_refresh_or_join_joiner_times_out_when_leader_wedges() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());

@@ -2591,7 +2560,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_refresh_or_join_returns_cluster_total_for_dirty_subset() {
        let manager = create_isolated_manager(HybridStrategyConfig::default());

@@ -2673,7 +2641,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_config_from_env() {
        let config = HybridStrategyConfig::from_env();

@@ -2687,7 +2654,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_config_from_env_with_override() {
        temp_env::with_var(ENV_CAPACITY_SCHEDULED_INTERVAL, Some("600"), || {
            let config = HybridStrategyConfig::from_env();
@@ -1069,7 +1069,6 @@ mod tests {
    #[cfg(unix)]
    use rustfs_config::ENV_CAPACITY_FOLLOW_SYMLINKS;
    use rustfs_config::{ENV_CAPACITY_MAX_FILES_THRESHOLD, ENV_CAPACITY_SAMPLE_RATE};
-    use serial_test::serial;

    /// Reference implementation using unbounded `u128` arithmetic, clamped to
    /// `u64::MAX`, used as the source of truth for the sampling extrapolation.
@@ -1274,7 +1273,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_get_dir_size_async_nonexistent_directory() {
        let result = get_dir_size_async(Path::new("/nonexistent/path")).await;
        assert!(result.is_err());
@@ -1648,7 +1646,6 @@ mod tests {
    }

    #[tokio::test]
-    #[serial]
    async fn test_metadata_incomplete_aggregate_does_not_replace_disk_cache() {
        use std::fs::File;
        use std::io::Write;
@@ -1783,7 +1780,6 @@ mod tests {

    #[cfg(unix)]
    #[tokio::test]
-    #[serial]
    async fn test_get_dir_size_async_ignores_symlink_targets_when_follow_disabled() {
        use std::fs::File;
        use std::io::Write;
@@ -1809,7 +1805,6 @@ mod tests {

    #[cfg(unix)]
    #[tokio::test]
-    #[serial]
    async fn test_get_dir_size_async_counts_symlink_targets_when_follow_enabled() {
        use std::fs::File;
        use std::io::Write;
@@ -75,7 +75,7 @@ pub struct BucketReplicationBandwidthStats {
 }

 #[derive(Debug, Clone, Default)]
-pub struct BucketReplicationStats {
+pub struct BucketReplicationMetricsSnapshot {
    pub bucket: String,
    pub total_failed_bytes: u64,
    pub total_failed_count: u64,
@@ -107,7 +107,7 @@ pub struct BucketReplicationStats {

 #[derive(Debug, Clone, Default)]
 pub(crate) struct BucketReplicationRuntimeStats {
-    pub(crate) stats: BucketReplicationStats,
+    pub(crate) stats: BucketReplicationMetricsSnapshot,
    pub(crate) target_flows: Vec<BucketReplicationTargetFlowStats>,
 }

@@ -182,7 +182,7 @@ fn push_proxy_request_result_metrics(
    }
 }

-pub fn collect_bucket_replication_metrics(stats: &[BucketReplicationStats]) -> Vec<PrometheusMetric> {
+pub fn collect_bucket_replication_metrics(stats: &[BucketReplicationMetricsSnapshot]) -> Vec<PrometheusMetric> {
    if stats.is_empty() {
        return Vec::new();
    }
@@ -572,7 +572,7 @@ mod tests {
    #[test]
    fn test_collect_bucket_replication_metrics() {
        let stats = vec![BucketReplicationRuntimeStats {
-            stats: BucketReplicationStats {
+            stats: BucketReplicationMetricsSnapshot {
                bucket: "b1".to_string(),
                total_failed_bytes: 64,
                total_failed_count: 2,
@@ -876,7 +876,7 @@ mod tests {

    #[test]
    fn test_collect_bucket_replication_metrics_empty() {
-        let stats: Vec<BucketReplicationStats> = Vec::new();
+        let stats: Vec<BucketReplicationMetricsSnapshot> = Vec::new();
        let metrics = collect_bucket_replication_metrics(&stats);
        assert!(metrics.is_empty());
    }
@@ -48,7 +48,7 @@ pub(crate) use bucket_replication::{
    BucketReplicationTargetFlowStats, collect_bucket_replication_backlog_metrics, collect_bucket_replication_runtime_metrics,
 };
 pub use bucket_replication::{
-    BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetStats,
+    BucketReplicationBandwidthStats, BucketReplicationMetricsSnapshot, BucketReplicationTargetStats,
    collect_bucket_replication_bandwidth_metrics, collect_bucket_replication_metrics,
 };
 pub use cluster::{ClusterStats, collect_cluster_metrics};
@@ -69,7 +69,7 @@ pub use notification::{NotificationStats, collect_notification_metrics};
 pub(crate) use notification_target::{NotificationTargetRuntimeStats, collect_notification_target_runtime_metrics};
 pub use notification_target::{NotificationTargetStats, collect_notification_target_metrics};
 pub(crate) use replication::{ReplicationRuntimeStats, collect_replication_runtime_metrics};
-pub use replication::{ReplicationStats, collect_replication_metrics};
+pub use replication::{ReplicationMetricsSnapshot, collect_replication_metrics};
 pub(crate) use request::{ApiRequestMetricSupport, ApiRequestStats, collect_request_metrics};
 pub use resource::{ResourceStats, collect_resource_metrics};
 pub(crate) use scanner::{ScannerRuntimeStats, collect_scanner_runtime_metrics};
@@ -22,7 +22,7 @@ use crate::metrics::schema::replication::*;

 /// Replication statistics.
 #[derive(Debug, Clone, Default)]
-pub struct ReplicationStats {
+pub struct ReplicationMetricsSnapshot {
    /// Average number of active replication workers
    pub average_active_workers: f64,
    /// Average queued bytes since server start
@@ -54,13 +54,13 @@ pub struct ReplicationStats {
 #[derive(Debug, Clone, Default)]
 pub(crate) struct ReplicationRuntimeStats {
    pub(crate) server: String,
-    pub(crate) stats: ReplicationStats,
+    pub(crate) stats: ReplicationMetricsSnapshot,
 }

 /// Collects replication metrics from the given stats.
 ///
 /// Returns a vector of Prometheus metrics for replication statistics.
-pub fn collect_replication_metrics(stats: &ReplicationStats) -> Vec<PrometheusMetric> {
+pub fn collect_replication_metrics(stats: &ReplicationMetricsSnapshot) -> Vec<PrometheusMetric> {
    vec![
        PrometheusMetric::from_descriptor(&REPLICATION_AVERAGE_ACTIVE_WORKERS_MD, stats.average_active_workers),
        PrometheusMetric::from_descriptor(&REPLICATION_AVERAGE_QUEUED_BYTES_MD, stats.average_queued_bytes as f64),
@@ -120,7 +120,7 @@ mod tests {

    #[test]
    fn test_collect_replication_metrics() {
-        let stats = ReplicationStats {
+        let stats = ReplicationMetricsSnapshot {
            average_active_workers: 8.5,
            average_queued_bytes: 1024 * 1024 * 40,
            average_queued_count: 240,
@@ -182,7 +182,7 @@ mod tests {

    #[test]
    fn test_collect_replication_metrics_default() {
-        let stats = ReplicationStats::default();
+        let stats = ReplicationMetricsSnapshot::default();
        let metrics = collect_replication_metrics(&stats);

        assert_eq!(metrics.len(), 13);
@@ -194,7 +194,7 @@ mod tests {

    #[test]
    fn replication_stats_struct_literal_keeps_legacy_fields() {
-        let stats = ReplicationStats {
+        let stats = ReplicationMetricsSnapshot {
            average_active_workers: 1.0,
            average_queued_bytes: 2,
            average_queued_count: 3,
@@ -2811,14 +2811,14 @@ mod tests {
    #[test]
    fn replication_proxy_bucket_keys_detect_removed_buckets() {
        let previous = repl_proxy_bucket_live_keys(&[BucketReplicationRuntimeStats {
-            stats: crate::metrics::BucketReplicationStats {
+            stats: crate::metrics::BucketReplicationMetricsSnapshot {
                bucket: "photos".to_string(),
                ..Default::default()
            },
            ..Default::default()
        }]);
        let current = repl_proxy_bucket_live_keys(&[BucketReplicationRuntimeStats {
-            stats: crate::metrics::BucketReplicationStats {
+            stats: crate::metrics::BucketReplicationMetricsSnapshot {
                bucket: "logs".to_string(),
                ..Default::default()
            },
@@ -21,12 +21,12 @@
 use crate::metrics::collectors::scanner::{ScannerActiveBucketDriveStats, ScannerBucketDriveResultStats, ScannerSourceWorkStats};
 use crate::metrics::collectors::{
    ApiRequestMetricSupport, ApiRequestStats, BucketReplicationBacklogStats, BucketReplicationBandwidthStats,
-    BucketReplicationRuntimeStats, BucketReplicationStats, BucketReplicationTargetBacklogStats, BucketReplicationTargetFlowStats,
+    BucketReplicationRuntimeStats, BucketReplicationMetricsSnapshot, BucketReplicationTargetBacklogStats, BucketReplicationTargetFlowStats,
    BucketReplicationTargetStats, BucketStats, BucketUsageStats, ClusterConfigStats, ClusterHealthStats, ClusterStats,
    ClusterUsageStats, CompressionClusterStats, CpuStats, DiskStats, DriveCountStats, DriveDetailedStats,
    DriveRuntimeDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmActionTaskStats, IlmBackpressureStats,
    IlmQueueTaskStats, IlmRuntimeStats, IlmStats, IlmTaskEventStats, MemoryStats, NetworkStats, ProcessStats, ProcessStatusType,
-    ReplicationStats, ResourceStats, ScannerRuntimeStats, ScannerStats,
+    ReplicationMetricsSnapshot, ResourceStats, ScannerRuntimeStats, ScannerStats,
 };
 use crate::metrics::runtime_sources::{ObsIlmRuntimeSnapshot, bucket_monitor_handle, iam_metrics_snapshot, ilm_runtime_snapshot};
 use crate::metrics::{
@@ -266,7 +266,7 @@ fn bucket_replication_detail_from_snapshot(stats: ObsBucketReplicationStatsSnaps

    BucketReplicationRuntimeStats {
        target_flows,
-        stats: BucketReplicationStats {
+        stats: BucketReplicationMetricsSnapshot {
            bucket,
            total_failed_bytes: stats.total_failed_bytes,
            total_failed_count: stats.total_failed_count,
@@ -298,7 +298,7 @@ fn bucket_replication_detail_from_snapshot(stats: ObsBucketReplicationStatsSnaps
    }
 }

-async fn obs_site_replication_stats() -> ReplicationStats {
+async fn obs_site_replication_stats() -> ReplicationMetricsSnapshot {
    let current_data_transfer_rate = obs_bucket_replication_bandwidth_stats()
        .into_iter()
        .flatten()
@@ -306,7 +306,7 @@ async fn obs_site_replication_stats() -> ReplicationStats {
        .sum::<f64>();
    let stats = obs_replication_site_stats_snapshot(current_data_transfer_rate).await;

-    ReplicationStats {
+    ReplicationMetricsSnapshot {
        average_active_workers: stats.average_active_workers,
        average_queued_bytes: stats.average_queued_bytes,
        average_queued_count: stats.average_queued_count,
@@ -648,7 +648,7 @@ pub fn collect_bucket_replication_bandwidth_stats() -> Vec<BucketReplicationBand
 }

 /// Collect bucket and target level replication stats from the global replication runtime.
-pub async fn collect_bucket_replication_detail_stats() -> Vec<BucketReplicationStats> {
+pub async fn collect_bucket_replication_detail_stats() -> Vec<BucketReplicationMetricsSnapshot> {
    obs_bucket_replication_stats_snapshot()
        .await
        .into_iter()
@@ -662,7 +662,7 @@ pub(crate) async fn collect_bucket_replication_stats_bundle()
 }

 /// Collect site-level replication stats from the global replication runtime.
-pub async fn collect_replication_stats() -> ReplicationStats {
+pub async fn collect_replication_stats() -> ReplicationMetricsSnapshot {
    obs_site_replication_stats().await
 }

@@ -103,8 +103,7 @@ hex-simd.workspace = true

 [dev-dependencies]
 tracing-subscriber = { workspace = true, features = ["json", "env-filter", "time"] }
-serial_test = { workspace = true }
-temp-env = { workspace = true }
+temp-env = { workspace = true, features = ["async_closure"] }
 tempfile = { workspace = true }
 uuid = { workspace = true, features = ["v4", "serde", "fast-rng", "macro-diagnostics"] }
 tokio = { workspace = true, features = ["test-util", "fs", "rt-multi-thread"] }
@@ -599,10 +599,8 @@ impl ScannerConfigObjectDelete for ECStore {
 #[cfg(test)]
 mod tests {
    use super::*;
-    use serial_test::serial;

    #[tokio::test]
-    #[serial]
    async fn runtime_tier_names_serves_cached_arc_within_ttl() {
        reset_tier_name_cache_for_test();
        // The tier config manager is unconfigured in unit tests, so the
@@ -616,7 +614,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn foreground_read_guard_tracks_stream_lifetime() {
        reset_foreground_read_activity_for_test();
        assert_eq!(current_foreground_read_activity(), 0);
@@ -630,7 +627,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn foreground_read_activity_keeps_larger_signal() {
        reset_foreground_read_activity_for_test();
        let _guard = ForegroundReadGuard::new();
@@ -643,7 +639,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn scanner_runtime_guard_tracks_runtime_lifetime() {
        reset_scanner_runtime_instances_for_test();
        assert!(!scanner_runtime_initialized());
@@ -868,7 +868,6 @@ mod tests {
        SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE,
        SCANNER_SPEED, SCANNER_SUB_SYS, ScannerSpeed,
    };
-    use serial_test::serial;
    use std::collections::HashMap;
    use std::time::Duration;
    use temp_env::{with_var, with_var_unset};
@@ -916,7 +915,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn scanner_runtime_config_uses_persisted_values_when_env_is_unset() {
        let config = server_config_with_scanner(&[
            (SCANNER_SPEED, "slow"),
@@ -944,7 +942,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn scanner_runtime_config_normalizes_persisted_default_speed() {
        let config = server_config_with_scanner(&[(SCANNER_SPEED, "default")]);

@@ -960,7 +957,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn scanner_runtime_config_prefers_env_over_persisted_config() {
        let config = server_config_with_scanner(&[(SCANNER_SPEED, "slowest"), (SCANNER_CYCLE, "600")]);

@@ -977,7 +973,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn scanner_runtime_config_prefers_heal_bitrot_cycle_over_scanner_compat_config() {
        let config = server_config_with_scanner_and_heal(&[(SCANNER_BITROT_CYCLE, "3600")], &[(HEAL_BITROT_CYCLE, "off")]);

@@ -990,7 +985,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn scanner_runtime_config_marks_scanner_bitrot_cycle_as_compat_source() {
        let config = server_config_with_scanner(&[(SCANNER_BITROT_CYCLE, "3600")]);

@@ -1007,7 +1001,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn scanner_runtime_config_normalizes_persisted_default_bitrot_cycles() {
        let default_cycle = DEFAULT_HEAL_BITROT_CYCLE_SECS.to_string();
        for config in [
@@ -1032,7 +1025,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn scanner_runtime_config_validation_rejects_invalid_persisted_speed_with_env_override() {
        let config = server_config_with_scanner(&[(SCANNER_SPEED, "warp")]);

@@ -1066,7 +1058,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn scanner_runtime_config_uses_derived_delay_for_excessive_env_override() {
        let config = server_config_with_scanner(&[(SCANNER_SPEED, "slow")]);

@@ -1087,7 +1078,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn scanner_runtime_config_status_reports_value_sources() {
        let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_OBJECTS, "100"), (SCANNER_CACHE_SAVE_TIMEOUT, "5")]);

@@ -1108,7 +1098,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn applied_runtime_config_is_the_authoritative_scheduler_state() {
        let config = server_config_with_scanner(&[(SCANNER_CYCLE, "321")]);

@@ -1125,7 +1114,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn scanner_runtime_config_status_reports_persisted_pacing_overrides() {
        let config = server_config_with_scanner(&[("delay", "3.5"), ("max_wait", "7")]);

@@ -1147,7 +1135,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn scanner_runtime_config_status_prefers_env_pacing_overrides() {
        let config = server_config_with_scanner(&[("delay", "3.5"), ("max_wait", "7")]);

@@ -1169,7 +1156,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn scanner_runtime_config_status_preserves_subsecond_max_wait() {
        let config = server_config_with_scanner(&[(SCANNER_SPEED, "fast")]);

@@ -19,7 +19,6 @@ use crate::{
    ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, ScannerPutObjReader as PutObjReader,
    init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests, init_local_disks_with_instance_ctx,
 };
-use serial_test::serial;
 use std::collections::HashMap;
 use std::io::Cursor;
 use std::task::Poll;
@@ -362,7 +361,6 @@ fn test_initial_scanner_delay_uses_configured_start_delay() {
 }

 #[test]
-#[serial]
 fn test_initial_scanner_delay_uses_cycle_without_explicit_start_delay() {
    with_var(ENV_SCANNER_CYCLE, Some("120"), || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -409,7 +407,6 @@ fn test_initial_scanner_delay_keeps_delay_for_replication_without_buckets() {
 }

 #[test]
-#[serial]
 fn test_scanner_cycle_max_duration_uses_env() {
    with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("42"), || {
        assert_eq!(scanner_cycle_max_duration(), Some(Duration::from_secs(42)));
@@ -417,7 +414,6 @@ fn test_scanner_cycle_max_duration_uses_env() {
 }

 #[test]
-#[serial]
 fn test_scanner_cycle_max_duration_default_is_disabled() {
    with_var_unset(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, || {
        assert_eq!(scanner_cycle_max_duration(), None);
@@ -461,7 +457,6 @@ async fn test_scanner_cycle_budget_drop_cancels_child_without_elapsed() {
 }

 #[test]
-#[serial]
 fn test_scanner_cycle_budget_config_uses_work_budget_env() {
    with_var(ENV_SCANNER_CYCLE_MAX_OBJECTS, Some("100"), || {
        with_var(ENV_SCANNER_CYCLE_MAX_DIRECTORIES, Some("25"), || {
@@ -473,7 +468,6 @@ fn test_scanner_cycle_budget_config_uses_work_budget_env() {
 }

 #[test]
-#[serial]
 fn test_scanner_cycle_budget_config_disables_zero_work_budgets() {
    with_var(ENV_SCANNER_CYCLE_MAX_OBJECTS, Some("0"), || {
        with_var(ENV_SCANNER_CYCLE_MAX_DIRECTORIES, Some("0"), || {
@@ -516,7 +510,6 @@ fn test_scan_cycle_partial_source_maps_budget_reason() {
 }

 #[tokio::test]
-#[serial]
 async fn test_mark_scan_cycle_idle_clears_published_cycle_state() {
    let mut cycle_info = CurrentCycle {
        current: 12,
@@ -545,7 +538,6 @@ async fn test_mark_scan_cycle_idle_clears_published_cycle_state() {
 }

 #[tokio::test]
-#[serial]
 async fn scanner_cycle_metrics_guard_covers_published_first_cycle_lifetime() {
    let cycle_started = Utc::now() - chrono::Duration::seconds(5);
    let mut cycle_info = CurrentCycle {
@@ -572,7 +564,6 @@ async fn scanner_cycle_metrics_guard_covers_published_first_cycle_lifetime() {
 }

 #[tokio::test]
-#[serial]
 async fn scanner_cycle_metrics_guard_keeps_active_cycle_published_during_finalization() {
    let mut cycle_info = CurrentCycle {
        current: 12,
@@ -597,7 +588,6 @@ async fn scanner_cycle_metrics_guard_keeps_active_cycle_published_during_finaliz
 }

 #[tokio::test]
-#[serial]
 async fn scanner_cycle_metrics_guard_drop_clears_activity() {
    let guard = ScannerCycleMetricsGuard::new(CurrentCycle {
        current: 12,
@@ -615,7 +605,6 @@ async fn scanner_cycle_metrics_guard_drop_clears_activity() {
 }

 #[tokio::test]
-#[serial]
 async fn run_data_scanner_cycle_publishes_activity_for_owner_lifetime() {
    let (_temp_dir, store) = setup_scanner_cycle_store().await;
    let ctx = CancellationToken::new();
@@ -666,7 +655,6 @@ async fn run_data_scanner_cycle_publishes_activity_for_owner_lifetime() {
 }

 #[tokio::test]
-#[serial]
 async fn test_finalize_partial_scan_cycle_advances_and_persists_counter() {
    let store = Arc::new(MemoryConfigStore::default());
    let ctx = CancellationToken::new();
@@ -702,7 +690,6 @@ async fn test_finalize_partial_scan_cycle_advances_and_persists_counter() {
 }

 #[tokio::test]
-#[serial]
 async fn scanner_cycle_recovers_to_newer_durable_cache_floor() {
    let store = Arc::new(MemoryConfigStore::default());
    let ctx = CancellationToken::new();
@@ -742,7 +729,6 @@ async fn scanner_cycle_recovers_to_newer_durable_cache_floor() {
 }

 #[tokio::test]
-#[serial]
 async fn scanner_cycle_rejects_invalid_cache_floor() {
    let store = Arc::new(MemoryConfigStore::default());
    let ctx = CancellationToken::new();
@@ -1013,7 +999,6 @@ async fn scanner_usage_floor_fails_closed_on_corrupt_or_exhausted_usage_state()
 }

 #[tokio::test]
-#[serial]
 async fn scanner_usage_backup_uses_durable_cycle_cadence_across_tasks() {
    let store = Arc::new(MemoryConfigStore::default());
    let ctx = CancellationToken::new();
@@ -1087,7 +1072,6 @@ fn scanner_cycle_advance_fails_before_reserved_exhausted_value() {
 }

 #[tokio::test]
-#[serial]
 async fn test_finalize_partial_scan_cycle_reports_persist_failure() {
    let store = Arc::new(MemoryConfigStore::default());
    let ctx = CancellationToken::new();
@@ -1111,7 +1095,6 @@ async fn test_finalize_partial_scan_cycle_reports_persist_failure() {
 }

 #[tokio::test]
-#[serial]
 async fn test_persist_scanner_cycle_state_reconciles_newer_winner() {
    let store = Arc::new(MemoryConfigStore::default());
    let ctx = CancellationToken::new();
@@ -1534,7 +1517,6 @@ async fn test_usage_save_route_barrier_prevents_missing_snapshot_creation() {
 }

 #[tokio::test]
-#[serial]
 async fn test_usage_route_barrier_precedes_durable_reconciliation() {
    let store = Arc::new(MemoryConfigStore::default());
    let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str());
@@ -1562,7 +1544,6 @@ async fn test_usage_route_barrier_precedes_durable_reconciliation() {
 }

 #[tokio::test]
-#[serial]
 async fn test_deferred_usage_save_keeps_last_real_save_metric() {
    let metrics = global_metrics();
    metrics.record_scanner_usage_save_result(ScannerUsageSaveResult::Success);
@@ -2626,7 +2607,6 @@ fn scanner_cycle_cache_floor_stays_pending_during_deferred_usage_publication() {
 }

 #[test]
-#[serial]
 fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() {
    crate::scanner_io::clear_dirty_usage_bucket("photos");
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -2653,7 +2633,6 @@ fn finalizing_a_saved_cycle_acknowledges_its_exact_dirty_snapshot() {
 }

 #[test]
-#[serial]
 fn finalizing_a_deferred_usage_save_keeps_dirty_work_pending() {
    crate::scanner_io::clear_dirty_usage_bucket("photos");
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -2696,7 +2675,6 @@ async fn scanner_cycle_keeps_remote_pending_acknowledgement() {
 }

 #[test]
-#[serial]
 fn finalizing_an_already_durable_cycle_acknowledges_its_exact_dirty_snapshot() {
    crate::scanner_io::clear_dirty_usage_bucket("photos");
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -2711,7 +2689,6 @@ fn finalizing_an_already_durable_cycle_acknowledges_its_exact_dirty_snapshot() {
 }

 #[test]
-#[serial]
 fn finalizing_a_prior_same_cycle_snapshot_keeps_new_dirty_work_pending() {
    crate::scanner_io::clear_dirty_usage_bucket("photos");
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -2727,7 +2704,6 @@ fn finalizing_a_prior_same_cycle_snapshot_keeps_new_dirty_work_pending() {
 }

 #[test]
-#[serial]
 fn finalizing_a_durable_superseded_snapshot_keeps_dirty_work_pending() {
    crate::scanner_io::clear_dirty_usage_bucket("photos");
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -2743,7 +2719,6 @@ fn finalizing_a_durable_superseded_snapshot_keeps_dirty_work_pending() {
 }

 #[test]
-#[serial]
 fn data_usage_persist_wait_covers_cache_retries_and_backup() {
    with_var(rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, Some("7"), || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -2796,7 +2771,6 @@ async fn maintenance_feature_inspection_preserves_base_cycle_after_timeout() {
 }

 #[tokio::test(start_paused = true)]
-#[serial]
 async fn stable_maintenance_detection_preserves_base_cycle_after_timeout() {
    let ctx = CancellationToken::new();

@@ -2862,7 +2836,6 @@ async fn maintenance_feature_inspection_stops_on_cancellation() {
 }

 #[test]
-#[serial]
 fn test_cycle_interval_prefers_explicit_cycle_override() {
    with_var(ENV_SCANNER_SPEED, Some("slowest"), || {
        with_var(ENV_SCANNER_CYCLE, Some("42"), || {
@@ -2872,7 +2845,6 @@ fn test_cycle_interval_prefers_explicit_cycle_override() {
 }

 #[test]
-#[serial]
 fn test_cycle_interval_prefers_explicit_cycle_over_default_cycle() {
    let _guard = ScannerDefaultCycleGuard::set(TEST_DEFAULT_SCANNER_CYCLE_SECS);

@@ -2882,7 +2854,6 @@ fn test_cycle_interval_prefers_explicit_cycle_over_default_cycle() {
 }

 #[test]
-#[serial]
 fn test_cycle_interval_uses_scanner_default_speed_override_when_unconfigured() {
    let _guard = ScannerDefaultSpeedGuard::set(ScannerSpeed::Slowest);

@@ -2892,7 +2863,6 @@ fn test_cycle_interval_uses_scanner_default_speed_override_when_unconfigured() {
 }

 #[test]
-#[serial]
 fn test_cycle_interval_prefers_explicit_speed_over_default_speed_override() {
    let _guard = ScannerDefaultSpeedGuard::set(ScannerSpeed::Slowest);

@@ -2910,7 +2880,6 @@ fn test_cycle_interval_prefers_explicit_speed_over_default_speed_override() {
 }

 #[test]
-#[serial]
 fn test_cycle_interval_uses_default_cycle_override_when_unconfigured() {
    let _guard = ScannerDefaultCycleGuard::set(TEST_DEFAULT_SCANNER_CYCLE_SECS);

@@ -3074,7 +3043,6 @@ fn scanner_cycle_wait_plan_drives_growth_resets_and_bitrot_cap() {
 }

 #[test]
-#[serial]
 fn scanner_cycle_schedule_status_reports_effective_backoff() {
    record_scanner_cycle_schedule(Duration::from_millis(86_400_001), true, 2_048, true, 7);

@@ -3354,7 +3322,6 @@ fn dirty_usage_wakes_are_disabled_for_explicit_cycle_policy() {
 }

 #[test]
-#[serial]
 fn clean_idle_cap_preserves_default_bitrot_coverage_window() {
    let config = ScannerRuntimeConfig {
        bitrot_cycle: Some(Duration::from_secs(30 * 24 * 60 * 60)),
@@ -3384,7 +3351,6 @@ fn clean_idle_cap_allows_policy_max_when_bitrot_is_disabled() {
 }

 #[test]
-#[serial]
 fn clean_idle_cap_never_shortens_the_base_cycle() {
    let config = ScannerRuntimeConfig {
        bitrot_cycle: Some(Duration::from_secs(60)),
@@ -3398,7 +3364,6 @@ fn clean_idle_cap_never_shortens_the_base_cycle() {
 }

 #[test]
-#[serial]
 fn test_cycle_interval_keeps_default_cycle_with_explicit_speed() {
    let _guard = ScannerDefaultCycleGuard::set(TEST_DEFAULT_SCANNER_CYCLE_SECS);

@@ -3416,7 +3381,6 @@ fn test_cycle_interval_keeps_default_cycle_with_explicit_speed() {
 }

 #[test]
-#[serial]
 fn test_cycle_interval_prefers_explicit_start_delay_over_default_cycle() {
    let _guard = ScannerDefaultCycleGuard::set(TEST_DEFAULT_SCANNER_CYCLE_SECS);

@@ -3430,7 +3394,6 @@ fn test_cycle_interval_prefers_explicit_start_delay_over_default_cycle() {
 }

 #[test]
-#[serial]
 fn test_cycle_interval_supports_minio_speed_alias() {
    with_var_unset(ENV_SCANNER_SPEED, || {
        with_var_unset(ENV_SCANNER_CYCLE, || {
@@ -3444,7 +3407,6 @@ fn test_cycle_interval_supports_minio_speed_alias() {
 }

 #[test]
-#[serial]
 fn test_cycle_interval_supports_minio_cycle_alias() {
    with_var_unset(ENV_SCANNER_CYCLE, || {
        with_var_unset(ENV_SCANNER_START_DELAY_SECS, || {
@@ -3464,7 +3426,6 @@ fn test_randomized_cycle_delay_handles_small_start_delay() {
 }

 #[tokio::test]
-#[serial]
 async fn test_wait_for_next_scanner_cycle_wakes_for_dirty_usage() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();

@@ -3490,7 +3451,6 @@ async fn test_wait_for_next_scanner_cycle_wakes_for_dirty_usage() {
 }

 #[tokio::test]
-#[serial]
 async fn test_wait_for_next_scanner_cycle_sees_unattempted_dirty_usage() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let dirty_generation = crate::scanner_io::dirty_usage_generation();
@@ -3512,7 +3472,6 @@ async fn test_wait_for_next_scanner_cycle_sees_unattempted_dirty_usage() {
 }

 #[tokio::test(start_paused = true)]
-#[serial]
 async fn test_wait_for_next_scanner_cycle_retries_stable_dirty_usage_on_timer() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -3534,7 +3493,6 @@ async fn test_wait_for_next_scanner_cycle_retries_stable_dirty_usage_on_timer()
 }

 #[tokio::test(start_paused = true)]
-#[serial]
 async fn test_wait_for_next_scanner_cycle_can_defer_dirty_wakes_until_timer() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -3553,7 +3511,6 @@ async fn test_wait_for_next_scanner_cycle_can_defer_dirty_wakes_until_timer() {
 }

 #[tokio::test]
-#[serial]
 async fn test_wait_for_next_scanner_cycle_wakes_for_repeated_dirty_bucket() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    crate::scanner_io::record_dirty_usage_bucket("photos");
@@ -3579,7 +3536,6 @@ async fn test_wait_for_next_scanner_cycle_wakes_for_repeated_dirty_bucket() {
 }

 #[tokio::test]
-#[serial]
 async fn test_wait_for_next_scanner_cycle_reschedules_for_runtime_config() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let observed_generation = crate::runtime_config::scanner_runtime_config_generation();
@@ -3607,7 +3563,6 @@ async fn test_wait_for_next_scanner_cycle_reschedules_for_runtime_config() {
 }

 #[tokio::test]
-#[serial]
 async fn test_wait_for_next_scanner_cycle_reschedules_for_maintenance_change() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let observed_generation = crate::scanner_io::scanner_maintenance_generation();
@@ -3851,7 +3806,6 @@ fn scanner_activity_after_a_cycle_restores_the_base_interval() {
 }

 #[tokio::test(start_paused = true)]
-#[serial]
 async fn distributed_clean_idle_wait_wakes_at_base_interval_for_remote_activity() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -3879,7 +3833,6 @@ async fn distributed_clean_idle_wait_wakes_at_base_interval_for_remote_activity(
 }

 #[tokio::test(start_paused = true)]
-#[serial]
 async fn superseded_retry_wait_defers_dirty_cluster_activity_until_timer() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -3907,7 +3860,6 @@ async fn superseded_retry_wait_defers_dirty_cluster_activity_until_timer() {
 }

 #[tokio::test(start_paused = true)]
-#[serial]
 async fn distributed_clean_idle_wait_blocks_backoff_for_unpropagated_maintenance() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -3934,7 +3886,6 @@ async fn distributed_clean_idle_wait_blocks_backoff_for_unpropagated_maintenance
 }

 #[tokio::test(start_paused = true)]
-#[serial]
 async fn distributed_clean_idle_wait_fails_closed_when_a_peer_is_unverifiable() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -3961,7 +3912,6 @@ async fn distributed_clean_idle_wait_fails_closed_when_a_peer_is_unverifiable()
 }

 #[tokio::test(start_paused = true)]
-#[serial]
 async fn distributed_clean_idle_wait_keeps_the_extended_deadline_when_peers_are_clean() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -3989,7 +3939,6 @@ async fn distributed_clean_idle_wait_keeps_the_extended_deadline_when_peers_are_
 }

 #[tokio::test(start_paused = true)]
-#[serial]
 async fn scanner_activity_probe_wait_is_cancellation_aware() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -4020,7 +3969,6 @@ async fn scanner_activity_probe_wait_is_cancellation_aware() {
 }

 #[tokio::test(start_paused = true)]
-#[serial]
 async fn scanner_activity_probe_wait_stops_after_leader_lock_loss() {
    crate::scanner_io::clear_dirty_usage_buckets_for_tests();
    let ctx = CancellationToken::new();
@@ -4052,7 +4000,6 @@ async fn scanner_activity_probe_wait_stops_after_leader_lock_loss() {
 }

 #[test]
-#[serial]
 fn test_get_cycle_scan_mode_runs_deep_until_selection_window_completes() {
    with_var(ENV_SCANNER_BITROT_CYCLE_SECS, Some("3600"), || {
        let mode = get_cycle_scan_mode(10, 0, Some(Utc::now()), bitrot_scan_cycle());
@@ -4061,7 +4008,6 @@ fn test_get_cycle_scan_mode_runs_deep_until_selection_window_completes() {
 }

 #[test]
-#[serial]
 fn test_get_cycle_scan_mode_respects_elapsed_bitrot_cycle() {
    with_var(ENV_SCANNER_BITROT_CYCLE_SECS, Some("3600"), || {
        let recent = Utc::now() - chrono::Duration::minutes(30);
@@ -4073,7 +4019,6 @@ fn test_get_cycle_scan_mode_respects_elapsed_bitrot_cycle() {
 }

 #[test]
-#[serial]
 fn test_get_cycle_scan_mode_can_disable_periodic_deep_scan() {
    with_var(ENV_SCANNER_BITROT_CYCLE_SECS, Some("off"), || {
        assert_eq!(get_cycle_scan_mode(1, 0, None, bitrot_scan_cycle()), HealScanMode::Normal);
@@ -4081,7 +4026,6 @@ fn test_get_cycle_scan_mode_can_disable_periodic_deep_scan() {
 }

 #[test]
-#[serial]
 fn test_background_heal_info_for_scan_start_marks_deep_active() {
    let now = Utc::now();
    let info =
@@ -4094,7 +4038,6 @@ fn test_background_heal_info_for_scan_start_marks_deep_active() {
 }

 #[test]
-#[serial]
 fn test_background_heal_info_for_scan_start_keeps_deep_window_start() {
    with_var_unset(ENV_SCANNER_BITROT_CYCLE_SECS, || {
        let started_at = Utc::now();
@@ -18,7 +18,6 @@ use super::*;
 use crate::storage_api::VersionPurgeStatusType;
 use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
 use rustfs_filemeta::{FileInfo, FileMeta};
-use serial_test::serial;
 use std::io::Write;
 #[cfg(unix)]
 use std::os::unix::fs::{PermissionsExt, symlink};
@@ -356,7 +355,6 @@ impl Drop for TestGuard {
 }

 #[tokio::test]
-#[serial]
 async fn test_should_skip_failed_respects_ttl() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
@@ -378,7 +376,6 @@ async fn test_should_skip_failed_respects_ttl() {
 }

 #[tokio::test]
-#[serial]
 async fn test_record_failed_ttl_zero_noop() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(0, 100, &mut scanner, temp_dir);
@@ -467,7 +464,6 @@ fn test_should_account_replication_stats_only_for_live_object_versions() {
 }

 #[tokio::test]
-#[serial]
 async fn test_heal_replication_only_queues_pending_null_deletes() {
    async fn replication_skipped_count() -> u64 {
        global_metrics()
@@ -716,7 +712,6 @@ async fn test_scanner_heal_admission_accounting_maps_deep_scan_to_bitrot() {
 }

 #[test]
-#[serial]
 fn test_excessive_version_alert_thresholds_use_env() {
    with_var(rustfs_config::ENV_SCANNER_ALERT_EXCESS_VERSIONS, Some("3"), || {
        with_var(rustfs_config::ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, Some("100"), || {
@@ -731,7 +726,6 @@ fn test_excessive_version_alert_thresholds_use_env() {
 }

 #[test]
-#[serial]
 fn test_excessive_folders_threshold_uses_env() {
    with_var(rustfs_config::ENV_SCANNER_ALERT_EXCESS_FOLDERS, Some("3"), || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -741,7 +735,6 @@ fn test_excessive_folders_threshold_uses_env() {
 }

 #[test]
-#[serial]
 fn test_excessive_folders_threshold_default_supports_pbs_layout() {
    with_var_unset(rustfs_config::ENV_SCANNER_ALERT_EXCESS_FOLDERS, || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -751,7 +744,6 @@ fn test_excessive_folders_threshold_default_supports_pbs_layout() {
 }

 #[test]
-#[serial]
 fn test_scanner_yield_every_n_objects_uses_env() {
    with_var(rustfs_config::ENV_SCANNER_YIELD_EVERY_N_OBJECTS, Some("32"), || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -761,7 +753,6 @@ fn test_scanner_yield_every_n_objects_uses_env() {
 }

 #[test]
-#[serial]
 fn test_scanner_yield_every_n_objects_uses_default() {
    with_var_unset(rustfs_config::ENV_SCANNER_YIELD_EVERY_N_OBJECTS, || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -888,7 +879,6 @@ fn test_order_folders_for_resume_reports_stale_hint() {
 }

 #[tokio::test]
-#[serial]
 async fn test_record_failed_prunes_to_max_entries() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(1000, 2, &mut scanner, temp_dir);
@@ -920,7 +910,6 @@ async fn test_record_failed_prunes_to_max_entries() {
 }

 #[tokio::test]
-#[serial]
 async fn test_prune_failed_objects_cache_drops_expired() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(5, 10, &mut scanner, temp_dir);
@@ -944,7 +933,6 @@ async fn test_prune_failed_objects_cache_drops_expired() {
 }

 #[tokio::test]
-#[serial]
 async fn test_prune_failed_objects_max_zero_keeps_fresh() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 0, &mut scanner, temp_dir);
@@ -1701,7 +1689,6 @@ async fn test_heal_actions_returns_actual_size_without_inline_heal() {
 }

 #[tokio::test]
-#[serial]
 #[cfg(unix)]
 async fn test_scan_folder_skips_unreadable_child_directory() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
@@ -1734,7 +1721,6 @@ async fn test_scan_folder_skips_unreadable_child_directory() {
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -1813,7 +1799,6 @@ async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_folder_xl_meta_named_directory_uses_namespace_descent() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -1859,7 +1844,6 @@ async fn test_scan_folder_xl_meta_named_directory_uses_namespace_descent() {
 }

 #[tokio::test(flavor = "current_thread")]
-#[serial]
 async fn test_scan_folder_corrupt_xl_meta_stops_erasure_data_dir_descent() {
    let logs = CapturedLogs::default();
    let subscriber = tracing_subscriber::fmt()
@@ -2021,7 +2005,6 @@ async fn test_scan_folder_corrupt_xl_meta_stops_erasure_data_dir_descent() {
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_folder_missing_xl_meta_stops_erasure_data_dir_descent() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -2099,7 +2082,6 @@ async fn test_scan_folder_missing_xl_meta_stops_erasure_data_dir_descent() {
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_folder_uuid_namespace_part_name_directory_is_not_data_dir() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -2161,7 +2143,6 @@ async fn test_scan_folder_uuid_namespace_part_name_directory_is_not_data_dir() {
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_folder_non_erasure_metadata_keeps_namespace_descent() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -2203,7 +2184,6 @@ async fn test_scan_folder_non_erasure_metadata_keeps_namespace_descent() {
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_folder_compacted_parent_sends_partial_update() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -2245,7 +2225,6 @@ async fn test_scan_folder_compacted_parent_sends_partial_update() {
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_data_folder_cancelled_before_scan_clears_current_path() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2290,7 +2269,6 @@ async fn test_scan_data_folder_cancelled_before_scan_clears_current_path() {
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_data_folder_returns_partial_cache_on_budget_cancel() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
@@ -2346,7 +2324,6 @@ async fn test_scan_data_folder_returns_partial_cache_on_budget_cancel() {
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_data_folder_reports_invalid_checkpoint_ignored_once() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2391,7 +2368,6 @@ async fn test_scan_data_folder_reports_invalid_checkpoint_ignored_once() {
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_data_folder_resume_hint_prioritizes_next_existing_folder() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2465,7 +2441,6 @@ async fn test_scan_data_folder_resume_hint_prioritizes_next_existing_folder() {
 }

 #[tokio::test]
-#[serial]
 async fn scan_data_folder_missing_bucket_returns_partial() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2517,7 +2492,6 @@ async fn scan_data_folder_missing_bucket_returns_partial() {
 }

 #[tokio::test]
-#[serial]
 async fn scan_data_folder_missing_scan_root_returns_partial() {
    let (scanner, temp_dir) = build_test_scanner().await;
    tokio::fs::remove_dir_all(&temp_dir)
@@ -2563,7 +2537,6 @@ async fn scan_data_folder_missing_scan_root_returns_partial() {
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_data_folder_resume_hint_orders_across_new_and_existing_folders() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2632,7 +2605,6 @@ async fn test_scan_data_folder_resume_hint_orders_across_new_and_existing_folder
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_data_folder_partial_object_budget_accumulates_progress() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2715,7 +2687,6 @@ async fn test_scan_data_folder_partial_object_budget_accumulates_progress() {
 }

 #[tokio::test]
-#[serial]
 async fn test_partial_compacted_entry_does_not_carry_children() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2761,7 +2732,6 @@ async fn test_partial_compacted_entry_does_not_carry_children() {
 }

 #[tokio::test]
-#[serial]
 async fn test_partial_entry_does_not_carry_missing_old_child() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2794,7 +2764,6 @@ async fn test_partial_entry_does_not_carry_missing_old_child() {
 }

 #[tokio::test]
-#[serial]
 async fn test_legacy_windows_cache_rebuilds_and_round_trips_portable_keys() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2861,7 +2830,6 @@ async fn test_legacy_windows_cache_rebuilds_and_round_trips_portable_keys() {
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_data_folder_success_clears_resume_hint() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2904,7 +2872,6 @@ async fn test_scan_data_folder_success_clears_resume_hint() {
 }

 #[tokio::test]
-#[serial]
 async fn test_scan_data_folder_keeps_unresolved_objects_partial() {
    let (scanner, temp_dir) = build_test_scanner().await;
    let _guard = TestGuard {
@@ -2951,7 +2918,6 @@ async fn test_scan_data_folder_keeps_unresolved_objects_partial() {
 }

 #[tokio::test]
-#[serial]
 #[cfg(unix)]
 async fn test_scan_folder_ignores_symlinked_child_directory() {
    let (mut scanner, temp_dir) = build_test_scanner().await;
@@ -27,7 +27,6 @@ use crate::{
    init_local_disks_with_instance_ctx, new_disk, path2_bucket_object_with_base_path,
 };
 use rustfs_filemeta::FileInfo;
-use serial_test::serial;
 use temp_env::with_var;
 use time::OffsetDateTime;
 use uuid::Uuid;
@@ -103,7 +102,6 @@ async fn setup_two_pool_scanner_store() -> (tempfile::TempDir, Arc<ECStore>) {
 }

 #[tokio::test]
-#[serial]
 async fn scanner_cache_locks_block_same_source_workers() {
    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
    let set = &store.pools[0].disk_set[0];
@@ -130,7 +128,6 @@ async fn scanner_cache_locks_block_same_source_workers() {
 }

 #[tokio::test]
-#[serial]
 async fn scanner_cache_locks_allow_cross_source_workers() {
    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
    let first_set = &store.pools[0].disk_set[0];
@@ -149,7 +146,6 @@ async fn scanner_cache_locks_allow_cross_source_workers() {
 }

 #[tokio::test]
-#[serial]
 async fn scanner_cycle_is_deferred_while_rebalance_is_active() {
    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
    let mut pool_stats = vec![EcstoreRebalanceStats::default(); store.pools.len()];
@@ -185,7 +181,6 @@ async fn scanner_cycle_is_deferred_while_rebalance_is_active() {
 }

 #[tokio::test]
-#[serial]
 async fn scanner_cycle_is_deferred_while_terminal_decommission_is_blocked() {
    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
    for decommission in [
@@ -230,7 +225,6 @@ async fn data_usage_publish_fails_when_receiver_is_closed() {
 }

 #[tokio::test]
-#[serial]
 async fn multi_pool_scanner_cycle_publishes_combined_usage() {
    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
    let bucket = format!("scanner-union-{}", Uuid::new_v4().simple());
@@ -278,7 +272,6 @@ async fn multi_pool_scanner_cycle_publishes_combined_usage() {
 }

 #[tokio::test]
-#[serial]
 async fn multi_pool_scanner_cycle_zero_fills_bucket_absent_from_first_pool() {
    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
    let bucket = format!("scanner-second-pool-{}", Uuid::new_v4().simple());
@@ -366,7 +359,6 @@ fn object_lock_config_enabled_accepts_enabled_only() {
 }

 #[test]
-#[serial]
 fn dirty_usage_snapshot_clear_preserves_newer_generation() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -381,7 +373,6 @@ fn dirty_usage_snapshot_clear_preserves_newer_generation() {
 }

 #[test]
-#[serial]
 fn dirty_usage_generation_acknowledgement_preserves_newer_mutations() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -407,7 +398,6 @@ fn dirty_usage_generation_acknowledgement_preserves_newer_mutations() {
 }

 #[test]
-#[serial]
 fn dirty_usage_generation_acknowledgement_rejects_stale_process_and_future_generation() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -437,7 +427,6 @@ fn dirty_usage_generation_acknowledgement_rejects_stale_process_and_future_gener
 }

 #[test]
-#[serial]
 fn dirty_usage_snapshot_detects_uncovered_generation() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -462,7 +451,6 @@ fn generation_saturates_instead_of_wrapping() {
 }

 #[test]
-#[serial]
 fn dirty_usage_snapshot_clears_a_stably_absent_bucket_after_durable_save() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -484,7 +472,6 @@ fn dirty_usage_snapshot_clears_a_stably_absent_bucket_after_durable_save() {
 }

 #[test]
-#[serial]
 fn dirty_usage_snapshot_preserves_an_absent_bucket_recorded_after_listing_started() {
    clear_dirty_usage_buckets_for_tests();
    let generation_before_bucket_list = dirty_usage_generation();
@@ -499,7 +486,6 @@ fn dirty_usage_snapshot_preserves_an_absent_bucket_recorded_after_listing_starte
 }

 #[test]
-#[serial]
 fn deleting_a_clean_bucket_invalidates_an_inflight_usage_snapshot() {
    clear_dirty_usage_buckets_for_tests();
    let snapshot = snapshot_dirty_usage_buckets(&[bucket_info("photos")], dirty_usage_generation());
@@ -513,7 +499,6 @@ fn deleting_a_clean_bucket_invalidates_an_inflight_usage_snapshot() {
 }

 #[test]
-#[serial]
 fn deleting_a_bucket_during_listing_invalidates_the_resulting_usage_snapshot() {
    clear_dirty_usage_buckets_for_tests();
    let generation_before_bucket_list = dirty_usage_generation();
@@ -527,7 +512,6 @@ fn deleting_a_bucket_during_listing_invalidates_the_resulting_usage_snapshot() {
 }

 #[test]
-#[serial]
 fn scanner_maintenance_change_advances_generation_and_marks_usage_dirty() {
    clear_dirty_usage_buckets_for_tests();
    let generation = scanner_maintenance_generation();
@@ -540,7 +524,6 @@ fn scanner_maintenance_change_advances_generation_and_marks_usage_dirty() {
 }

 #[test]
-#[serial]
 fn dirty_usage_clear_excludes_failed_buckets() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -572,7 +555,6 @@ fn dirty_usage_clear_plan_excludes_cache_save_failures() {
 }

 #[test]
-#[serial]
 fn dirty_usage_is_acknowledged_only_after_durable_usage_confirmation() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -590,7 +572,6 @@ fn dirty_usage_is_acknowledged_only_after_durable_usage_confirmation() {
 }

 #[test]
-#[serial]
 fn clear_dirty_usage_bucket_removes_deleted_bucket_marker() {
    clear_dirty_usage_buckets_for_tests();
    record_dirty_usage_bucket("photos");
@@ -917,35 +898,30 @@ async fn bucket_cache_pending_heal_reaches_cycle_maintenance_state() {
 }

 #[test]
-#[serial]
 fn scanner_concurrency_limit_preserves_available_when_unconfigured() {
    crate::reset_foreground_read_activity_for_test();
    assert_eq!(scanner_concurrency_limit(0, 4), 4);
 }

 #[test]
-#[serial]
 fn scanner_concurrency_limit_caps_to_configured_value() {
    crate::reset_foreground_read_activity_for_test();
    assert_eq!(scanner_concurrency_limit(2, 4), 2);
 }

 #[test]
-#[serial]
 fn scanner_concurrency_limit_never_exceeds_available_work() {
    crate::reset_foreground_read_activity_for_test();
    assert_eq!(scanner_concurrency_limit(8, 4), 4);
 }

 #[test]
-#[serial]
 fn scanner_concurrency_limit_handles_no_available_work() {
    crate::reset_foreground_read_activity_for_test();
    assert_eq!(scanner_concurrency_limit(2, 0), 0);
 }

 #[test]
-#[serial]
 fn scanner_concurrency_limit_yields_to_foreground_reads() {
    crate::reset_foreground_read_activity_for_test();
    crate::set_foreground_read_activity(8);
@@ -955,7 +931,6 @@ fn scanner_concurrency_limit_yields_to_foreground_reads() {
 }

 #[test]
-#[serial]
 fn scanner_concurrency_limit_yields_to_streaming_reads() {
    crate::reset_foreground_read_activity_for_test();
    let _guard = crate::ForegroundReadGuard::new();
@@ -979,7 +954,6 @@ fn increment_atomic_usize_saturates_at_max() {
 }

 #[test]
-#[serial]
 fn scanner_max_concurrent_set_scans_uses_env_cap() {
    with_var(ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, Some("2"), || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -989,7 +963,6 @@ fn scanner_max_concurrent_set_scans_uses_env_cap() {
 }

 #[test]
-#[serial]
 fn scanner_max_concurrent_disk_scans_uses_env_cap() {
    with_var(ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, Some("1"), || {
        crate::runtime_config::refresh_scanner_runtime_config_for_tests();
@@ -258,7 +258,6 @@ impl SleepTimer {
 #[cfg(test)]
 mod tests {
    use super::*;
-    use serial_test::serial;
    use temp_env::{with_var, with_var_unset};

    struct ScannerDefaultSpeedGuard;
@@ -326,7 +325,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn test_refresh_from_env_applies_speed_and_idle_mode_for_next_cycle() {
        let prev_mode = SCANNER_IDLE_MODE.load(Ordering::Relaxed);
        SCANNER_IDLE_MODE.store(true, Ordering::Relaxed);
@@ -346,7 +344,6 @@ mod tests {
    }

    #[test]
-    #[serial]
    fn test_refresh_from_env_uses_default_speed_override_when_speed_unset() {
        let _guard = ScannerDefaultSpeedGuard::set(ScannerSpeed::Slowest);
        let s = DynamicSleeper::new(ScannerSpeed::Default);
@@ -362,7 +359,6 @@ mod tests {
    }

    #[tokio::test(start_paused = true)]
-    #[serial]
    async fn test_fastest_never_sleeps() {
        let prev_mode = SCANNER_IDLE_MODE.load(Ordering::Relaxed);
        SCANNER_IDLE_MODE.store(true, Ordering::Relaxed);
@@ -376,7 +372,6 @@ mod tests {
    }

    #[tokio::test(start_paused = true)]
-    #[serial]
    async fn test_idle_mode_off_skips_sleep() {
        let prev_mode = SCANNER_IDLE_MODE.load(Ordering::Relaxed);
        SCANNER_IDLE_MODE.store(false, Ordering::Relaxed);
@@ -14,7 +14,6 @@

 #![recursion_limit = "256"]

-use futures::FutureExt;
 use rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT;
 use rustfs_scanner::scanner_folder::ScannerItem;
 use rustfs_scanner::scanner_io::ScannerIODisk;
@@ -23,10 +22,8 @@ use rustfs_scanner::{
    scanner::init_data_scanner,
 };
 use s3s::dto::RestoreRequest;
-use serial_test::serial;
 use std::{
    collections::HashMap,
-    env,
    path::{Path, PathBuf},
    sync::{Arc, Once, OnceLock},
    time::Duration,
@@ -535,31 +532,19 @@ async fn wait_for_transition(ecstore: &Arc<ECStore>, bucket: &str, object: &str,
    }
 }

-// SAFETY: this helper is used only by `#[serial]` tests and runs under the single-threaded Tokio
-// runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the
-// `env::set_var` / `env::remove_var` window.
-#[allow(unsafe_code)]
+// Run `test_fn` with `ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT`
+// set to `"1"` for its duration. `temp_env` serializes environment mutations
+// globally, preventing data races when multiple tests run in parallel.
 async fn with_forced_immediate_enqueue_timeout<F, Fut>(test_fn: F)
 where
    F: FnOnce() -> Fut,
    Fut: std::future::Future<Output = ()>,
 {
-    let original = env::var_os(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT);
-    unsafe {
-        env::set_var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT, "1");
-    }
-    let result = std::panic::AssertUnwindSafe(test_fn()).catch_unwind().await;
-    match original {
-        Some(value) => unsafe {
-            env::set_var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT, value);
-        },
-        None => unsafe {
-            env::remove_var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT);
-        },
-    }
-    if let Err(err) = result {
-        std::panic::resume_unwind(err);
-    }
+    temp_env::async_with_vars(
+        [(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT, Some("1"))],
+        test_fn(),
+    )
+    .await;
 }

 mod serial_tests {
@@ -592,7 +577,6 @@ mod serial_tests {
    ///     body (GET won) or a clean object/version-not-found (expiry won). A
    ///     tier-fetch failure -- the #3491 symptom -- is never tolerated.
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-2)"]
    async fn test_expire_transitioned_object_never_races_concurrent_get() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -738,7 +722,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"]
    async fn rejected_transition_candidate_is_recovered_from_persisted_delete_journal() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -825,7 +808,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"]
    async fn cancelled_before_cleanup_store_resolution_persists_journal() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -919,7 +901,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"]
    async fn rejected_transition_cleanup_durability_matrix() {
        #[derive(Clone, Copy)]
@@ -1059,7 +1040,6 @@ mod serial_tests {
    }

    #[test]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    fn test_transition_and_restore_flows() {
        std::thread::Builder::new()
@@ -1385,7 +1365,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_scanner_enqueues_free_version_cleanup_for_stale_transitioned_object() {
        let (disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -1446,7 +1425,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_scanner_cleanup_still_works_after_immediate_compensation_transition() {
        let (disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -1504,7 +1482,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_existing_object_backfill_is_idempotent_after_immediate_compensation_transition() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -1547,7 +1524,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "FAILING on main: excluded from the serial ILM lane pending a fix, see rustfs/backlog#1148 (ilm-1 partial)"]
    async fn test_noncurrent_expiry_still_works_after_immediate_compensation_transition() {
        let (disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -1631,7 +1607,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "FAILING on main: excluded from the serial ILM lane pending a fix, see rustfs/backlog#1148 (ilm-1 partial)"]
    async fn test_noncurrent_transition_still_works_after_immediate_compensation_transition() {
        let (disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -1714,7 +1689,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_modeled_versioned_delete_creates_delete_marker_after_immediate_compensation_transition() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -1762,7 +1736,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_modeled_delete_marker_cleanup_after_immediate_compensation_transition() {
        let (disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -1839,7 +1812,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_scanner_expires_zero_day_current_version() {
        let (disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -1866,7 +1838,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_put_object_immediately_enqueues_zero_day_current_expiry() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -1904,7 +1875,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_scanner_expires_zero_day_noncurrent_version() {
        let (disk_paths, ecstore) = setup_isolated_test_env(false).await;
@@ -1971,7 +1941,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_put_object_immediately_enqueues_zero_day_noncurrent_expiry() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -2032,7 +2001,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    async fn test_background_scanner_expires_zero_day_current_version() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(true).await;

@@ -2056,7 +2024,6 @@ mod serial_tests {
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
    async fn test_background_scanner_expires_zero_day_current_version_for_exact_key_prefix() {
        let (_disk_paths, ecstore) = setup_isolated_test_env(true).await;
@@ -2122,7 +2089,6 @@ mod serial_tests {
    /// tier object is untouched (zero `remove` calls) -> GET streams from the
    /// tier again -> a second restore succeeds.
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-8)"]
    async fn test_restore_chain_local_read_expiry_keeps_remote_and_allows_re_restore() {
        let (_disk_paths, ecstore) = setup_test_env().await;
@@ -2254,7 +2220,6 @@ mod serial_tests {
    /// parts) must reassemble the exact part layout: part count and sizes,
    /// the multipart ETag, and byte-identical content across part boundaries.
    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[serial]
    #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-8)"]
    async fn test_multipart_restore_preserves_parts_and_etag() {
        let (_disk_paths, ecstore) = setup_test_env().await;
@@ -622,8 +622,7 @@ mod test {
        let _resolver_lock = DNS_RESOLVER_TEST_LOCK.lock().unwrap();
        reset_dns_resolver_inner();

-        // DNS labels are limited to 63 bytes, so the system resolver rejects this before lookup.
-        let err = resolve_domain("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.invalid").unwrap_err();
+        let err = resolve_domain("rustfs-resolver-provenance.invalid").unwrap_err();

        assert_ne!(err.kind(), std::io::ErrorKind::Other, "system resolver error was wrapped: {err}");
    }
@@ -5,10 +5,12 @@
 > ```bash
 > cargo nextest list -p e2e_test --message-format json | jq -r '.["rust-suites"][]?.testcases | to_entries[] | select(.value.ignored == false) | .key | split("::")[0]' | sort | uniq -c
 > ```
-> Modules marked ✅ are in the PR smoke profile `e2e-smoke`; 🌙 marks the
-> cluster, protocol, and replication subsets in the consolidated nightly
-> workflow. The `e2e-full` merge/main profile covers the remaining default
-> single-node tests. Committed test-ID digests are enforced before each run.
+> Modules marked ✅ are in the PR smoke profile `e2e-smoke`
+> (`.config/nextest.toml`); admission criteria: `crates/e2e_test/README.md`.
+> 🌙 marks tests in the scheduled `e2e-repl-nightly` profile (backlog#1147
+> repl-1): `replication_extension_test` splits 20 fast tests into the PR smoke
+> lane and 28 slow / `_real_dual_node` / `_real_three_node` / `_real_single_node` tests into the
+> nightly lane (`.github/workflows/e2e-replication-nightly.yml`).
 > Note: counts exclude `#[ignore]`d tests (nextest lists them separately).
 > Managed-SSE (SSE-S3/SSE-KMS) replication contracts assert successful
 > re-encryption on the target (backlog#1783); SSE-C replication still pins a
@@ -17,21 +19,19 @@
 | module | tests | PR smoke |
 |---|---|---|
 | admin_auth_test | 4 | ✅ |
-| admin_iam_crud_test | 3 | ✅ |
+| admin_iam_crud_test | 2 | ✅ |
 | admin_pools_test | 1 | ✅ |
-| admin_timeout_regression_test | 1 | 🌙 |
+| admin_timeout_regression_test | 1 |  |
 | anonymous_access_test | 4 | ✅ |
 | api_rate_limit_test | 3 |  |
 | archive_download_integrity_test | 13 |  |
 | bucket_logging_test | 3 |  |
 | bucket_policy_check_test | 1 | ✅ |
-| bucket_stats_regression_test | 3 |  |
-| chaos | 2 |  |
 | checksum_upload_test | 7 |  |
-| cluster_concurrency_test | 2 | 🌙 |
-| cluster_multidrive_pool_test | 2 | 🌙 |
-| common | 14 |  |
-| compression_test | 6 | ✅ |
+| cluster_concurrency_test | 2 |  |
+| cluster_multidrive_pool_test | 2 |  |
+| common | 12 |  |
+| compression_test | 1 |  |
 | connection_cap_test | 2 |  |
 | console_smoke_test | 1 | ✅ |
 | content_encoding_test | 3 | ✅ |
@@ -45,58 +45,48 @@
 | delete_marker_migration_semantics_test | 2 | ✅ |
 | delete_object_no_content_length_test | 1 |  |
 | delete_objects_versioning_test | 2 | ✅ |
-| delete_regression_test | 5 |  |
-| distributed_startup_regression_test | 3 |  |
 | existing_object_tag_policy_test | 4 |  |
-| fake_s3_target | 6 | ✅ |
+| fake_s3_target | 4 | ✅ |
 | fault_proxy | 7 |  |
 | get_codec_streaming_compat_test | 1 |  |
-| get_stream_failure_observability_test | 1 |  |
-| group_delete_test | 1 |  |
 | head_object_consistency_test | 1 | ✅ |
 | head_object_range_test | 1 | ✅ |
-| heal_erasure_disk_rebuild_test | 4 | 🌙 |
-| inline_fast_path_cluster_test | 16 |  |
+| heal_erasure_disk_rebuild_test | 3 |  |
+| inline_fast_path_cluster_test | 14 |  |
 | internode_rpc_signature_e2e_test | 5 |  |
-| kms | 48 |  |
+| kms | 41 |  |
 | leading_slash_key_test | 2 | ✅ |
-| lifecycle_regression_test | 4 |  |
-| list_buckets_auth_test | 1 | ✅ |
 | list_buckets_double_slash_test | 3 | ✅ |
-| list_buckets_iam_filter_test | 1 | ✅ |
 | list_object_versions_metadata_extension_test | 1 |  |
 | list_object_versions_regression_test | 2 | ✅ |
 | list_objects_duplicates_test | 3 | ✅ |
 | list_objects_v2_metadata_extension_test | 1 |  |
 | list_objects_v2_pagination_test | 12 | ✅ |
-| listing_regression_test | 4 |  |
 | mc_mirror_small_bucket_test | 1 |  |
 | multipart_auth_test | 75 |  |
 | multipart_storage_class_test | 3 | ✅ |
-| namespace_lock_quorum_test | 2 | 🌙 |
+| namespace_lock_quorum_test | 2 |  |
 | negative_sigv4_test | 6 | ✅ |
-| notification_startup_regression_test | 2 |  |
 | notification_webhook_test | 3 | ✅ |
-| object_lambda_test | 16 | 🌙 |
-| object_lock | 34 |  |
+| object_lambda_test | 16 |  |
+| object_lock | 33 |  |
 | overwrite_cleanup_regression_test | 1 |  |
 | presigned_negative_test | 7 | ✅ |
-| protocols | 16 | 🌙 |
+| protocols | 16 |  |
 | quota_test | 14 |  |
-| reliability_disk_fault_test | 4 |  |
-| reliant | 25 | 19 ✅ |
-| replication_extension_test | 75 | 20 ✅ +55 🌙 |
+| reliability_disk_fault_test | 3 |  |
+| reliant | 24 | 18 ✅ |
+| replication_extension_test | 50 | 20 ✅ +30 🌙 |
 | security_boundary_test | 4 |  |
+| ssec_copy_test | 2 | ✅ |
 | server_startup_failfast_test | 1 |  |
 | snowball_auto_extract_test | 6 |  |
 | special_chars_test | 14 | ✅ |
-| ssec_copy_test | 2 | ✅ |
-| stale_multipart_cleanup_cluster_test | 1 | 🌙 |
+| stale_multipart_cleanup_cluster_test | 1 |  |
 | storage_class_capability_test | 4 | ✅ |
-| sts_query_compat_test | 6 | ✅ |
-| tier_transition_regression_test | 3 |  |
+| sts_query_compat_test | 3 | ✅ |
 | tls_gen | 3 |  |
 | tls_hot_reload_test | 1 | ✅ |
 | version_id_regression_test | 10 | ✅ |

-**Total listed: 577 tests across 82 modules · PR smoke: 163 tests / 36 modules · merge/main full: 455 tests / 73 modules · nightly replication: 55 tests · nightly cluster faults: 28 tests / 7 modules · nightly protocols: 16 tests** · updated 2026-08-21.
+**Total listed: 530 tests across 70 modules · PR smoke subset: 148 tests / 33 modules** (31 full modules + 18 `reliant` tests + 20 of `replication_extension_test`) **· nightly `e2e-repl-nightly`: 30 tests** · updated 2026-08-09.
@@ -48,8 +48,11 @@ fn materialize_case(path: String, prefix: Option<String>, flags: &[String]) -> (
        .fold((path, prefix), |(path, prefix), flag| apply_flag(path, prefix, flag))
 }

-fn has_parent_segments(path: &str) -> bool {
-    path.split(['/', '\\']).any(|segment| segment == "..")
+fn has_dot_segments(path: &str) -> bool {
+    path.split(['/', '\\']).any(|segment| {
+        let trimmed = segment.trim();
+        trimmed == "." || trimmed == ".."
+    })
 }

 fuzz_target!(|data: &[u8]| {
@@ -63,8 +66,8 @@ fuzz_target!(|data: &[u8]| {

    if let Ok(key) = normalize_extract_entry_key(&path, prefix.as_deref(), is_dir) {
        assert!(
-            !has_parent_segments(&key),
-            "accepted archive entry retained parent segments: path={:?} prefix={:?} key={:?}",
+            !has_dot_segments(&key),
+            "accepted archive entry retained dot segments: path={:?} prefix={:?} key={:?}",
            path,
            prefix,
            key
@@ -69,11 +69,7 @@ fuzz_target!(|data: &[u8]| {
        && let Some(object) = value.as_object()
    {
        let mut legacy_doc = Map::new();
-        if let Some(policy) = object
-            .get("Policy")
-            .or_else(|| object.get("policy"))
-            .filter(|policy| serde_json::from_value::<Policy>((*policy).clone()).is_ok())
-        {
+        if let Some(policy) = object.get("Policy").or_else(|| object.get("policy")) {
            legacy_doc.insert("version".to_string(), json!(1));
            legacy_doc.insert("policy".to_string(), policy.clone());
            legacy_doc.insert("create_date".to_string(), json!("2025-03-07T12:00:00Z"));
@@ -17,220 +17,6 @@
 //! This binary shares RustFS's existing subcommand dispatcher and provides the
 //! documented entry point for offline tooling such as `inspect bucket-meta`.

-use std::fs;
-use std::io::{Read as _, Write as _};
-use std::path::{Path, PathBuf};
-use std::process::ExitCode;
-use std::time::{SystemTime, UNIX_EPOCH};
-
-use rustfs::connect::offline::{OfflineEnrollment, OfflineKeyStore};
-
-/// Owner read/write only. The response names the key being enrolled and the
-/// challenge it answers; neither belongs to anyone else on the machine.
-#[cfg(unix)]
-const RESPONSE_MODE: u32 = 0o600;
-
-const USAGE: &str = "\
-Usage: rustfs-cli connect offline enroll --challenge <path|-> --output <path> [--key-dir <path>]
-
-Answers a Connect offline enrolment challenge without a network. Reads the
-challenge from a file or from stdin when the path is `-`, verifies it against the
-enrolment root compiled into this binary, mints the key being enrolled on first
-use, and writes the signed response.
-
-No secret is ever accepted on the command line.
-";
-
-fn main() -> ExitCode {
-    let arguments: Vec<String> = std::env::args().skip(1).collect();
-
-    // Offline enrolment is handled before the server dispatcher is reached, and
-    // the reason is the surface's whole point: `run_process` builds a Tokio
-    // runtime and enters the server's async main. An air-gapped enrolment must
-    // not start a runtime, a task, or anything that could open a socket, so the
-    // two paths cannot share an entry.
-    if matches!(
-        arguments.first().map(String::as_str),
-        Some("connect") if matches!(arguments.get(1).map(String::as_str), Some("offline"))
-    ) {
-        return match run_offline(&arguments[2..]) {
-            Ok(()) => ExitCode::SUCCESS,
-            Err(message) => {
-                eprintln!("rustfs-cli: {message}");
-                ExitCode::FAILURE
-            }
-        };
-    }
-
+fn main() {
    rustfs::startup_entrypoint::run_process();
-
-    ExitCode::SUCCESS
-}
-
-fn run_offline(arguments: &[String]) -> Result<(), String> {
-    match arguments.first().map(String::as_str) {
-        Some("enroll") => enroll(&arguments[1..]),
-        Some(other) => Err(format!("unknown offline subcommand `{other}`\n\n{USAGE}")),
-        None => Err(format!("missing offline subcommand\n\n{USAGE}")),
-    }
-}
-
-fn enroll(arguments: &[String]) -> Result<(), String> {
-    let mut challenge_path: Option<String> = None;
-    let mut output_path: Option<String> = None;
-    let mut key_directory: Option<String> = None;
-
-    let mut index = 0;
-    while index < arguments.len() {
-        let flag = arguments[index].as_str();
-        let take_value = |name: &str| -> Result<String, String> {
-            arguments
-                .get(index + 1)
-                .cloned()
-                .ok_or_else(|| format!("`{name}` needs a value\n\n{USAGE}"))
-        };
-
-        match flag {
-            "--challenge" => challenge_path = Some(take_value("--challenge")?),
-            "--output" => output_path = Some(take_value("--output")?),
-            "--key-dir" => key_directory = Some(take_value("--key-dir")?),
-            "-h" | "--help" => {
-                println!("{USAGE}");
-                return Ok(());
-            }
-            other => return Err(format!("unknown option `{other}`\n\n{USAGE}")),
-        }
-
-        index += 2;
-    }
-
-    let challenge_path = challenge_path.ok_or_else(|| format!("`--challenge` is required\n\n{USAGE}"))?;
-    let output_path = output_path.ok_or_else(|| format!("`--output` is required\n\n{USAGE}"))?;
-    let key_directory = key_directory.unwrap_or_else(|| ".".to_string());
-
-    let challenge = read_challenge(&challenge_path)?;
-
-    let now = SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .map_err(|_| "the system clock is before the Unix epoch".to_string())?
-        .as_secs() as i64;
-
-    let verified = OfflineEnrollment::verify_challenge(&challenge, now).map_err(|error| error.to_string())?;
-
-    // First use mints the key; a retry answers with the one already enrolled,
-    // because the operator may already be carrying a response naming it.
-    let key = OfflineKeyStore::new(&key_directory)
-        .load_or_create()
-        .map_err(|error| error.to_string())?;
-
-    let mut device_nonce = [0u8; 32];
-    getrandom(&mut device_nonce)?;
-
-    let response = OfflineEnrollment::build_response(&verified, &key, &device_nonce, now).map_err(|error| error.to_string())?;
-
-    write_response(Path::new(&output_path), &response)?;
-
-    Ok(())
-}
-
-/// Reads the challenge from a file, or from stdin when the path is `-`.
-///
-/// A challenge is not a secret — it is signed, public, and carried in by hand —
-/// so accepting a path is safe. The response's key never arrives this way.
-fn read_challenge(path: &str) -> Result<Vec<u8>, String> {
-    if path == "-" {
-        let mut buffer = Vec::new();
-        std::io::stdin()
-            .read_to_end(&mut buffer)
-            .map_err(|error| format!("cannot read the challenge from stdin: {error}"))?;
-        return Ok(buffer);
-    }
-
-    fs::read(path).map_err(|error| format!("cannot read the challenge at {path}: {error}"))
-}
-
-/// Writes the response durably and atomically at mode 0600.
-///
-/// Not the no-clobber publish `IdentityStore` performs for a key: an operator
-/// who reruns an enrolment expects the response file to be replaced, whereas a
-/// second key would strand the first. Same durability, deliberately different
-/// publication rule.
-fn write_response(path: &Path, response: &[u8]) -> Result<(), String> {
-    let parent = path.parent().filter(|parent| !parent.as_os_str().is_empty());
-    let temporary: PathBuf = match parent {
-        Some(parent) => parent.join(format!(".{}.tmp", file_name(path))),
-        None => PathBuf::from(format!(".{}.tmp", file_name(path))),
-    };
-
-    let mut options = fs::OpenOptions::new();
-    options.write(true).create(true).truncate(true);
-    #[cfg(unix)]
-    {
-        use std::os::unix::fs::OpenOptionsExt as _;
-        options.mode(RESPONSE_MODE);
-    }
-
-    let write = (|| -> std::io::Result<()> {
-        let mut file = options.open(&temporary)?;
-        file.write_all(response)?;
-
-        // The umask can only narrow the creation mode, so set the exact mode
-        // before the bytes become durable.
-        #[cfg(unix)]
-        {
-            use std::os::unix::fs::PermissionsExt as _;
-            file.set_permissions(fs::Permissions::from_mode(RESPONSE_MODE))?;
-        }
-
-        file.sync_all()
-    })();
-
-    if let Err(error) = write {
-        let _ = fs::remove_file(&temporary);
-        return Err(format!("cannot write the response to {}: {error}", path.display()));
-    }
-
-    fs::rename(&temporary, path).map_err(|error| {
-        let _ = fs::remove_file(&temporary);
-        format!("cannot publish the response at {}: {error}", path.display())
-    })?;
-
-    if let Some(parent) = parent {
-        sync_directory(parent);
-    }
-
-    Ok(())
-}
-
-fn file_name(path: &Path) -> String {
-    path.file_name()
-        .map(|name| name.to_string_lossy().into_owned())
-        .unwrap_or_else(|| "response".to_string())
-}
-
-/// Fsync the directory so the renamed entry survives power loss. Directories
-/// cannot be opened for syncing on Windows, where this is a no-op.
-fn sync_directory(directory: &Path) {
-    #[cfg(unix)]
-    {
-        if let Ok(handle) = fs::File::open(directory) {
-            let _ = handle.sync_all();
-        }
-    }
-    #[cfg(not(unix))]
-    let _ = directory;
-}
-
-/// Fills `buffer` with operating-system randomness.
-///
-/// The device nonce must be unpredictable: it is what stops a captured response
-/// being replayed as a fresh one. Sourced through p256's pinned rand_core 0.6
-/// rather than the workspace `rand` 0.10, matching `identity.rs`; the two are
-/// different crate versions and only the pinned one is on p256's own path.
-fn getrandom(buffer: &mut [u8]) -> Result<(), String> {
-    use p256::elliptic_curve::rand_core::{OsRng, RngCore as _};
-
-    OsRng
-        .try_fill_bytes(buffer)
-        .map_err(|error| format!("the operating system random source failed: {error}"))
 }
@@ -27,8 +27,6 @@

 pub mod identity;
 pub mod identity_store;
-pub mod offline;

 pub use identity::{DeviceIdentity, IdentityError, RegistrationProof, RegistrationTranscript};
 pub use identity_store::{IdentityStore, StoreError};
-pub use offline::{EnrollmentError, OfflineEnrollment, OfflineKeyStore, VerifiedChallenge};
@@ -1,684 +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.
-
-//! Challenge verification and response production for offline enrolment.
-//!
-//! Two invariants carry the security of this surface and both are easy to break
-//! by accident:
-//!
-//! - Every signature is checked over the octets that arrived, never over a
-//!   re-serialised document. Parsing happens only to route the verification, and
-//!   nothing a parse yields is believed until the signature over those same
-//!   octets has verified.
-//! - The enrolment root is the constant in this file. It is never taken from a
-//!   challenge, a configuration file, or an operator prompt, so there is no
-//!   trust-on-first-use path an operator could be talked into.
-//!
-//! The order of the checks in [`OfflineEnrollment::verify_challenge`] is frozen
-//! by `verificationOrder.enrollmentChallenge` in
-//! `protocol/agent/v1/fixtures/offline-enrollment/trust-model.json`, and the
-//! signature encoding, the domain separation tags, and every rejection reason
-//! are frozen beside it. Reordering the checks changes which reason a given
-//! artifact produces, which is itself part of the contract.
-
-use base64::Engine as _;
-use base64::engine::general_purpose::{STANDARD as BASE64_STANDARD, URL_SAFE_NO_PAD as BASE64_URL_NO_PAD};
-use p256::ecdsa::signature::{Signer as _, Verifier as _};
-use p256::ecdsa::{Signature, SigningKey, VerifyingKey};
-use p256::pkcs8::DecodePrivateKey as _;
-use serde::{Deserialize, Serialize};
-use sha2::{Digest as _, Sha256};
-use time::{Date, Month, OffsetDateTime, PrimitiveDateTime, Time};
-
-use crate::connect::identity::DeviceIdentity;
-
-/// The hosted enrolment root, compiled in. Both halves are pinned: the
-/// fingerprint identifies the root, and the point is what actually verifies the
-/// first link, so a build cannot be pointed at a different key by supplying one.
-const PINNED_ROOT_KEY_ID: &str = "df22e2806112debbe953672aafa186d699af0e97dd3fd2b09fa8359005fe348f";
-const PINNED_ROOT_PUBLIC_KEY: &str = "BFfx-K-FfEA5nK_Rz3IHacvRCkJyQ7JOd1geLyU6HKRZDgNezmVuKhvJ22VhemyjV__Gshk8JGGqOBzYPMD0p6s";
-
-/// Domain separation tags. A document that verifies under one of these must not
-/// be accepted for another artifact type, so the tag is part of the signature
-/// input rather than a property of the caller.
-const TAG_TRUST_LINK: &[u8] = b"rustfs-offline-trust-link-v1";
-const TAG_CHALLENGE: &[u8] = b"rustfs-offline-enrollment-challenge-v1";
-const TAG_RESPONSE: &[u8] = b"rustfs-offline-enrollment-response-v1";
-
-/// The single octet between the tag and the signed document.
-const DOMAIN_SEPARATOR: u8 = 0x00;
-
-const SIGNATURE_ALGORITHM: &str = "ES256";
-const PROTOCOL_VERSION: &str = "v1";
-const FORMAT_TRUST_LINK: &str = "rustfs.connect.offline.trustLink/1";
-const FORMAT_CHALLENGE: &str = "rustfs.connect.offline.enrollmentChallenge/1";
-const FORMAT_RESPONSE: &str = "rustfs.connect.offline.enrollmentResponse/1";
-
-/// DER SubjectPublicKeyInfo header for an uncompressed P-256 point. A keyId is
-/// the SHA-256 of this prefix followed by the 65 octet point, so the prefix is
-/// also how a device public key is recovered from its own DER encoding.
-const SPKI_PREFIX: [u8; 26] = [
-    0x30, 0x59, 0x30, 0x13, 0x06, 0x07, 0x2a, 0x86, 0x48, 0xce, 0x3d, 0x02, 0x01, 0x06, 0x08, 0x2a, 0x86, 0x48, 0xce, 0x3d, 0x03,
-    0x01, 0x07, 0x03, 0x42, 0x00,
-];
-
-/// Order of the P-256 group, and half of it. `r` and `s` must lie in `[1, n)`,
-/// and `s` additionally in `[1, n/2]`: ECDSA admits both `s` and `n - s`, and a
-/// signature with two spellings cannot serve as an artifact identity. Every
-/// ECDSA library accepts the malleated form, so the encoding layer rejects it.
-const GROUP_ORDER: [u8; 32] = [
-    0xff, 0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x00, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xbc, 0xe6, 0xfa, 0xad, 0xa7,
-    0x17, 0x9e, 0x84, 0xf3, 0xb9, 0xca, 0xc2, 0xfc, 0x63, 0x25, 0x51,
-];
-const MAX_S: [u8; 32] = [
-    0x7f, 0xff, 0xff, 0xff, 0x80, 0x00, 0x00, 0x00, 0x7f, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xde, 0x73, 0x7d, 0x56, 0xd3,
-    0x8b, 0xcf, 0x42, 0x79, 0xdc, 0xe5, 0x61, 0x7e, 0x31, 0x92, 0xa8,
-];
-
-const SCALAR_OCTETS: usize = 32;
-const SIGNATURE_OCTETS: usize = 64;
-/// 64 octets as unpadded base64url. The length is checked before decoding so
-/// that `=` padding, the standard alphabet, DER, and a truncated value are all
-/// refused rather than repaired.
-const SIGNATURE_VALUE_CHARS: usize = 86;
-
-const PUBLIC_KEY_OCTETS: usize = 65;
-const PUBLIC_KEY_CHARS: usize = 87;
-/// SEC1 tag of an uncompressed point. Compressed and hybrid forms are refused.
-const UNCOMPRESSED_POINT: u8 = 0x04;
-
-const TIMESTAMP_CHARS: usize = 20;
-
-/// The chain is exactly two links: a pinned root issues the intermediate, and
-/// the intermediate issues the signing key. Roles are positional and the
-/// enumeration is closed.
-const CHAIN_LINK_COUNT: usize = 2;
-const CHAIN_ROLES: [&str; CHAIN_LINK_COUNT] = ["intermediate", "signing"];
-
-/// Skew allowed on the challenge window. A device may have no synchronised
-/// clock at all, so its own reading of "now" is advisory.
-const CLOCK_SKEW_TOLERANCE: i64 = 300;
-
-/// Longest life a challenge may claim. The issuer sets both ends of its own
-/// window, so the protocol bound is applied on top of the declared expiry
-/// rather than trusted from it.
-const MAX_CHALLENGE_LIFETIME: i64 = 604_800;
-
-/// A challenge that verified, with the fields the response has to echo.
-///
-/// Construction is the proof: a value of this type only exists after the chain
-/// closed on the pinned root and the challenge signature verified over the
-/// received octets.
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub struct VerifiedChallenge {
-    pub challenge_id: String,
-    pub organization_name: String,
-    pub cluster_name: String,
-    pub nonce: String,
-    pub issued_at: String,
-    pub expires_at: String,
-    pub connect_key_id: String,
-    /// The signature value of the challenge, verbatim. It binds a response to
-    /// the one challenge it answers, so it is carried rather than recomputed.
-    pub challenge_proof: String,
-}
-
-/// Why an offline enrolment artifact was refused.
-///
-/// The variants are the frozen `reason` vocabulary of
-/// `fixtures/offline-enrollment/error-codes.json`, which spans both halves of
-/// the exchange. The device half implemented here produces the encoding, chain,
-/// version, and freshness reasons; the reasons that describe a response being
-/// evaluated against stored state — [`Self::ChallengeUnknown`],
-/// [`Self::ChallengeProofInvalid`], [`Self::DeviceProofInvalid`],
-/// [`Self::EnrollmentReplayed`], [`Self::OrganizationMismatch`], and
-/// [`Self::ClusterMismatch`] — are Connect's to raise and are named here so the
-/// two sides share one vocabulary.
-///
-/// No variant carries a payload: a rejection must never disclose key material,
-/// signature octets, nonces, or document bytes.
-#[derive(Debug, PartialEq, Eq, thiserror::Error)]
-pub enum EnrollmentError {
-    #[error("protocolVersion is missing, malformed, or names an unsupported major version")]
-    UnsupportedProtocol,
-
-    #[error("formatVersion is not a supported offline enrollment format")]
-    UnsupportedFormat,
-
-    #[error("the signature is not 64 octets of fixed-width r||s in unpadded base64url")]
-    SignatureMalformed,
-
-    #[error("the signature is not in its canonical low-S form")]
-    SignatureNotCanonical,
-
-    #[error("the signature does not verify over the received octets")]
-    SignatureInvalid,
-
-    #[error("the trust chain is not issued by a root pinned in this build")]
-    EnrollmentRootUnknown,
-
-    #[error("a trust link is invalid, misordered, or outside its validity at the challenge issuedAt")]
-    TrustChainInvalid,
-
-    #[error("connectKeyId is not the subject of the last trust link")]
-    ConnectKeyUnchained,
-
-    #[error("no issued challenge matches this challengeId")]
-    ChallengeUnknown,
-
-    #[error("the challenge is not yet valid at the evaluation time")]
-    ChallengeNotYetValid,
-
-    #[error("the challenge has expired at the evaluation time")]
-    ChallengeExpired,
-
-    #[error("the response nonce or challengeProof is not the one issued for this challenge")]
-    ChallengeProofInvalid,
-
-    #[error("the response does not prove possession of the device key it presents")]
-    DeviceProofInvalid,
-
-    #[error("the challenge was already consumed")]
-    EnrollmentReplayed,
-
-    #[error("the response names a different organization than the challenge it answers")]
-    OrganizationMismatch,
-
-    #[error("the response names a different cluster than the challenge it answers")]
-    ClusterMismatch,
-
-    /// The artifact could not be read as a signed enrolment document at all: the
-    /// envelope, the base64 of the signed octets, or a field the frozen order
-    /// reads before the signature verifies did not parse. The frozen reason set
-    /// has no code for a structurally unreadable document, so this variant maps
-    /// to none of them.
-    #[error("the offline enrollment document is not well formed")]
-    MalformedDocument,
-
-    /// A fault on this side of the exchange rather than in the artifact: the
-    /// device key did not round-trip through its own PKCS#8 encoding, or the
-    /// caller named an instant outside the representable calendar. Fails closed
-    /// because a half-produced response must never reach removable media.
-    #[error("the enrollment response could not be produced on this device")]
-    ResponseNotProduced,
-}
-
-impl EnrollmentError {
-    /// The frozen `reason` an operator and Connect both branch on.
-    ///
-    /// The `Display` message is prose and may be reworded; this is the stable
-    /// identifier, so nothing should parse the message instead. The two
-    /// variants with no frozen counterpart deliberately return codes outside
-    /// the frozen set rather than borrowing the nearest one, so a document that
-    /// simply failed to parse can never be reported as a signature or freshness
-    /// failure.
-    pub fn reason(&self) -> &'static str {
-        match self {
-            Self::UnsupportedProtocol => "UNSUPPORTED_PROTOCOL",
-            Self::UnsupportedFormat => "UNSUPPORTED_FORMAT",
-            Self::SignatureMalformed => "SIGNATURE_MALFORMED",
-            Self::SignatureNotCanonical => "SIGNATURE_NOT_CANONICAL",
-            Self::SignatureInvalid => "SIGNATURE_INVALID",
-            Self::EnrollmentRootUnknown => "ENROLLMENT_ROOT_UNKNOWN",
-            Self::TrustChainInvalid => "TRUST_CHAIN_INVALID",
-            Self::ConnectKeyUnchained => "CONNECT_KEY_UNCHAINED",
-            Self::ChallengeUnknown => "CHALLENGE_UNKNOWN",
-            Self::ChallengeNotYetValid => "CHALLENGE_NOT_YET_VALID",
-            Self::ChallengeExpired => "CHALLENGE_EXPIRED",
-            Self::ChallengeProofInvalid => "CHALLENGE_PROOF_INVALID",
-            Self::DeviceProofInvalid => "DEVICE_PROOF_INVALID",
-            Self::EnrollmentReplayed => "ENROLLMENT_REPLAYED",
-            Self::OrganizationMismatch => "ORGANIZATION_MISMATCH",
-            Self::ClusterMismatch => "CLUSTER_MISMATCH",
-            Self::MalformedDocument => "MALFORMED_DOCUMENT",
-            Self::ResponseNotProduced => "RESPONSE_NOT_PRODUCED",
-        }
-    }
-}
-
-/// A signed document, in the shape both directions carry it. `bytes` is
-/// standard padded base64 of the exact octets that were signed; nothing else is
-/// ever used as the signature input.
-#[derive(Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct SignedDocument {
-    bytes: String,
-    signature: DocumentSignature,
-}
-
-#[derive(Serialize, Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct DocumentSignature {
-    algorithm: String,
-    key_id: String,
-    value: String,
-}
-
-/// The three fields the frozen order permits reading before anything verifies.
-/// They route the verification and are not facts until it has.
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct ChallengeRouting {
-    connect_key_id: String,
-    issued_at: String,
-    trust_chain: Vec<SignedDocument>,
-}
-
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct ChallengeDocument {
-    format_version: String,
-    protocol_version: String,
-    challenge_id: String,
-    organization_name: String,
-    cluster_name: String,
-    nonce: String,
-    issued_at: String,
-    expires_at: String,
-    connect_key_id: String,
-}
-
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct TrustLink {
-    format_version: String,
-    protocol_version: String,
-    role: String,
-    issuer_key_id: String,
-    subject_key_id: String,
-    subject_public_key: String,
-    not_before: String,
-    not_after: String,
-}
-
-#[derive(Serialize)]
-#[serde(rename_all = "camelCase")]
-struct ResponseDocument<'a> {
-    format_version: &'a str,
-    protocol_version: &'a str,
-    challenge_id: &'a str,
-    organization_name: &'a str,
-    cluster_name: &'a str,
-    challenge_nonce: &'a str,
-    challenge_proof: &'a str,
-    device_key_id: String,
-    device_public_key: String,
-    device_nonce: String,
-    produced_at: String,
-}
-
-/// The device half of the offline enrolment exchange: bytes in, bytes out.
-pub struct OfflineEnrollment;
-
-impl OfflineEnrollment {
-    /// Verify an enrolment challenge and return what a response must echo.
-    ///
-    /// `now_unix` is the device's reading of the current time, which the clock
-    /// skew tolerance treats as advisory.
-    pub fn verify_challenge(document: &[u8], now_unix: i64) -> Result<VerifiedChallenge, EnrollmentError> {
-        let envelope: SignedDocument = serde_json::from_slice(document).map_err(|_| EnrollmentError::MalformedDocument)?;
-
-        // Step 1: the encoding is checked before anything is decoded from it, so
-        // a DER, padded, truncated, out-of-range, or high-S signature is refused
-        // on its spelling rather than handed to a library that would accept it.
-        let signature = decode_signature(&envelope.signature)?;
-
-        // The octets that were transmitted. They are never re-serialised: every
-        // later step signs and parses this same buffer.
-        let bytes = BASE64_STANDARD
-            .decode(envelope.bytes.as_bytes())
-            .map_err(|_| EnrollmentError::MalformedDocument)?;
-
-        // Step 2: routing only.
-        let routing: ChallengeRouting = serde_json::from_slice(&bytes).map_err(|_| EnrollmentError::MalformedDocument)?;
-        let issued_at = parse_timestamp(&routing.issued_at)?;
-
-        // Steps 3 to 5.
-        let connect_key = verify_trust_chain(&routing.trust_chain, &routing.connect_key_id, issued_at)?;
-
-        // Step 6. The verification key comes from the chain, so `signature.keyId`
-        // is a label rather than an input: a value naming some other key simply
-        // fails to verify here.
-        if !verifies(&connect_key, TAG_CHALLENGE, &bytes, &signature) {
-            return Err(EnrollmentError::SignatureInvalid);
-        }
-
-        // Step 7: only now is the document read as a fact.
-        let challenge: ChallengeDocument = serde_json::from_slice(&bytes).map_err(|_| EnrollmentError::MalformedDocument)?;
-        if challenge.protocol_version != PROTOCOL_VERSION {
-            return Err(EnrollmentError::UnsupportedProtocol);
-        }
-        if challenge.format_version != FORMAT_CHALLENGE {
-            return Err(EnrollmentError::UnsupportedFormat);
-        }
-
-        // Step 8.
-        let expires_at = parse_timestamp(&challenge.expires_at)?;
-        check_challenge_window(issued_at, expires_at, now_unix)?;
-
-        Ok(VerifiedChallenge {
-            challenge_id: challenge.challenge_id,
-            organization_name: challenge.organization_name,
-            cluster_name: challenge.cluster_name,
-            nonce: challenge.nonce,
-            issued_at: challenge.issued_at,
-            expires_at: challenge.expires_at,
-            connect_key_id: challenge.connect_key_id,
-            challenge_proof: envelope.signature.value,
-        })
-    }
-
-    /// Build the signed response an operator carries back to Connect.
-    ///
-    /// `device_nonce` is the response's own replay value and must come from a
-    /// cryptographic source. The private key never appears in the result: only
-    /// the public point, its fingerprint, and a signature over the document
-    /// that presents them, which is what makes presenting the key safe.
-    pub fn build_response(
-        challenge: &VerifiedChallenge,
-        key: &DeviceIdentity,
-        device_nonce: &[u8; 32],
-        produced_at_unix: i64,
-    ) -> Result<Vec<u8>, EnrollmentError> {
-        let issued_at = parse_timestamp(&challenge.issued_at)?;
-        let expires_at = parse_timestamp(&challenge.expires_at)?;
-        // Connect re-checks producedAt against the same window, so a response
-        // outside it is refused here rather than written to media and rejected
-        // after the operator has carried it out.
-        check_challenge_window(issued_at, expires_at, produced_at_unix)?;
-
-        let point = device_public_point(key)?;
-        let produced_at = format_timestamp(produced_at_unix)?;
-
-        let document = ResponseDocument {
-            format_version: FORMAT_RESPONSE,
-            protocol_version: PROTOCOL_VERSION,
-            challenge_id: &challenge.challenge_id,
-            organization_name: &challenge.organization_name,
-            cluster_name: &challenge.cluster_name,
-            challenge_nonce: &challenge.nonce,
-            challenge_proof: &challenge.challenge_proof,
-            device_key_id: key_id(&point),
-            device_public_key: BASE64_URL_NO_PAD.encode(point),
-            device_nonce: BASE64_URL_NO_PAD.encode(device_nonce),
-            produced_at,
-        };
-
-        // Serialised once. These octets are what is signed and what is carried,
-        // so no second serialisation can disagree with the signature.
-        let bytes = serde_json::to_vec(&document).map_err(|_| EnrollmentError::ResponseNotProduced)?;
-        let signature = sign(key, TAG_RESPONSE, &bytes)?;
-
-        let envelope = SignedDocument {
-            bytes: BASE64_STANDARD.encode(&bytes),
-            signature: DocumentSignature {
-                algorithm: SIGNATURE_ALGORITHM.to_owned(),
-                key_id: document.device_key_id,
-                value: signature,
-            },
-        };
-
-        serde_json::to_vec(&envelope).map_err(|_| EnrollmentError::ResponseNotProduced)
-    }
-}
-
-/// Walk the chain from the pinned root to the signing key, returning the key
-/// `connect_key_id` names once the chain vouches for it.
-fn verify_trust_chain(
-    chain: &[SignedDocument],
-    connect_key_id: &str,
-    challenge_issued_at: i64,
-) -> Result<VerifyingKey, EnrollmentError> {
-    // The pinned root gate runs before the chain's shape is examined, so a
-    // chain that is internally consistent under a foreign root — exactly what
-    // trust on first use would have accepted — is refused for its root rather
-    // than for its length.
-    let first = chain.first().ok_or(EnrollmentError::EnrollmentRootUnknown)?;
-    let root = decode_trust_link(first)?;
-    if root.0.issuer_key_id != PINNED_ROOT_KEY_ID {
-        return Err(EnrollmentError::EnrollmentRootUnknown);
-    }
-
-    let [_, second] = chain else {
-        return Err(EnrollmentError::TrustChainInvalid);
-    };
-    let links = [root, decode_trust_link(second)?];
-
-    let mut issuer_key_id = PINNED_ROOT_KEY_ID.to_owned();
-    let (mut issuer_key, _) = decode_public_key(PINNED_ROOT_PUBLIC_KEY).ok_or(EnrollmentError::EnrollmentRootUnknown)?;
-
-    for (index, ((link, link_bytes), entry)) in links.iter().zip(chain).enumerate() {
-        if link.format_version != FORMAT_TRUST_LINK
-            || link.protocol_version != PROTOCOL_VERSION
-            || link.role != CHAIN_ROLES[index]
-            || link.issuer_key_id != issuer_key_id
-            // A link that names itself as its own issuer would let a stolen
-            // intermediate mint its own root.
-            || link.subject_key_id == link.issuer_key_id
-        {
-            return Err(EnrollmentError::TrustChainInvalid);
-        }
-
-        let (subject_key, subject_point) =
-            decode_public_key(&link.subject_public_key).ok_or(EnrollmentError::TrustChainInvalid)?;
-        if key_id(&subject_point) != link.subject_key_id {
-            return Err(EnrollmentError::TrustChainInvalid);
-        }
-
-        let signature = decode_signature(&entry.signature)?;
-        if !verifies(&issuer_key, TAG_TRUST_LINK, link_bytes, &signature) {
-            return Err(EnrollmentError::TrustChainInvalid);
-        }
-
-        // The issuer controls both ends of a link's window, so it is evaluated
-        // with no skew tolerance, and against the challenge's issuedAt rather
-        // than against the device clock: a challenge carries the chain that was
-        // valid when it was issued.
-        let not_before = parse_timestamp(&link.not_before)?;
-        let not_after = parse_timestamp(&link.not_after)?;
-        if challenge_issued_at < not_before || challenge_issued_at > not_after {
-            return Err(EnrollmentError::TrustChainInvalid);
-        }
-
-        issuer_key_id = link.subject_key_id.clone();
-        issuer_key = subject_key;
-    }
-
-    if issuer_key_id != connect_key_id {
-        return Err(EnrollmentError::ConnectKeyUnchained);
-    }
-
-    Ok(issuer_key)
-}
-
-/// Decode a link and keep the octets it was signed over: the signature is
-/// checked against these, never against a re-encoding of the parsed link.
-fn decode_trust_link(entry: &SignedDocument) -> Result<(TrustLink, Vec<u8>), EnrollmentError> {
-    let bytes = BASE64_STANDARD
-        .decode(entry.bytes.as_bytes())
-        .map_err(|_| EnrollmentError::MalformedDocument)?;
-    let link = serde_json::from_slice(&bytes).map_err(|_| EnrollmentError::TrustChainInvalid)?;
-    Ok((link, bytes))
-}
-
-/// Check a signature's spelling and range, then admit it.
-///
-/// `r` and `s` are compared against the group order here rather than left to
-/// the ECDSA library, because a library that accepts high-S — every library
-/// does — would let a malleated copy of an artifact pass as a second artifact.
-fn decode_signature(signature: &DocumentSignature) -> Result<Signature, EnrollmentError> {
-    if signature.algorithm != SIGNATURE_ALGORITHM {
-        return Err(EnrollmentError::SignatureMalformed);
-    }
-
-    let value = signature.value.as_bytes();
-    if value.len() != SIGNATURE_VALUE_CHARS || !value.iter().all(|byte| is_base64url(*byte)) {
-        return Err(EnrollmentError::SignatureMalformed);
-    }
-
-    let decoded = BASE64_URL_NO_PAD
-        .decode(value)
-        .map_err(|_| EnrollmentError::SignatureMalformed)?;
-    let octets: [u8; SIGNATURE_OCTETS] = decoded
-        .as_slice()
-        .try_into()
-        .map_err(|_| EnrollmentError::SignatureMalformed)?;
-
-    // Big-endian octets of equal length order lexicographically exactly as the
-    // integers they spell, so a slice comparison is the range check.
-    let (r, s) = octets.split_at(SCALAR_OCTETS);
-    let out_of_range = |scalar: &[u8]| scalar.iter().all(|byte| *byte == 0) || scalar >= &GROUP_ORDER[..];
-    if out_of_range(r) || out_of_range(s) {
-        return Err(EnrollmentError::SignatureMalformed);
-    }
-    if s > &MAX_S[..] {
-        return Err(EnrollmentError::SignatureNotCanonical);
-    }
-
-    Signature::from_slice(&octets).map_err(|_| EnrollmentError::SignatureMalformed)
-}
-
-fn verifies(key: &VerifyingKey, tag: &[u8], bytes: &[u8], signature: &Signature) -> bool {
-    key.verify(&signature_input(tag, bytes), signature).is_ok()
-}
-
-fn signature_input(tag: &[u8], bytes: &[u8]) -> Vec<u8> {
-    let mut input = Vec::with_capacity(tag.len() + 1 + bytes.len());
-    input.extend_from_slice(tag);
-    input.push(DOMAIN_SEPARATOR);
-    input.extend_from_slice(bytes);
-    input
-}
-
-fn sign(key: &DeviceIdentity, tag: &[u8], bytes: &[u8]) -> Result<String, EnrollmentError> {
-    // `DeviceIdentity` publishes no general signing operation, so the key is
-    // rebuilt from its own PKCS#8 encoding; the encoding is wiped when the
-    // wrapper drops.
-    let pkcs8 = key.to_pkcs8_der().map_err(|_| EnrollmentError::ResponseNotProduced)?;
-    let signing_key = SigningKey::from_pkcs8_der(pkcs8.as_slice()).map_err(|_| EnrollmentError::ResponseNotProduced)?;
-
-    let signature: Signature = signing_key.sign(&signature_input(tag, bytes));
-    let canonical = signature.normalize_s().unwrap_or(signature);
-
-    Ok(BASE64_URL_NO_PAD.encode(canonical.to_bytes()))
-}
-
-/// The device's public point, recovered from the DER encoding the identity
-/// publishes so that one prefix constant governs both the fingerprint and the
-/// wire form.
-fn device_public_point(key: &DeviceIdentity) -> Result<[u8; PUBLIC_KEY_OCTETS], EnrollmentError> {
-    key.public_key_der()
-        .strip_prefix(&SPKI_PREFIX)
-        .and_then(|point| <[u8; PUBLIC_KEY_OCTETS]>::try_from(point).ok())
-        .ok_or(EnrollmentError::ResponseNotProduced)
-}
-
-/// Decode an uncompressed SEC1 point and check that it is on the curve.
-///
-/// The length and alphabet are checked before decoding so that a padded or
-/// standard-alphabet spelling is refused, and the point tag is checked so that
-/// the compressed and hybrid forms — which no keyId would match — cannot be
-/// spelled at all.
-fn decode_public_key(value: &str) -> Option<(VerifyingKey, [u8; PUBLIC_KEY_OCTETS])> {
-    let value = value.as_bytes();
-    if value.len() != PUBLIC_KEY_CHARS || !value.iter().all(|byte| is_base64url(*byte)) {
-        return None;
-    }
-
-    let point: [u8; PUBLIC_KEY_OCTETS] = BASE64_URL_NO_PAD.decode(value).ok()?.try_into().ok()?;
-    if point[0] != UNCOMPRESSED_POINT {
-        return None;
-    }
-
-    VerifyingKey::from_sec1_bytes(&point).ok().map(|key| (key, point))
-}
-
-/// Lowercase SHA-256 hex of the DER SubjectPublicKeyInfo built from a 65 octet
-/// uncompressed point.
-fn key_id(point: &[u8]) -> String {
-    let mut digest = Sha256::new();
-    digest.update(SPKI_PREFIX);
-    digest.update(point);
-    hex_simd::encode_to_string(digest.finalize(), hex_simd::AsciiCase::Lower)
-}
-
-fn is_base64url(byte: u8) -> bool {
-    byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_'
-}
-
-/// Parse `YYYY-MM-DDTHH:MM:SSZ` into a Unix instant.
-///
-/// The shape is checked before the fields are read: offsets other than `Z` and
-/// fractional seconds are refused rather than normalised, so two producers
-/// cannot spell the same instant two ways.
-fn parse_timestamp(value: &str) -> Result<i64, EnrollmentError> {
-    let octets = value.as_bytes();
-    if octets.len() != TIMESTAMP_CHARS
-        || octets[4] != b'-'
-        || octets[7] != b'-'
-        || octets[10] != b'T'
-        || octets[13] != b':'
-        || octets[16] != b':'
-        || octets[19] != b'Z'
-    {
-        return Err(EnrollmentError::MalformedDocument);
-    }
-
-    let field = |range: std::ops::Range<usize>| -> Result<u32, EnrollmentError> {
-        let text = &value[range];
-        if !text.bytes().all(|byte| byte.is_ascii_digit()) {
-            return Err(EnrollmentError::MalformedDocument);
-        }
-        text.parse().map_err(|_| EnrollmentError::MalformedDocument)
-    };
-
-    let month = Month::try_from(field(5..7)? as u8).map_err(|_| EnrollmentError::MalformedDocument)?;
-    let date = Date::from_calendar_date(field(0..4)? as i32, month, field(8..10)? as u8)
-        .map_err(|_| EnrollmentError::MalformedDocument)?;
-    let clock = Time::from_hms(field(11..13)? as u8, field(14..16)? as u8, field(17..19)? as u8)
-        .map_err(|_| EnrollmentError::MalformedDocument)?;
-
-    Ok(PrimitiveDateTime::new(date, clock).assume_utc().unix_timestamp())
-}
-
-fn format_timestamp(unix: i64) -> Result<String, EnrollmentError> {
-    let moment = OffsetDateTime::from_unix_timestamp(unix).map_err(|_| EnrollmentError::ResponseNotProduced)?;
-    Ok(format!(
-        "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z",
-        moment.year(),
-        u8::from(moment.month()),
-        moment.day(),
-        moment.hour(),
-        moment.minute(),
-        moment.second()
-    ))
-}
-
-/// `at` must fall within `[issuedAt - 300, expiresAt + 300]`.
-///
-/// The declared expiry is capped at the protocol's maximum challenge lifetime
-/// because the issuer sets both ends of its own window; a challenge claiming a
-/// longer life expires at the bound.
-fn check_challenge_window(issued_at: i64, expires_at: i64, at: i64) -> Result<(), EnrollmentError> {
-    if at < issued_at.saturating_sub(CLOCK_SKEW_TOLERANCE) {
-        return Err(EnrollmentError::ChallengeNotYetValid);
-    }
-
-    let effective_expiry = expires_at.min(issued_at.saturating_add(MAX_CHALLENGE_LIFETIME));
-    if at > effective_expiry.saturating_add(CLOCK_SKEW_TOLERANCE) {
-        return Err(EnrollmentError::ChallengeExpired);
-    }
-
-    Ok(())
-}
@@ -1,72 +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.
-
-//! On-disk home of the offline enrollment key.
-//!
-//! An air-gapped device enrols with a key that is not its online device
-//! identity: the online key is minted during a registration exchange this
-//! device cannot perform, and an operator who carries an enrolment response out
-//! on removable media is enrolling exactly one key that Connect will pin. Losing
-//! it means asking for a fresh challenge, so it is written durably and published
-//! exactly once.
-//!
-//! The durability protocol is not reimplemented here. [`IdentityStore`] already
-//! seals a P-256 key at mode 0600, fsyncs it, and publishes it through a
-//! no-clobber link so a retry or a concurrent start converges on one key; it is
-//! pointed at a directory of this key's own rather than generalised into a
-//! key-store abstraction that would have to describe both lifecycles.
-
-use std::path::{Path, PathBuf};
-
-use super::super::identity::DeviceIdentity;
-use super::super::identity_store::{IdentityStore, StoreError};
-
-/// Subdirectory holding the offline enrolment key, kept apart from the online
-/// device identity so neither can be read in place of the other.
-const OFFLINE_DIRECTORY: &str = "offline";
-
-/// The offline enrolment key of one deployment.
-#[derive(Clone, Debug)]
-pub struct OfflineKeyStore {
-    inner: IdentityStore,
-}
-
-impl OfflineKeyStore {
-    pub fn new(directory: impl AsRef<Path>) -> Self {
-        Self {
-            inner: IdentityStore::new(directory.as_ref().join(OFFLINE_DIRECTORY)),
-        }
-    }
-
-    pub fn key_path(&self) -> PathBuf {
-        self.inner.key_path()
-    }
-
-    /// Return the stored key, or `None` when this deployment has never enrolled
-    /// offline. Reading never creates one, so a deployment that only ever
-    /// registers online holds no offline key.
-    pub fn load(&self) -> Result<Option<DeviceIdentity>, StoreError> {
-        self.inner.load()
-    }
-
-    /// Return the stored key, generating and publishing one the first time.
-    ///
-    /// A second enrolment attempt returns the original key rather than minting a
-    /// replacement: the operator may already be carrying a response for it, and
-    /// two keys would mean the response and the device disagree about which one
-    /// Connect pinned.
-    pub fn load_or_create(&self) -> Result<DeviceIdentity, StoreError> {
-        self.inner.load_or_create()
-    }
-}
@@ -1,35 +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.
-
-//! Offline enrolment: joining a Connect tenant without a network.
-//!
-//! An air-gapped cluster cannot perform the registration exchange, so an
-//! operator carries a signed challenge in and a signed response out. The device
-//! half of that exchange lives here: verifying the challenge against a root
-//! whose fingerprint is compiled into this binary, minting the key being
-//! enrolled, and signing the response.
-//!
-//! Nothing here opens a socket. That is the point of the surface, and it is
-//! asserted rather than assumed: the enrolment path takes bytes and returns
-//! bytes.
-//!
-//! The trust model, the signing convention, and every rejection reason are
-//! frozen by `protocol/agent/v1/fixtures/offline-enrollment/` and by
-//! `docs/adr/0009-offline-signing.md` on the Connect side.
-
-pub mod enrollment;
-pub mod key_store;
-
-pub use enrollment::{EnrollmentError, OfflineEnrollment, VerifiedChallenge};
-pub use key_store::OfflineKeyStore;
@@ -1,951 +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.
-
-//! Offline enrollment conformance against the frozen Connect fixtures.
-//!
-//! The device half of the air-gapped exchange verifies a challenge Connect
-//! signed and produces a response Connect will verify. Neither side can talk to
-//! the other while it does so, which means every disagreement about encoding,
-//! trust, or clock windows surfaces as a failed enrollment in the field rather
-//! than as an error at development time. The fixtures under
-//! `protocol/agent/v1/fixtures/offline-enrollment/` are the shared statement of
-//! what both sides must do, so this suite replays them rather than restating
-//! them: accept vectors must be accepted with the fields the document carries,
-//! reject vectors must fail with the single reason `error-codes.json` freezes,
-//! and the signature encoding rules in `trust-model.json` must hold even where
-//! the underlying ECDSA library is happy.
-
-use std::fs;
-use std::path::PathBuf;
-
-use base64::Engine as _;
-use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
-use base64::engine::general_purpose::URL_SAFE_NO_PAD as BASE64_URL_NO_PAD;
-use rustfs::connect::identity::DeviceIdentity;
-use rustfs::connect::offline::{EnrollmentError, OfflineEnrollment, VerifiedChallenge};
-use serde_json::Value;
-use sha2::{Digest as _, Sha256};
-
-/// DER prefix of a P-256 `SubjectPublicKeyInfo`, frozen by
-/// `trust-model.json` as `signature.subjectPublicKeyInfoDerPrefix`. The 65
-/// octet uncompressed point follows it, so a SEC1 point published in a fixture
-/// becomes a decodable public key by concatenation.
-const SPKI_PREFIX_HEX: &str = "3059301306072a8648ce3d020106082a8648ce3d030107034200";
-
-/// `clockSkew.toleranceSeconds` in `trust-model.json`.
-const SKEW_TOLERANCE_SECONDS: i64 = 300;
-
-// ---------------------------------------------------------------------------
-// Fixture access
-// ---------------------------------------------------------------------------
-
-fn fixture_dir() -> PathBuf {
-    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../protocol/agent/v1/fixtures/offline-enrollment")
-}
-
-fn sha256_hex(bytes: &[u8]) -> String {
-    Sha256::digest(bytes).iter().map(|byte| format!("{byte:02x}")).collect()
-}
-
-/// Read one fixture file and refuse it unless its bytes match the digest
-/// `MANIFEST.sha256` freezes.
-///
-/// Every vector in this suite arrives through here. A fixture edited on this
-/// side therefore fails the tests that depend on it instead of quietly
-/// redefining what conformance means, which is the failure mode a
-/// fixture-driven suite is otherwise blind to.
-fn read_fixture(name: &str) -> Vec<u8> {
-    let dir = fixture_dir();
-    let manifest = fs::read_to_string(dir.join("MANIFEST.sha256")).expect("read MANIFEST.sha256");
-
-    let expected = manifest
-        .lines()
-        .filter(|line| !line.trim().is_empty())
-        .find_map(|line| {
-            let (digest, file) = line
-                .split_once("  ")
-                .unwrap_or_else(|| panic!("malformed manifest line: {line}"));
-            (file == name).then(|| digest.to_string())
-        })
-        .unwrap_or_else(|| panic!("{name} is not listed in MANIFEST.sha256"));
-
-    let bytes = fs::read(dir.join(name)).unwrap_or_else(|error| panic!("read {name}: {error}"));
-    assert_eq!(sha256_hex(&bytes), expected, "{name} does not match the digest MANIFEST.sha256 freezes");
-    bytes
-}
-
-fn fixture_json(name: &str) -> Value {
-    serde_json::from_slice(&read_fixture(name)).unwrap_or_else(|error| panic!("{name} parses: {error}"))
-}
-
-fn accept_vectors() -> Value {
-    fixture_json("accept-vectors.json")
-}
-
-fn reject_vectors() -> Value {
-    fixture_json("reject-vectors.json")
-}
-
-fn trust_model() -> Value {
-    fixture_json("trust-model.json")
-}
-
-fn vector_list(fixture: &Value) -> Vec<Value> {
-    fixture["vectors"].as_array().expect("fixture carries a vector list").clone()
-}
-
-fn field<'a>(value: &'a Value, key: &str) -> &'a str {
-    value[key]
-        .as_str()
-        .unwrap_or_else(|| panic!("expected a string at '{key}' in {value}"))
-}
-
-/// The octets an operator carries in on removable media.
-///
-/// The fixture's `document` object *is* the transmitted artifact: a padded
-/// base64 `bytes` field holding the raw signed octets, plus the detached
-/// signature over them. Only `bytes` is covered by the signature, so
-/// re-serialising the surrounding envelope here cannot change what a verifier
-/// checks.
-fn envelope(document: &Value) -> Vec<u8> {
-    serde_json::to_vec(document).expect("envelope serialises")
-}
-
-/// The raw octets the signature covers, exactly as transmitted.
-fn signed_octets(document: &Value) -> Vec<u8> {
-    BASE64_STANDARD
-        .decode(field(document, "bytes"))
-        .expect("document bytes are padded base64")
-}
-
-/// The parsed signed document. Parsing is a convenience for the assertions
-/// below; the implementation under test is required to verify before it parses.
-fn signed_document(document: &Value) -> Value {
-    serde_json::from_slice(&signed_octets(document)).expect("signed document parses")
-}
-
-fn unix(rfc3339: &str) -> i64 {
-    chrono::DateTime::parse_from_rfc3339(rfc3339)
-        .unwrap_or_else(|error| panic!("'{rfc3339}' is not RFC 3339: {error}"))
-        .timestamp()
-}
-
-fn hex_to_bytes(hex: &str) -> Vec<u8> {
-    (0..hex.len())
-        .step_by(2)
-        .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).expect("valid hex"))
-        .collect()
-}
-
-/// Turn a fixture's unpadded-base64url SEC1 point into a usable verifying key.
-fn verifying_key(sec1_base64url: &str) -> p256::ecdsa::VerifyingKey {
-    let point = BASE64_URL_NO_PAD.decode(sec1_base64url).expect("public key is base64url");
-    assert_eq!(point.len(), 65, "the protocol freezes a 65 octet uncompressed SEC1 point");
-
-    let mut der = hex_to_bytes(SPKI_PREFIX_HEX);
-    der.extend_from_slice(&point);
-    <p256::ecdsa::VerifyingKey as p256::pkcs8::DecodePublicKey>::from_public_key_der(&der).expect("public key decodes")
-}
-
-fn published_key(role_or_name: &str) -> Value {
-    fixture_json("trust-chain.json")["keys"]
-        .as_array()
-        .expect("trust chain publishes keys")
-        .iter()
-        .find(|key| field(key, "name") == role_or_name)
-        .unwrap_or_else(|| panic!("trust-chain.json publishes no key named '{role_or_name}'"))
-        .clone()
-}
-
-/// `signatureInput = domainSeparationTag || 0x00 || the received octets`, the
-/// rule `trust-model.json` freezes under `domainSeparation`.
-fn signing_input(artifact_tag: &str, received: &[u8]) -> Vec<u8> {
-    let mut input = artifact_tag.as_bytes().to_vec();
-    input.push(0x00);
-    input.extend_from_slice(received);
-    input
-}
-
-fn domain_tag(artifact: &str) -> String {
-    let model = trust_model();
-    assert_eq!(
-        field(&model["domainSeparation"], "separatorByte"),
-        "0x00",
-        "the separator byte this suite encodes is the one the trust model freezes"
-    );
-    field(&model["domainSeparation"]["tags"], artifact).to_string()
-}
-
-/// Locate an accept vector by the name other vectors reference it by.
-fn accept_vector_named(name: &str) -> Value {
-    vector_list(&accept_vectors())
-        .into_iter()
-        .find(|vector| field(vector, "name") == name)
-        .unwrap_or_else(|| panic!("accept-vectors.json carries no vector named '{name}'"))
-}
-
-/// Verify the challenge a response vector answers, at that challenge's own
-/// evaluation time.
-fn answered_challenge(response_vector: &Value) -> (Value, VerifiedChallenge) {
-    let challenge_vector = accept_vector_named(field(response_vector, "answersChallenge"));
-    let now = unix(field(&challenge_vector, "evaluationTime"));
-    let verified = OfflineEnrollment::verify_challenge(&envelope(&challenge_vector["document"]), now)
-        .expect("the answered challenge is an accept vector and must verify");
-    (challenge_vector, verified)
-}
-
-fn device_nonce_of(document: &Value) -> [u8; 32] {
-    let raw = BASE64_URL_NO_PAD
-        .decode(field(&signed_document(document), "deviceNonce"))
-        .expect("deviceNonce is base64url");
-    raw.try_into().expect("replay.nonceLengthBytes freezes a 32 octet nonce")
-}
-
-// ---------------------------------------------------------------------------
-// Accept vectors
-// ---------------------------------------------------------------------------
-
-/// Every challenge accept vector must verify at its own evaluation time and
-/// expose exactly what the signed document says.
-///
-/// Two of these vectors sit on the skew boundary — 120 seconds before
-/// `issuedAt` and 300 seconds after `expiresAt` — so a verifier that compares
-/// against the raw window instead of the tolerated one fails here rather than
-/// in an air-gapped data centre. `challenge_proof` is pinned to the challenge's
-/// own detached signature value because that is what the response has to echo;
-/// deriving it from anything else would silently break the binding.
-#[test]
-fn every_challenge_accept_vector_verifies_and_exposes_the_signed_fields() {
-    let mut verified_count = 0usize;
-
-    for vector in vector_list(&accept_vectors()) {
-        if field(&vector, "artifact") != "challenge" {
-            continue;
-        }
-
-        let name = field(&vector, "name");
-        let document = &vector["document"];
-        let now = unix(field(&vector, "evaluationTime"));
-
-        let verified = OfflineEnrollment::verify_challenge(&envelope(document), now)
-            .unwrap_or_else(|error| panic!("accept vector '{name}' must verify: {}", error.reason()));
-
-        let signed = signed_document(document);
-        assert_eq!(verified.challenge_id, field(&signed, "challengeId"), "vector '{name}' challengeId");
-        assert_eq!(
-            verified.organization_name,
-            field(&signed, "organizationName"),
-            "vector '{name}' organizationName"
-        );
-        assert_eq!(verified.cluster_name, field(&signed, "clusterName"), "vector '{name}' clusterName");
-        assert_eq!(verified.nonce, field(&signed, "nonce"), "vector '{name}' nonce");
-        assert_eq!(verified.issued_at, field(&signed, "issuedAt"), "vector '{name}' issuedAt");
-        assert_eq!(verified.expires_at, field(&signed, "expiresAt"), "vector '{name}' expiresAt");
-        assert_eq!(verified.connect_key_id, field(&signed, "connectKeyId"), "vector '{name}' connectKeyId");
-        assert_eq!(
-            verified.challenge_proof,
-            field(&document["signature"], "value"),
-            "vector '{name}' must carry the challenge's own signature as the proof a response echoes"
-        );
-
-        verified_count += 1;
-    }
-
-    assert_eq!(
-        verified_count, 3,
-        "accept-vectors.json publishes three challenge vectors; a fourth is a protocol change"
-    );
-}
-
-/// Connect's own producer wrote the response accept vectors. Rebuilding them
-/// from the challenge they answer, with the device nonce and production time
-/// they used, must reproduce every field that does not depend on which device
-/// key signed — including the discarded-unknown-field vector, whose extra
-/// `telemetryHint` must not survive into anything this side produces.
-#[test]
-fn response_accept_vectors_are_reproduced_field_for_field_by_build_response() {
-    let key = DeviceIdentity::generate();
-    let mut reproduced = 0usize;
-
-    for vector in vector_list(&accept_vectors()) {
-        if field(&vector, "artifact") != "response" {
-            continue;
-        }
-
-        let name = field(&vector, "name");
-        let published = signed_document(&vector["document"]);
-        let (_, challenge) = answered_challenge(&vector);
-
-        let produced_at = unix(field(&published, "producedAt"));
-        let nonce = device_nonce_of(&vector["document"]);
-
-        let built_envelope: Value = serde_json::from_slice(
-            &OfflineEnrollment::build_response(&challenge, &key, &nonce, produced_at)
-                .unwrap_or_else(|error| panic!("vector '{name}' must be reproducible: {}", error.reason())),
-        )
-        .expect("the built response is JSON");
-        let built = signed_document(&built_envelope);
-
-        for shared in [
-            "formatVersion",
-            "protocolVersion",
-            "challengeId",
-            "organizationName",
-            "clusterName",
-            "challengeNonce",
-            "challengeProof",
-            "deviceNonce",
-        ] {
-            assert_eq!(
-                field(&built, shared),
-                field(&published, shared),
-                "vector '{name}' field {shared} must match the response Connect published"
-            );
-        }
-        assert_eq!(
-            unix(field(&built, "producedAt")),
-            produced_at,
-            "vector '{name}' producedAt must be the instant it was given"
-        );
-
-        // `versioning.additive` says an unknown optional field is discarded and
-        // never echoed back; a producer that copied the challenge or a previous
-        // response wholesale would carry it forward.
-        assert!(
-            built.get("telemetryHint").is_none(),
-            "vector '{name}' must not echo an unknown optional field"
-        );
-
-        reproduced += 1;
-    }
-
-    assert_eq!(
-        reproduced, 2,
-        "accept-vectors.json publishes two response vectors; a third is a protocol change"
-    );
-}
-
-// ---------------------------------------------------------------------------
-// Reject vectors
-// ---------------------------------------------------------------------------
-
-/// Every challenge reject vector must fail, and fail for the one reason
-/// `error-codes.json` freezes.
-///
-/// Asserting only that verification failed would pass for an implementation
-/// that rejects everything, and would let a tampered document be reported as an
-/// expiry — a rejection reason is what an operator acts on, so it is part of the
-/// contract rather than a diagnostic detail.
-#[test]
-fn every_challenge_reject_vector_fails_with_its_frozen_reason() {
-    let known_reasons: Vec<String> = fixture_json("error-codes.json")["reasons"]
-        .as_array()
-        .expect("error-codes.json carries reasons")
-        .iter()
-        .map(|entry| field(entry, "reason").to_string())
-        .collect();
-
-    let mut rejected = 0usize;
-
-    for vector in vector_list(&reject_vectors()) {
-        if field(&vector, "artifact") != "challenge" {
-            continue;
-        }
-
-        let name = field(&vector, "name");
-        let expected = field(&vector["expected"], "reason");
-        assert!(
-            known_reasons.iter().any(|reason| reason == expected),
-            "vector '{name}' names reason {expected}, which error-codes.json does not freeze"
-        );
-
-        let now = unix(field(&vector, "evaluationTime"));
-        let error = OfflineEnrollment::verify_challenge(&envelope(&vector["document"]), now)
-            .expect_err(&format!("reject vector '{name}' must not verify"));
-
-        assert_eq!(error.reason(), expected, "vector '{name}' must fail as {expected}");
-        rejected += 1;
-    }
-
-    assert_eq!(
-        rejected, 8,
-        "reject-vectors.json publishes eight challenge vectors; losing one silently narrows the suite"
-    );
-}
-
-/// The response reject vectors are artifacts Connect refuses. This side never
-/// verifies a response, so the device-side statement is the stronger one: given
-/// the challenge each vector answers, `build_response` must not be capable of
-/// emitting that artifact in the first place.
-///
-/// Each arm pins the specific field a compromised or careless producer would
-/// have to get wrong, so an implementation that copied values out of the wrong
-/// place — the response's own document, an operator-supplied argument, a
-/// previous exchange — fails here.
-#[test]
-fn response_reject_vectors_are_artifacts_build_response_cannot_emit() {
-    let key = DeviceIdentity::generate();
-    let mut covered = 0usize;
-
-    for vector in vector_list(&reject_vectors()) {
-        if field(&vector, "artifact") != "response" {
-            continue;
-        }
-
-        let name = field(&vector, "name");
-        let refused = signed_document(&vector["document"]);
-        let (_, challenge) = answered_challenge(&vector);
-        let produced_at = unix(field(&refused, "producedAt"));
-        let nonce = device_nonce_of(&vector["document"]);
-
-        let outcome = OfflineEnrollment::build_response(&challenge, &key, &nonce, produced_at);
-
-        match field(&vector["expected"], "reason") {
-            // `responseWindow` in trust-model.json: a device that emits a
-            // response outside the tolerated challenge window has produced an
-            // artifact Connect will refuse, so the refusal belongs here rather
-            // than at the far end of a courier run.
-            "CHALLENGE_EXPIRED" => {
-                let error = outcome.expect_err(&format!("vector '{name}': producing this response must be refused"));
-                assert_eq!(error.reason(), "CHALLENGE_EXPIRED", "vector '{name}' must refuse as CHALLENGE_EXPIRED");
-                covered += 1;
-                continue;
-            }
-            reason => {
-                let built_envelope: Value = serde_json::from_slice(
-                    &outcome.unwrap_or_else(|error| panic!("vector '{name}' baseline must build: {}", error.reason())),
-                )
-                .expect("the built response is JSON");
-                let built = signed_document(&built_envelope);
-
-                match reason {
-                    "ORGANIZATION_MISMATCH" => {
-                        assert_ne!(
-                            field(&refused, "organizationName"),
-                            challenge.organization_name,
-                            "vector '{name}' is only a mismatch if it names another organization"
-                        );
-                        assert_eq!(
-                            field(&built, "organizationName"),
-                            challenge.organization_name,
-                            "vector '{name}': the organization must come from the challenge, never from elsewhere"
-                        );
-                    }
-                    "CLUSTER_MISMATCH" => {
-                        assert_ne!(
-                            field(&refused, "clusterName"),
-                            challenge.cluster_name,
-                            "vector '{name}' is only a mismatch if it names another cluster"
-                        );
-                        assert_eq!(
-                            field(&built, "clusterName"),
-                            challenge.cluster_name,
-                            "vector '{name}': the cluster must come from the challenge, never from elsewhere"
-                        );
-                    }
-                    "CHALLENGE_PROOF_INVALID" => {
-                        // Two distinct vectors land here: a nonce the challenge
-                        // never carried, and a proof lifted from another
-                        // challenge. Both must be impossible to produce.
-                        assert_eq!(
-                            field(&built, "challengeNonce"),
-                            challenge.nonce,
-                            "vector '{name}': the echoed nonce must be the challenge's own"
-                        );
-                        assert_eq!(
-                            field(&built, "challengeProof"),
-                            challenge.challenge_proof,
-                            "vector '{name}': the proof must be the answered challenge's signature"
-                        );
-                        assert!(
-                            field(&refused, "challengeNonce") != challenge.nonce
-                                || field(&refused, "challengeProof") != challenge.challenge_proof,
-                            "vector '{name}' must differ from the challenge in nonce or proof to be rejectable"
-                        );
-                    }
-                    "DEVICE_PROOF_INVALID" => {
-                        // The refused vector presents one key and is signed by
-                        // another; hold the fixture to that claim, then require
-                        // the built response to be the opposite. Proof of
-                        // possession is the only thing that makes presenting a
-                        // key in an unauthenticated document safe.
-                        use p256::ecdsa::signature::Verifier as _;
-
-                        let presented = verifying_key(field(&refused, "devicePublicKey"));
-                        let raw = BASE64_URL_NO_PAD
-                            .decode(field(&vector["document"]["signature"], "value"))
-                            .expect("signature is base64url");
-                        let signature = p256::ecdsa::Signature::from_slice(&raw).expect("signature parses");
-                        assert!(
-                            presented
-                                .verify(
-                                    &signing_input(&domain_tag("enrollmentResponse"), &signed_octets(&vector["document"])),
-                                    &signature
-                                )
-                                .is_err(),
-                            "vector '{name}' is only a possession failure if it does not verify under the key it presents"
-                        );
-
-                        assert_response_proves_possession(&built_envelope, name);
-                    }
-                    "UNSUPPORTED_FORMAT" => {
-                        assert_ne!(
-                            field(&refused, "formatVersion"),
-                            field(&built, "formatVersion"),
-                            "vector '{name}' is only unsupported if it names another format version"
-                        );
-                        assert_eq!(
-                            field(&built, "formatVersion"),
-                            "rustfs.connect.offline.enrollmentResponse/1",
-                            "vector '{name}': the format version is frozen"
-                        );
-                    }
-                    "UNSUPPORTED_PROTOCOL" => {
-                        assert_ne!(
-                            field(&refused, "protocolVersion"),
-                            field(&built, "protocolVersion"),
-                            "vector '{name}' is only unsupported if it names another protocol major"
-                        );
-                        assert_eq!(field(&built, "protocolVersion"), "v1", "vector '{name}': the protocol major is frozen");
-                    }
-                    "ENROLLMENT_REPLAYED" => {
-                        // The vector claims to be a byte-identical replay of an
-                        // accepted response; hold it to that, because a replay
-                        // vector that is not byte identical proves nothing about
-                        // single use.
-                        let accepted = accept_vector_named("response binding the device public key and the challenge proof");
-                        assert_eq!(
-                            signed_octets(&vector["document"]),
-                            signed_octets(&accepted["document"]),
-                            "vector '{name}' must be the accepted response octet for octet"
-                        );
-                        assert_eq!(
-                            field(&vector["document"]["signature"], "value"),
-                            field(&accepted["document"]["signature"], "value"),
-                            "vector '{name}' must carry the accepted response's signature"
-                        );
-
-                        // A fresh device nonce is a different artifact, so a
-                        // second enrollment is never mistaken for a replay of
-                        // the first.
-                        let other = OfflineEnrollment::build_response(&challenge, &key, &[0x5a; 32], produced_at)
-                            .expect("a second response builds");
-                        assert_ne!(
-                            signed_octets(&built_envelope),
-                            signed_octets(&serde_json::from_slice::<Value>(&other).expect("JSON")),
-                            "vector '{name}': a different device nonce must yield a different artifact"
-                        );
-                    }
-                    other => panic!("vector '{name}' names an unhandled reason {other}; extend this test"),
-                }
-            }
-        }
-
-        covered += 1;
-    }
-
-    assert_eq!(
-        covered, 9,
-        "reject-vectors.json publishes nine response vectors; losing one silently narrows the suite"
-    );
-}
-
-// ---------------------------------------------------------------------------
-// Signature encoding
-// ---------------------------------------------------------------------------
-
-/// The high-S malleation is the rejection the whole encoding rule exists for.
-///
-/// `(r, n - s)` is a second valid signature over the same document under the
-/// same key. Every mainstream ECDSA library verifies it, so an implementation
-/// that hands the decoded octets straight to `p256` accepts a forged-looking
-/// duplicate of a genuine challenge — and because the 64 octets differ, that
-/// duplicate is a distinct artifact identity that slips past any deduplication
-/// keyed on the signature. This test proves the rejection came from the
-/// encoding rule and not from a failed verification: it first shows the
-/// malleated signature verifying mathematically, then requires
-/// `verify_challenge` to refuse it as SIGNATURE_NOT_CANONICAL.
-#[test]
-fn malleated_high_s_signature_is_refused_although_it_verifies_mathematically() {
-    use p256::ecdsa::signature::Verifier as _;
-
-    let model = trust_model();
-    let malleated = model["rejectedSignatureEncodings"]
-        .as_array()
-        .expect("trust-model.json publishes rejected encodings")
-        .iter()
-        .find(|entry| field(entry, "reason") == "SIGNATURE_NOT_CANONICAL")
-        .expect("trust-model.json publishes the high-S malleation")
-        .clone();
-    assert!(
-        malleated["acceptedByALenientVerifier"].as_bool() == Some(true),
-        "this vector is only interesting because a lenient verifier accepts it"
-    );
-
-    let vector = accept_vector_named("challenge signed by a chained signing key under the pinned root");
-    let genuine_value = field(&vector["document"]["signature"], "value").to_string();
-    let malleated_value = field(&malleated, "value").to_string();
-    assert_ne!(genuine_value, malleated_value, "the malleation must be a different encoding");
-
-    let genuine = BASE64_URL_NO_PAD.decode(&genuine_value).expect("signature is base64url");
-    let raw = BASE64_URL_NO_PAD.decode(&malleated_value).expect("signature is base64url");
-    assert_eq!(raw.len(), 64, "the malleation is well formed at 64 octets");
-    assert_eq!(raw[..32], genuine[..32], "the malleation shares r with the genuine signature");
-    assert_ne!(raw[32..], genuine[32..], "the malleation replaces s with n - s");
-
-    // Step one: the malleated pair really does verify under the signing key, so
-    // a verifier cannot be excused for accepting it on mathematical grounds.
-    let signature = p256::ecdsa::Signature::from_slice(&raw).expect("the malleated signature parses");
-    assert!(signature.normalize_s().is_some(), "the malleated signature must be the high-S form");
-    let key = verifying_key(field(&published_key("signing"), "publicKey"));
-    let input = signing_input(&domain_tag("enrollmentChallenge"), &signed_octets(&vector["document"]));
-    key.verify(&input, &signature)
-        .expect("the malleated signature must verify mathematically, or this test proves nothing");
-
-    // Step two: the implementation must refuse it anyway, and say why.
-    let mut tampered = vector["document"].clone();
-    tampered["signature"]["value"] = Value::String(malleated_value);
-
-    let now = unix(field(&vector, "evaluationTime"));
-    let error = OfflineEnrollment::verify_challenge(&envelope(&tampered), now)
-        .expect_err("a high-S signature must be refused even though it verifies");
-    assert_eq!(
-        error.reason(),
-        "SIGNATURE_NOT_CANONICAL",
-        "a malleated signature is a canonicality failure, not a verification failure"
-    );
-}
-
-/// Every encoding `trust-model.json` names as rejected must fail with the
-/// reason it names — DER, padded base64url, truncation, and out-of-range
-/// scalars alongside the malleation. Three of the five are accepted by a
-/// lenient verifier, so a single blanket "signature did not verify" answer would
-/// be both wrong and undiagnosable.
-#[test]
-fn every_rejected_signature_encoding_fails_with_its_frozen_reason() {
-    let vector = accept_vector_named("challenge signed by a chained signing key under the pinned root");
-    let now = unix(field(&vector, "evaluationTime"));
-    let model = trust_model();
-    let encodings = model["rejectedSignatureEncodings"]
-        .as_array()
-        .expect("trust-model.json publishes rejected encodings");
-
-    for entry in encodings {
-        let name = field(entry, "name");
-        let mut tampered = vector["document"].clone();
-        tampered["signature"]["value"] = Value::String(field(entry, "value").to_string());
-
-        let error: EnrollmentError = OfflineEnrollment::verify_challenge(&envelope(&tampered), now)
-            .err()
-            .unwrap_or_else(|| panic!("rejected encoding '{name}' must not verify"));
-
-        assert_eq!(error.reason(), field(entry, "reason"), "rejected encoding '{name}'");
-    }
-
-    assert_eq!(encodings.len(), 5, "trust-model.json freezes five rejected encodings");
-}
-
-// ---------------------------------------------------------------------------
-// Clock window
-// ---------------------------------------------------------------------------
-
-/// The tolerated window is `[issuedAt - 300, expiresAt + 300]`, inclusive at
-/// both ends. An air-gapped device has no synchronised clock, so an
-/// off-by-one here either strands a legitimate enrollment or widens the window
-/// a stolen challenge stays usable in. Both ends are checked at the exact bound
-/// and one second past it, and the reason distinguishes the two directions.
-#[test]
-fn challenge_is_accepted_at_the_exact_skew_bound_and_refused_one_second_past_it() {
-    let vector = accept_vector_named("challenge signed by a chained signing key under the pinned root");
-    let document = envelope(&vector["document"]);
-    let signed = signed_document(&vector["document"]);
-
-    let issued_at = unix(field(&signed, "issuedAt"));
-    let expires_at = unix(field(&signed, "expiresAt"));
-
-    let earliest = issued_at - SKEW_TOLERANCE_SECONDS;
-    OfflineEnrollment::verify_challenge(&document, earliest).expect("the earliest tolerated instant is inside the window");
-    let error =
-        OfflineEnrollment::verify_challenge(&document, earliest - 1).expect_err("one second earlier is outside the window");
-    assert_eq!(error.reason(), "CHALLENGE_NOT_YET_VALID");
-
-    let latest = expires_at + SKEW_TOLERANCE_SECONDS;
-    OfflineEnrollment::verify_challenge(&document, latest).expect("the latest tolerated instant is inside the window");
-    let error = OfflineEnrollment::verify_challenge(&document, latest + 1).expect_err("one second later is outside the window");
-    assert_eq!(error.reason(), "CHALLENGE_EXPIRED");
-}
-
-// ---------------------------------------------------------------------------
-// Response production
-// ---------------------------------------------------------------------------
-
-/// Assert a built response proves possession of the key it presents: the
-/// fingerprint matches the presented key, and the detached signature is a
-/// canonical low-S ES256 signature that verifies under that key over the exact
-/// octets transmitted.
-fn assert_response_proves_possession(built_envelope: &Value, label: &str) {
-    use p256::ecdsa::signature::Verifier as _;
-
-    let raw = signed_octets(built_envelope);
-    let built = signed_document(built_envelope);
-    let signature_block = &built_envelope["signature"];
-
-    assert_eq!(field(signature_block, "algorithm"), "ES256", "{label}: the algorithm is frozen");
-
-    let value = field(signature_block, "value");
-    assert_eq!(value.len(), 86, "{label}: the transfer encoding is 86 unpadded base64url characters");
-    assert!(
-        value.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_'),
-        "{label}: the signature must use the base64url alphabet with no padding"
-    );
-
-    let bytes = BASE64_URL_NO_PAD.decode(value).expect("signature is base64url");
-    assert_eq!(bytes.len(), 64, "{label}: the signature is a fixed-width r || s");
-    let signature = p256::ecdsa::Signature::from_slice(&bytes).expect("signature parses");
-    assert!(
-        signature.normalize_s().is_none(),
-        "{label}: this side must never emit the malleated high-S form it refuses to accept"
-    );
-
-    let presented = field(&built, "devicePublicKey");
-    let key = verifying_key(presented);
-    key.verify(&signing_input(&domain_tag("enrollmentResponse"), &raw), &signature)
-        .unwrap_or_else(|error| panic!("{label}: the response must verify under the key it presents: {error}"));
-
-    // `signature.keyIdAlgorithm`: the lowercase SHA-256 of the DER
-    // SubjectPublicKeyInfo, not of the bare point and not of the transfer
-    // encoding.
-    let mut spki = hex_to_bytes(SPKI_PREFIX_HEX);
-    spki.extend_from_slice(&BASE64_URL_NO_PAD.decode(presented).expect("public key is base64url"));
-    let fingerprint = sha256_hex(&spki);
-    assert_eq!(
-        field(&built, "deviceKeyId"),
-        fingerprint,
-        "{label}: deviceKeyId must be the fingerprint of the key the document presents"
-    );
-    assert_eq!(
-        field(signature_block, "keyId"),
-        fingerprint,
-        "{label}: the detached signature must name the same key"
-    );
-}
-
-/// A response is the only thing Connect will ever see from this device, so it
-/// has to carry the whole binding on its own: the challenge it answers, the
-/// proof that challenge was genuine, the key being enrolled, and possession of
-/// that key.
-#[test]
-fn built_response_binds_the_challenge_proof_and_proves_possession_of_the_device_key() {
-    let vector = accept_vector_named("response binding the device public key and the challenge proof");
-    let (challenge_vector, challenge) = answered_challenge(&vector);
-    let key = DeviceIdentity::generate();
-    let produced_at = unix(field(&signed_document(&vector["document"]), "producedAt"));
-
-    let bytes = OfflineEnrollment::build_response(&challenge, &key, &[0x11; 32], produced_at).expect("the response builds");
-    let built_envelope: Value = serde_json::from_slice(&bytes).expect("the response is JSON");
-    let built = signed_document(&built_envelope);
-
-    assert_response_proves_possession(&built_envelope, "built response");
-
-    // The proof is the challenge's own detached signature. A producer that
-    // echoed the nonce alone, or hashed something, would let a response be
-    // built from an unverified challenge.
-    assert_eq!(
-        field(&built, "challengeProof"),
-        field(&challenge_vector["document"]["signature"], "value"),
-        "the proof must be the signature of the challenge being answered"
-    );
-    assert_eq!(field(&built, "challengeNonce"), challenge.nonce);
-    assert_eq!(field(&built, "challengeId"), challenge.challenge_id);
-
-    assert_eq!(
-        field(&built, "devicePublicKey"),
-        BASE64_URL_NO_PAD.encode(&key.public_key_der()[hex_to_bytes(SPKI_PREFIX_HEX).len()..]),
-        "the presented key must be the key that was passed in"
-    );
-    assert_eq!(
-        field(&built, "deviceNonce"),
-        BASE64_URL_NO_PAD.encode([0x11; 32]),
-        "the device nonce must be the one that was passed in"
-    );
-    assert!(field(&built, "producedAt").ends_with('Z'), "producedAt is a UTC RFC 3339 instant");
-}
-
-/// The response leaves the air gap on removable media and is read by anyone who
-/// handles it. A producer that serialised the key pair instead of the public
-/// key, or logged a debug rendering into the document, would put the enrolled
-/// private key on that medium — and the enrollment would still succeed, so
-/// nothing else in this suite would notice.
-#[test]
-fn built_response_carries_no_private_key_material() {
-    let vector = accept_vector_named("response binding the device public key and the challenge proof");
-    let (_, challenge) = answered_challenge(&vector);
-    let key = DeviceIdentity::generate();
-    let produced_at = unix(field(&signed_document(&vector["document"]), "producedAt"));
-
-    let response = OfflineEnrollment::build_response(&challenge, &key, &[0x22; 32], produced_at).expect("the response builds");
-
-    // The envelope carries the signed document base64-encoded, so a needle
-    // present in the document is not present in the envelope octets. Both
-    // layers are searched: an operator handling the medium can read either.
-    let envelope_value: Value = serde_json::from_slice(&response).expect("the response is JSON");
-    let mut haystack = response;
-    haystack.extend_from_slice(&signed_octets(&envelope_value));
-
-    let pkcs8 = key.to_pkcs8_der().expect("serialise the key");
-    let secret = <p256::SecretKey as p256::pkcs8::DecodePrivateKey>::from_pkcs8_der(&pkcs8).expect("the key parses");
-    let scalar = secret.to_bytes();
-
-    // Every spelling the scalar could plausibly reach a document in: raw, and
-    // the three encodings this protocol already uses elsewhere.
-    let scalar_hex: String = scalar.iter().map(|byte| format!("{byte:02x}")).collect();
-    for (description, needle) in [
-        ("the PKCS#8 encoding", pkcs8.to_vec()),
-        ("the raw private scalar", scalar.to_vec()),
-        ("the scalar in base64url", BASE64_URL_NO_PAD.encode(scalar).into_bytes()),
-        ("the scalar in standard base64", BASE64_STANDARD.encode(scalar).into_bytes()),
-        ("the scalar in hex", scalar_hex.into_bytes()),
-    ] {
-        assert!(
-            !haystack.windows(needle.len()).any(|window| window == needle.as_slice()),
-            "the response must not contain {description}"
-        );
-    }
-
-    // The public half must be there, so the absence above is a statement about
-    // what was excluded rather than about a haystack that would not have found
-    // the private half either.
-    let point = BASE64_URL_NO_PAD.encode(&key.public_key_der()[hex_to_bytes(SPKI_PREFIX_HEX).len()..]);
-    assert!(
-        haystack.windows(point.len()).any(|window| window == point.as_bytes()),
-        "the response must still present the public key"
-    );
-}
-
-// ---------------------------------------------------------------------------
-// The offline invariant
-// ---------------------------------------------------------------------------
-
-/// The whole surface exists because there is no network. This asserts that
-/// three different ways, because no single one of them is conclusive on its own.
-///
-/// 1. The process opens no descriptor across a full verify-and-respond cycle. A
-///    socket, a DNS resolver, a pooled HTTP client, or a revocation-list fetch
-///    all show up here — including one that is opened and cached rather than
-///    opened and closed, which is what a lazily built client does.
-/// 2. The cycle is a pure byte transform: the same inputs produce the same
-///    verified fields, and the evaluation instant is an argument rather than an
-///    ambient read, so nothing about the outcome can depend on reachability.
-/// 3. Repeating the cycle changes nothing observable, so a first call cannot be
-///    quietly initialising shared state that a later one reuses.
-#[cfg(unix)]
-#[test]
-fn enrollment_opens_no_descriptor_and_is_a_pure_byte_transform() {
-    let vector = accept_vector_named("challenge signed by a chained signing key under the pinned root");
-    let document = envelope(&vector["document"]);
-    let now = unix(field(&vector, "evaluationTime"));
-    let key = DeviceIdentity::generate();
-
-    // Warm anything the test harness itself lazily opens before the baseline.
-    let _ = open_descriptors();
-    let baseline = open_descriptors();
-    assert!(
-        !baseline.is_empty(),
-        "the descriptor table must be readable for this test to mean anything"
-    );
-
-    let mut fields = Vec::new();
-    for _ in 0..2 {
-        let challenge = OfflineEnrollment::verify_challenge(&document, now).expect("the challenge verifies");
-        let response = OfflineEnrollment::build_response(&challenge, &key, &[0x33; 32], now).expect("the response builds");
-        fields.push((
-            challenge.challenge_id.clone(),
-            challenge.nonce.clone(),
-            challenge.challenge_proof.clone(),
-            signed_octets(&serde_json::from_slice::<Value>(&response).expect("JSON")),
-        ));
-    }
-
-    assert_eq!(
-        open_descriptors(),
-        baseline,
-        "the enrollment path must not open a descriptor: no socket, no resolver, no cached client"
-    );
-
-    let (first, second) = (&fields[0], &fields[1]);
-    assert_eq!(first.0, second.0, "verification must be deterministic");
-    assert_eq!(first.1, second.1, "verification must be deterministic");
-    assert_eq!(first.2, second.2, "verification must be deterministic");
-    assert_eq!(
-        first.3, second.3,
-        "the signed response octets are a function of the challenge, the key, the nonce, and the instant"
-    );
-}
-
-#[cfg(unix)]
-fn open_descriptors() -> Vec<String> {
-    // Linux publishes the table at /proc/self/fd; the BSDs and macOS at /dev/fd.
-    let path = if PathBuf::from("/proc/self/fd").is_dir() {
-        "/proc/self/fd"
-    } else {
-        "/dev/fd"
-    };
-
-    let mut entries: Vec<String> = fs::read_dir(path)
-        .unwrap_or_else(|error| panic!("read {path}: {error}"))
-        .map(|entry| entry.expect("read dir entry").file_name().to_string_lossy().into_owned())
-        .collect();
-    entries.sort();
-    entries
-}
-
-/// A descriptor count taken around a call cannot see a socket that was opened
-/// and closed inside it, so the invariant is also asserted where it can be
-/// stated absolutely: the implementation names no network API at all.
-///
-/// This is the shape the regression actually takes — someone adds a
-/// revocation-list fetch, a time-server check, or a "just confirm the challenge
-/// with Connect" call — and it is caught at the source rather than by observing
-/// its effects.
-#[test]
-fn enrollment_implementation_names_no_network_api() {
-    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("src/connect/offline/enrollment.rs");
-    let source = fs::read_to_string(&path).unwrap_or_else(|error| panic!("read {}: {error}", path.display()));
-
-    // Prose is allowed to discuss the invariant it is documenting, so only code
-    // is scanned.
-    let code: String = source
-        .lines()
-        .filter(|line| !line.trim_start().starts_with("//"))
-        .collect::<Vec<_>>()
-        .join("\n");
-
-    for forbidden in [
-        "std::net",
-        "tokio::net",
-        "TcpStream",
-        "TcpListener",
-        "UdpSocket",
-        "UnixStream",
-        "ToSocketAddrs",
-        "reqwest",
-        "hyper",
-        "tonic",
-    ] {
-        assert!(
-            !code.contains(forbidden),
-            "offline enrollment must not reach the network, but the implementation names {forbidden}"
-        );
-    }
-}
@@ -1,419 +0,0 @@
-#!/usr/bin/env python3
-"""Fail when committed tests silently fall out of their execution wiring."""
-
-from __future__ import annotations
-
-import hashlib
-import json
-import re
-import sys
-import tempfile
-import tomllib
-import unittest
-from unittest import mock
-from pathlib import Path
-
-
-ROOT = Path(__file__).resolve().parents[1]
-
-
-def words(value: str) -> set[str]:
-    return {item.strip() for item in value.split(",") if item.strip()}
-
-
-def rust_code_only(source: str) -> str:
-    """Blank Rust comments and literals while preserving byte positions."""
-    def quoted_end(quote_index: int, delimiter: str) -> int:
-        end = quote_index + 1
-        while end < len(source):
-            if source[end] == "\\":
-                end += 2
-            elif source[end] == delimiter:
-                return end + 1
-            else:
-                end += 1
-        return end
-
-    code = list(source)
-    index = 0
-    while index < len(source):
-        if source.startswith("//", index):
-            end = source.find("\n", index)
-            end = len(source) if end < 0 else end
-        elif source.startswith("/*", index):
-            depth = 1
-            end = index + 2
-            while end < len(source) and depth:
-                if source.startswith("/*", end):
-                    depth += 1
-                    end += 2
-                elif source.startswith("*/", end):
-                    depth -= 1
-                    end += 2
-                else:
-                    end += 1
-        else:
-            raw = re.match(r'(?:br|cr|r)(?P<hashes>#{0,})"', source[index:])
-            if raw:
-                marker = '"' + raw.group("hashes")
-                end = source.find(marker, index + raw.end())
-                end = len(source) if end < 0 else end + len(marker)
-            elif source[index] == '"' or source.startswith(('b"', 'c"'), index):
-                start_quote = index if source[index] == '"' else index + 1
-                end = quoted_end(start_quote, '"')
-            elif source[index] == "'" and index + 2 < len(source) and (
-                source[index + 1] == "\\" or source[index + 2] == "'"
-            ):
-                end = quoted_end(index, "'")
-            elif source.startswith("b'", index):
-                end = quoted_end(index + 1, "'")
-            else:
-                index += 1
-                continue
-
-        for offset in range(index, end):
-            if code[offset] != "\n":
-                code[offset] = " "
-        index = end
-    return "".join(code)
-
-
-def declared(parent: Path, module: str) -> bool:
-    source = parent.read_text()
-    code = rust_code_only(source)
-    pattern = re.compile(rf"^\s*(?:pub(?:\([^)]*\))?\s+)?mod\s+{re.escape(module)}\s*;", re.MULTILINE)
-    allowed_preambles = {
-        "#[cfg(test)]": "#[cfg(test)]",
-        "#[cfg(all(test,target_os=))]": '#[cfg(all(test,target_os="linux"))]',
-    }
-    for match in pattern.finditer(code):
-        prefix = code[: match.start()]
-        depths = {"(": 0, "[": 0, "{": 0}
-        pairs = {")": "(", "]": "[", "}": "{"}
-        for char in prefix:
-            if char in depths:
-                depths[char] += 1
-            elif char in pairs:
-                depths[pairs[char]] -= 1
-        if any(depths.values()):
-            continue
-
-        boundary = max(prefix.rfind(";"), prefix.rfind("{"), prefix.rfind("}"))
-        code_preamble = re.sub(r"\s+", "", prefix[boundary + 1 :])
-        if not code_preamble:
-            return True
-        if code_preamble in allowed_preambles:
-            attr_start = prefix.rfind("#[cfg", boundary + 1)
-            if attr_start >= 0 and re.sub(r"\s+", "", source[attr_start : match.start()]) == allowed_preambles[code_preamble]:
-                return True
-    return False
-
-
-def module_source(src: Path, directory: Path) -> Path | None:
-    if not directory.parts:
-        return src / "lib.rs"
-
-    mod_file = src / directory / "mod.rs"
-    if mod_file.is_file():
-        return mod_file
-
-    sibling = src.joinpath(*directory.parts[:-1], f"{directory.name}.rs")
-    return sibling if sibling.is_file() else None
-
-
-def check_e2e_modules(root: Path) -> list[str]:
-    src = root / "crates/e2e_test/src"
-    errors: list[str] = []
-    for test_file in sorted(src.rglob("*_test.rs")):
-        relative = test_file.relative_to(root).as_posix()
-        directory = test_file.relative_to(src).parent
-        parent = module_source(src, directory)
-        if parent is None:
-            errors.append(f"{relative}: no canonical parent module")
-            continue
-
-        if not declared(parent, test_file.stem):
-            errors.append(f"{relative}: not declared by {parent.relative_to(root).as_posix()}")
-        while directory.parts:
-            module = directory.name
-            directory = directory.parent
-            parent = module_source(src, directory)
-            if parent is None:
-                errors.append(f"{relative}: module {module} has no canonical parent")
-                break
-            if not declared(parent, module):
-                errors.append(f"{relative}: module {module} not declared by {parent.relative_to(root).as_posix()}")
-    return errors
-
-
-def check_fuzz_targets(root: Path) -> list[str]:
-    manifest = tomllib.loads((root / "fuzz/Cargo.toml").read_text())
-    expected = {item["name"] for item in manifest.get("bin", []) if "name" in item}
-    errors: list[str] = []
-    if not expected:
-        return ["fuzz/Cargo.toml: no [[bin]] fuzz targets found"]
-
-    runner = (root / "scripts/fuzz/run.sh").read_text()
-    match = re.search(r'^targets="([^"]+)"', runner, re.MULTILINE)
-    runner_targets = set(match.group(1).split()) if match else set()
-    if runner_targets != expected:
-        errors.append(f"scripts/fuzz/run.sh targets {sorted(runner_targets)} != manifest {sorted(expected)}")
-
-    workflow = (root / ".github/workflows/fuzz.yml").read_text()
-    matrices = [words(value) for value in re.findall(r"^\s*target:\s*\[([^]]+)]", workflow, re.MULTILINE)]
-    if len(matrices) != 2:
-        errors.append(f".github/workflows/fuzz.yml: expected smoke and nightly target matrices, found {len(matrices)}")
-    for index, matrix in enumerate(matrices, start=1):
-        if matrix != expected:
-            errors.append(f".github/workflows/fuzz.yml matrix {index} {sorted(matrix)} != manifest {sorted(expected)}")
-
-    runtime_targets = re.findall(r"^\s*FUZZ_TARGET:\s*(\S.*?)\s*$", workflow, re.MULTILINE)
-    if runtime_targets != ["${{ matrix.target }}", "${{ matrix.target }}"]:
-        errors.append(".github/workflows/fuzz.yml: smoke and nightly jobs must pass matrix.target to FUZZ_TARGET")
-
-    dependency_paths = {
-        f"{path.removeprefix('../')}/**"
-        for dependency in manifest.get("dependencies", {}).values()
-        if isinstance(dependency, dict)
-        and isinstance(path := dependency.get("path"), str)
-        and path.startswith("../crates/")
-    }
-    missing_paths = sorted(path for path in dependency_paths if f'"{path}"' not in workflow)
-    if missing_paths:
-        errors.append(f".github/workflows/fuzz.yml missing direct dependency paths: {', '.join(missing_paths)}")
-
-    staged_matches = re.findall(r"^\s*for target in ([^;]+); do", workflow, re.MULTILINE)
-    staged = set(staged_matches[0].split()) if staged_matches else set()
-    if staged != expected:
-        errors.append(f".github/workflows/fuzz.yml staged binaries {sorted(staged)} != manifest {sorted(expected)}")
-
-    return errors
-
-
-def check_runner_selection(root: Path) -> list[str]:
-    runner = (root / "scripts/run_e2e_tests.sh").read_text()
-    errors: list[str] = []
-    if "--include-ignored" not in runner:
-        errors.append("scripts/run_e2e_tests.sh: runner must include default and ignored tests")
-    if "--test-threads=1" not in runner:
-        errors.append("scripts/run_e2e_tests.sh: runner must serialize fixed-port protocol tests")
-    if re.search(r"(?<!include-)--ignored\b", runner):
-        errors.append("scripts/run_e2e_tests.sh: bare --ignored silently excludes default tests")
-    if "--exact" in runner:
-        errors.append("scripts/run_e2e_tests.sh: --test is documented as a pattern and must not force exact matching")
-    if 'eval "$test_cmd"' in runner:
-        errors.append("scripts/run_e2e_tests.sh: command construction must not use eval")
-    return errors
-
-
-def profile_selection(root: Path, profile: str) -> str:
-    if not re.fullmatch(r"e2e-[a-z0-9-]+", profile):
-        raise ValueError(f"invalid e2e profile name: {profile}")
-    path = root / f".config/{profile}-selection.txt"
-    lines = [line for line in path.read_text().splitlines() if line.strip()]
-    values = dict(line.split("=", 1) for line in lines if "=" in line)
-    if len(values) != len(lines) or any(not re.fullmatch(r"sha256(?:-[a-z0-9]+)?", key) for key in values):
-        raise ValueError(f"{path.relative_to(root).as_posix()}: invalid sha256 entry")
-    key = f"sha256-{sys.platform}"
-    digest = values.get(key, values.get("sha256", ""))
-    if not re.fullmatch(r"[0-9a-f]{64}", digest):
-        raise ValueError(f"{path.relative_to(root).as_posix()}: missing sha256 for {sys.platform}")
-    return digest
-
-
-def check_profile_definitions(root: Path) -> list[str]:
-    config = tomllib.loads((root / ".config/nextest.toml").read_text())
-    profiles = {
-        profile
-        for profile in config.get("profile", {})
-        if profile.startswith("e2e-")
-    }
-    selection_profiles = {
-        path.name.removesuffix("-selection.txt") for path in (root / ".config").glob("e2e-*-selection.txt")
-    }
-    errors: list[str] = []
-    for profile in sorted(profiles | selection_profiles):
-        if profile not in profiles:
-            errors.append(f".config/nextest.toml: missing profile.{profile}")
-        if profile not in selection_profiles:
-            errors.append(f".config/{profile}-selection.txt: missing expected profile selection")
-            continue
-        try:
-            profile_selection(root, profile)
-        except (FileNotFoundError, ValueError) as error:
-            errors.append(str(error))
-    return errors
-
-
-def check_profile_listing(root: Path, profile: str, listing: Path) -> list[str]:
-    try:
-        expected_digest = profile_selection(root, profile)
-        data = json.loads(listing.read_text())
-        selected = sorted(
-            f"{suite_id}::{test_name}"
-            for suite_id, suite in data["rust-suites"].items()
-            for test_name, testcase in suite["testcases"].items()
-            if testcase.get("filter-match", {}).get("status") == "matches"
-        )
-        digest = hashlib.sha256(("\n".join(selected) + "\n").encode()).hexdigest()
-    except (FileNotFoundError, KeyError, TypeError, ValueError, json.JSONDecodeError) as error:
-        return [f"cannot read {profile} nextest listing: {error}"]
-    if digest != expected_digest:
-        return [
-            f"{profile} selection changed: count={len(selected)} sha256={digest}; "
-            f"expected sha256={expected_digest}"
-        ]
-    print(f"{profile} selection OK: {len(selected)} tests, sha256={digest}")
-    return []
-
-
-def validate(root: Path) -> list[str]:
-    errors: list[str] = []
-    errors.extend(check_e2e_modules(root))
-    errors.extend(check_fuzz_targets(root))
-    errors.extend(check_runner_selection(root))
-    errors.extend(check_profile_definitions(root))
-    return errors
-
-
-class SelfTests(unittest.TestCase):
-    def test_e2e_requires_registration(self) -> None:
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp)
-            src = root / "crates/e2e_test/src"
-            src.mkdir(parents=True)
-            (src / "lib.rs").write_text("")
-            test_file = src / "boundary_test.rs"
-            test_file.write_text("#[test]\nfn boundary() {}\n")
-            self.assertEqual(len(check_e2e_modules(root)), 1)
-            (src / "lib.rs").write_text("mod boundary_test;\n")
-            self.assertEqual(check_e2e_modules(root), [])
-
-            nested = src / "protocols"
-            nested.mkdir()
-            (nested / "mod.rs").write_text("mod fixed_port_test;\n")
-            (nested / "fixed_port_test.rs").write_text("#[test]\nfn fixed_port() {}\n")
-            self.assertEqual(len(check_e2e_modules(root)), 1)
-            (src / "lib.rs").write_text("mod boundary_test;\nmod protocols;\n")
-            self.assertEqual(check_e2e_modules(root), [])
-
-            (src / "lib.rs").write_text("#[cfg(any())]\nmod boundary_test;\nmod protocols;\n")
-            self.assertEqual(len(check_e2e_modules(root)), 1)
-
-            (src / "lib.rs").write_text(
-                "#[cfg(any())]\n/// hidden module\nmod boundary_test;\n#[cfg_attr(test, cfg(any()))]\nmod protocols;\n"
-            )
-            self.assertEqual(len(check_e2e_modules(root)), 2)
-
-            (src / "lib.rs").write_text(
-                'const PHANTOM: &str = r#"{\nmod boundary_test;\n"#;\ndiscard! { mod protocols; }\n'
-            )
-            self.assertEqual(len(check_e2e_modules(root)), 2)
-
-            (src / "lib.rs").write_text(
-                '#[cfg(all(test, target_os = r"windows" /* target_os = "linux" */))]\n'
-                "mod boundary_test;\nmod protocols;\n"
-            )
-            self.assertEqual(len(check_e2e_modules(root)), 1)
-
-            (src / "lib.rs").write_text(
-                '#[cfg(all(test, target_os = r"windows"))] // #[cfg(all(test, target_os = "linux"))]\n'
-                "mod boundary_test;\nmod protocols;\n"
-            )
-            self.assertEqual(len(check_e2e_modules(root)), 1)
-
-    def test_fuzz_runtime_uses_matrix_target(self) -> None:
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp)
-            (root / "fuzz").mkdir()
-            (root / "scripts/fuzz").mkdir(parents=True)
-            (root / ".github/workflows").mkdir(parents=True)
-            (root / "fuzz/Cargo.toml").write_text(
-                'dep = { path = "../crates/dep" }\n[[bin]]\nname = "one"\n'
-            )
-            (root / "scripts/fuzz/run.sh").write_text('targets="one"\n')
-            (root / ".github/workflows/fuzz.yml").write_text(
-                'paths:\n  - "crates/dep/**"\n'
-                "target: [one]\nFUZZ_TARGET: fixed\n"
-                "target: [one]\nFUZZ_TARGET: ${{ matrix.target }}\n"
-                "for target in one; do\n"
-                "  fuzz/prebuilt/${{ env.CARGO_BUILD_TARGET }}/release/one\n"
-            )
-            self.assertEqual(len(check_fuzz_targets(root)), 1)
-
-    def test_profile_listing_enforces_selection(self) -> None:
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp)
-            (root / ".config").mkdir()
-            digest = hashlib.sha256(b"suite::two\n").hexdigest()
-            (root / ".config/e2e-smoke-selection.txt").write_text(f"sha256={digest}\n")
-            listing = root / "listing.json"
-            listing.write_text(
-                json.dumps(
-                    {
-                        "rust-suites": {
-                            "suite": {
-                                "testcases": {
-                                    "one": {"filter-match": {"status": "matches"}},
-                                    "two": {"filter-match": {"status": "mismatch"}},
-                                }
-                            }
-                        }
-                    }
-                )
-            )
-            self.assertEqual(len(check_profile_listing(root, "e2e-smoke", listing)), 1)
-
-    def test_profile_listing_binds_platform_digest(self) -> None:
-        with tempfile.TemporaryDirectory() as tmp:
-            root = Path(tmp)
-            (root / ".config").mkdir()
-            darwin_digest = hashlib.sha256(b"suite::darwin\n").hexdigest()
-            linux_digest = hashlib.sha256(b"suite::linux\n").hexdigest()
-            (root / ".config/e2e-full-selection.txt").write_text(
-                f"sha256-darwin={darwin_digest}\nsha256-linux={linux_digest}\n"
-            )
-            listing = root / "listing.json"
-            listing.write_text(
-                json.dumps(
-                    {
-                        "rust-suites": {
-                            "suite": {
-                                "testcases": {"darwin": {"filter-match": {"status": "matches"}}}
-                            }
-                        }
-                    }
-                )
-            )
-            with mock.patch.object(sys, "platform", "linux"):
-                self.assertEqual(len(check_profile_listing(root, "e2e-full", listing)), 1)
-
-def main() -> int:
-    if sys.argv[1:] == ["--self-test"]:
-        suite = unittest.defaultTestLoader.loadTestsFromTestCase(SelfTests)
-        return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1
-    if len(sys.argv) == 4 and sys.argv[1] == "--check-profile":
-        errors = check_profile_listing(ROOT, sys.argv[2], Path(sys.argv[3]))
-        if errors:
-            for error in errors:
-                print(f"ERROR: {error}", file=sys.stderr)
-            return 1
-        return 0
-    if sys.argv[1:]:
-        print(
-            "usage: check_test_wiring.py [--self-test | --check-profile PROFILE LISTING]",
-            file=sys.stderr,
-        )
-        return 2
-
-    errors = validate(ROOT)
-    if errors:
-        for error in errors:
-            print(f"ERROR: {error}", file=sys.stderr)
-        return 1
-    print("OK: e2e modules, runner selection, fuzz matrices, and profile guards are wired")
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())
@@ -20,8 +20,6 @@ DATA_DIR="$TARGET_DIR/rustfs_test_data"
 RUSTFS_PID=""
 TEST_FILTER=""
 TEST_TYPE="all"
-RUSTFS_BUILD_FEATURES="${RUSTFS_BUILD_FEATURES:-ftps,webdav,sftp}"
-export RUSTFS_BUILD_FEATURES

 # Function to print colored output
 print_info() {
@@ -94,7 +92,7 @@ build_rustfs() {
    print_info "Building RustFS..."
    cd "$PROJECT_ROOT"
    
-    if ! cargo build --bin rustfs --features "$RUSTFS_BUILD_FEATURES"; then
+    if ! cargo build --bin rustfs; then
        print_error "Failed to build RustFS"
        exit 1
    fi
@@ -221,28 +219,27 @@ start_rustfs() {
 run_tests() {
    print_info "Running e2e tests..."
    cd "$PROJECT_ROOT"
-
-    local test_cmd=(cargo test --package e2e_test --lib)
-
+    
+    local test_cmd="cargo test --package e2e_test --lib"
+    
    case "$TEST_TYPE" in
        "specific")
-            test_cmd+=(-- "$TEST_FILTER")
+            test_cmd="$test_cmd -- $TEST_FILTER --exact --show-output --ignored"
            print_info "Running specific test: $TEST_FILTER"
            ;;
        "file")
-            test_cmd+=(-- "$TEST_FILTER")
+            test_cmd="$test_cmd -- $TEST_FILTER --show-output --ignored"
            print_info "Running tests in file/module: $TEST_FILTER"
            ;;
        "all")
-            test_cmd+=(--)
+            test_cmd="$test_cmd -- --show-output --ignored"
            print_info "Running all e2e tests"
            ;;
    esac
-    test_cmd+=(--show-output --include-ignored --test-threads=1)
-
-    print_info "Test command: ${test_cmd[*]}"
-
-    if "${test_cmd[@]}"; then
+    
+    print_info "Test command: $test_cmd"
+    
+    if eval "$test_cmd"; then
        print_success "All tests passed!"
        return 0
    else
@@ -253,9 +253,6 @@ Test results are saved in the `artifacts/s3tests-${TEST_MODE}/` directory (defau

 - `junit.xml`: Test results in JUnit format (compatible with CI/CD systems)
 - `pytest.log`: Detailed pytest logs with full test output
- `all-collected-nodeids.txt`: Exact node IDs in the pinned upstream suite
- `selected-nodeids.txt`: Exact node IDs expected in this run
- `unsharded-selected-nodeids.txt`: Exact node IDs before deterministic sharding
 - `compat-report.md`: Classification report generated by `report_compat.py` —
  regressions against `implemented_tests.txt`, promotion candidates (tests
  that pass but are still listed as unimplemented/excluded), and tests missing
@@ -452,11 +449,9 @@ RustFS. Two GitHub Actions workflows delegate to it:
 - **Full sweep** (`.github/workflows/e2e-s3tests.yml`): weekly scheduled (and
  manually dispatchable) run of the ENTIRE upstream suite (`TEST_SCOPE=all`)
  against a Docker deployment — single node or a real 4-node distributed
-  cluster behind HAProxy. Regressions, unclassified tests, incomplete
-  execution, and infrastructure errors fail the sweep; classified unsupported
-  behavior remains informational. Scheduled topology runs are split into four
-  deterministic exact-node-ID shards, and every case has a five-minute timeout,
-  so one stalled case cannot erase the entire sweep's evidence.
+  cluster behind HAProxy. The sweep fails only on regressions in the
+  implemented whitelist; everything else is reported by `report_compat.py`
+  as promotion candidates or unclassified tests.

 Keeping both workflows on this script means local runs, the PR gate, and the
 scheduled sweep always execute tests the same way (same pinned s3-tests
@@ -471,9 +466,8 @@ pass/fail table in the job summary.
 ## Companion Tools

 - `report_compat.py` — diffs a junit.xml result against the classification
-  lists and the exact pytest collection; run before execution to reject stale
-  or missing classifications, then after execution to detect regressions and
-  incomplete parameterized cases.
+  lists; run automatically at the end of `run.sh`, and used by the weekly
+  sweep to gate on whitelist regressions only (`--fail-on-regression`).
 - `api_coverage.py` — quantifies S3 API surface coverage by comparing the
  s3s `S3` trait (at the revision pinned in Cargo.toml) against the methods
  RustFS overrides in `impl S3 for FS`:
@@ -307,10 +307,3 @@ test_object_acl_write
 test_object_acl_writeacp
 test_put_bucket_acl_grant_group_read
 test_object_raw_get_bucket_acl
-
-# Require upstream cloud-storage or IAM account services
-test_bucket_logging_requester_assumed_role
-test_lifecycle_cloud_transition_target_by_bucket
-test_lifecycle_cloud_transition_target_by_bucket_multiple_buckets
-test_list_object_versions_restore_status
-test_list_objects_restore_status
@@ -521,13 +521,9 @@ test_atomic_dual_conditional_write_1mb
 test_atomic_write_bucket_gone
 test_bucket_acl_canned_private_to_private
 test_bucket_concurrent_set_canned_acl
-test_bucket_create_delete
-test_bucket_policy
 test_bucket_policy_acl
 test_bucket_policy_put_obj_acl
 test_bucketv2_policy_acl
-test_copy_enc
-test_copy_part_enc
 test_copy_object_ifmatch_failed
 test_copy_object_ifnonematch_good
 test_cors_presigned_put_object_tenant_with_acl
@@ -21,17 +21,14 @@ Classifies every executed test into:
  - unclassified passes:    passed but not present in any list (new upstream tests)
  - unclassified failures:  failed and not present in any list (new upstream tests)

-Writes a markdown report and prints a summary to stdout. Optional gates reject
-regressions, unclassified tests, stale classifications, and incomplete node-ID
-execution.
+Writes a markdown report and prints a summary to stdout. Exit code is 0 unless
+--fail-on-regression is given and at least one regression was found.
 """

 from __future__ import annotations

 import argparse
-from collections import Counter
 import pathlib
-import re
 import sys
 import xml.etree.ElementTree as ET

@@ -48,47 +45,33 @@ LIST_FILES = {
 }


-def load_entries(path: pathlib.Path) -> list[str]:
-    names: list[str] = []
+def load_list(path: pathlib.Path) -> set[str]:
+    names: set[str] = set()
    if not path.is_file():
        return names
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if line and not line.startswith("#"):
-            names.append(line)
+            names.add(line)
    return names


-def classification_errors(entries: dict[str, list[str]]) -> list[str]:
-    errors: list[str] = []
-    lists = {key: set(names) for key, names in entries.items()}
-    for key, names in entries.items():
-        duplicates = sorted(name for name, count in Counter(names).items() if count > 1)
-        if duplicates:
-            errors.append(f"{LIST_FILES[key]} has duplicates: {', '.join(duplicates)}")
-    keys = tuple(lists)
-    for index, left in enumerate(keys):
-        for right in keys[index + 1 :]:
-            overlap = sorted(lists[left] & lists[right])
-            if overlap:
-                errors.append(f"{left}/{right} classifications overlap: {', '.join(overlap)}")
-    return errors
-
-
 def base_name(testcase_name: str) -> str:
    """Strip pytest parametrization (test_foo[param]) to match list entries."""
    return testcase_name.split("[", 1)[0]


-def parse_junit(path: pathlib.Path) -> tuple[dict[str, str], list[str], list[tuple[str, str, str, str]]]:
-    """Return exact statuses, pytest-timeout cases, and failure summaries."""
+def parse_junit(path: pathlib.Path) -> dict[str, str]:
+    """Return {test name: status} with status in passed/failed/error/skipped.
+
+    Parametrized cases collapse onto their base name; any failing variant marks
+    the whole test failed.
+    """
    results: dict[str, str] = {}
-    timed_out: list[str] = []
-    failures: list[tuple[str, str, str, str]] = []
    severity = {"skipped": 0, "passed": 1, "failed": 2, "error": 2}
    root = ET.parse(path).getroot()
    for case in root.iter("testcase"):
-        name = case.get("name", "")
+        name = base_name(case.get("name", ""))
        if not name:
            continue
        if case.find("failure") is not None:
@@ -102,35 +85,7 @@ def parse_junit(path: pathlib.Path) -> tuple[dict[str, str], list[str], list[tup
        prev = results.get(name)
        if prev is None or severity[status] > severity[prev]:
            results[name] = status
-        node = case.find("failure") if status == "failed" else case.find("error")
-        if node is not None:
-            details = " ".join(filter(None, [node.get("message", ""), node.text or ""]))
-            message = node.get("message") or next(iter((node.text or "").strip().splitlines()), "")
-            failures.append((case.get("classname", ""), name, case.get("time", "0"), message))
-            if re.search(r"\bTimeout\s*(?:>|\()", details, re.IGNORECASE):
-                timed_out.append(name)
-    return results, timed_out, failures
-
-
-def collapse_results(results: dict[str, str]) -> dict[str, str]:
-    """Collapse parametrized cases for classification-level reporting."""
-    collapsed: dict[str, str] = {}
-    severity = {"skipped": 0, "passed": 1, "failed": 2, "error": 2}
-    for exact_name, status in results.items():
-        name = base_name(exact_name)
-        previous = collapsed.get(name)
-        if previous is None or severity[status] > severity[previous]:
-            collapsed[name] = status
-    return collapsed
-
-
-def load_collected_nodeids(path: pathlib.Path) -> set[str]:
-    names: set[str] = set()
-    for line in path.read_text(encoding="utf-8").splitlines():
-        nodeid = line.strip()
-        if nodeid:
-            names.add(nodeid.rsplit("::", 1)[-1])
-    return names
+    return results


 def render_section(title: str, rows: list[str], hint: str = "") -> list[str]:
@@ -147,7 +102,7 @@ def render_section(title: str, rows: list[str], hint: str = "") -> list[str]:

 def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--junit", type=pathlib.Path, help="junit.xml produced by pytest")
+    parser.add_argument("--junit", required=True, type=pathlib.Path, help="junit.xml produced by pytest")
    parser.add_argument(
        "--lists-dir",
        type=pathlib.Path,
@@ -160,60 +115,14 @@ def main() -> int:
        action="store_true",
        help="exit non-zero when a test from implemented_tests.txt failed",
    )
-    parser.add_argument(
-        "--fail-on-unclassified",
-        action="store_true",
-        help="exit non-zero when an executed test is absent from every classification",
-    )
-    parser.add_argument(
-        "--collected-nodeids",
-        type=pathlib.Path,
-        help="exact pytest node IDs from the pinned suite's collect-only pass",
-    )
-    parser.add_argument(
-        "--check-classifications-only",
-        action="store_true",
-        help="validate classification names against collected node IDs without reading JUnit",
-    )
    args = parser.parse_args()

-    entries = {key: load_entries(args.lists_dir / fname) for key, fname in LIST_FILES.items()}
-    lists = {key: set(names) for key, names in entries.items()}
-    invalid_classifications = classification_errors(entries)
-    collected: set[str] = set()
-    if args.collected_nodeids:
-        collected = load_collected_nodeids(args.collected_nodeids)
-        collected_base = {base_name(name) for name in collected}
-        classified = set().union(*lists.values())
-        missing_classifications = sorted(collected_base - classified)
-        stale_classifications = sorted(classified - collected_base)
-    else:
-        missing_classifications = []
-        stale_classifications = []
-
-    if args.check_classifications_only:
-        if not args.collected_nodeids:
-            parser.error("--check-classifications-only requires --collected-nodeids")
-        for error in invalid_classifications:
-            print(f"[INVALID] {error}")
-        for name in missing_classifications:
-            print(f"[UNCLASSIFIED] {name}")
-        for name in stale_classifications:
-            print(f"[STALE] {name}")
-        return 1 if invalid_classifications or missing_classifications or stale_classifications else 0
-
-    if invalid_classifications:
-        for error in invalid_classifications:
-            print(f"[ERROR] {error}", file=sys.stderr)
-        return 2
-
-    if not args.junit or not args.junit.is_file():
+    if not args.junit.is_file():
        print(f"[ERROR] junit file not found: {args.junit}", file=sys.stderr)
        return 2

-    exact_results, timed_out, failures = parse_junit(args.junit)
-    results = collapse_results(exact_results)
-    missing_results = sorted(collected - exact_results.keys()) if collected else []
+    lists = {key: load_list(args.lists_dir / fname) for key, fname in LIST_FILES.items()}
+    results = parse_junit(args.junit)

    regressions: list[str] = []
    promotions: dict[str, list[str]] = {"unimplemented": [], "excluded": []}
@@ -246,9 +155,7 @@ def main() -> int:
    lines = [
        "# S3 compatibility report",
        "",
-        f"Executed: {len(exact_results)} exact cases across {len(results)} classified tests.",
-        "",
-        "Classification status — "
+        f"Executed: {len(results)} tests — "
        f"{counts['passed']} passed, {counts['failed']} failed, "
        f"{counts['error']} errored, {counts['skipped']} skipped.",
        "",
@@ -284,16 +191,6 @@ def main() -> int:
        unclassified_failed,
        "Failing and absent from every list — triage into `unimplemented_tests.txt` or `excluded_tests.txt`.",
    )
-    lines += render_section(
-        "Missing results",
-        missing_results,
-        "Present in the pinned upstream suite but absent from JUnit — the sweep was incomplete.",
-    )
-    lines += render_section(
-        "Timed out",
-        timed_out,
-        "Per-test timeout is an infrastructure failure regardless of compatibility classification.",
-    )

    report = "\n".join(lines)
    if args.output:
@@ -304,28 +201,13 @@ def main() -> int:
    print(
        f"[INFO] {len(regressions)} regression(s), "
        f"{len(promotions['unimplemented']) + len(promotions['excluded']) + len(unclassified_passed)} promotion candidate(s), "
-        f"{len(unclassified_failed)} unclassified failure(s), "
-        f"{len(missing_results)} missing result(s), "
-        f"{len(timed_out)} timeout(s)"
+        f"{len(unclassified_failed)} unclassified failure(s)"
    )
    for name in sorted(regressions):
        print(f"[REGRESSION] {name}")
-    if failures:
-        print("[ERROR] s3-tests failed testcase summary:")
-        for classname, name, duration, message in failures[:20]:
-            nodeid = f"{classname}::{name}" if classname else name
-            print(f"[ERROR] - {nodeid} ({duration}s): {message}")
-        if len(failures) > 20:
-            print(f"[ERROR] - ... {len(failures) - 20} additional failed testcases omitted")

    if args.fail_on_regression and regressions:
        return 1
-    if args.fail_on_unclassified and (unclassified_passed or unclassified_failed):
-        return 1
-    if args.collected_nodeids and missing_results:
-        return 1
-    if timed_out:
-        return 1
    return 0


@@ -58,19 +58,6 @@ if [[ "${TEST_SCOPE}" != "implemented" && "${TEST_SCOPE}" != "all" ]]; then
    echo "[ERROR] Invalid TEST_SCOPE: ${TEST_SCOPE} (must be \"implemented\" or \"all\")" >&2
    exit 1
 fi
-S3_SHARD_COUNT="${S3_SHARD_COUNT:-1}"
-S3_SHARD_INDEX="${S3_SHARD_INDEX:-0}"
-TEST_TIMEOUT="${TEST_TIMEOUT:-300}"
-if [[ ! "${S3_SHARD_COUNT}" =~ ^[1-9][0-9]*$ ]] \
-    || [[ ! "${S3_SHARD_INDEX}" =~ ^[0-9]+$ ]] \
-    || (( S3_SHARD_INDEX >= S3_SHARD_COUNT )); then
-    echo "[ERROR] Invalid S3 shard ${S3_SHARD_INDEX}/${S3_SHARD_COUNT}" >&2
-    exit 1
-fi
-if [[ ! "${TEST_TIMEOUT}" =~ ^[1-9][0-9]*$ ]]; then
-    echo "[ERROR] Invalid TEST_TIMEOUT: ${TEST_TIMEOUT}" >&2
-    exit 1
-fi

 # Upstream ceph/s3-tests suite, pinned for reproducible runs.
 # Bump S3TESTS_REV deliberately: upstream changes can rename tests or change
@@ -109,6 +96,55 @@ log_error() {
    echo -e "${RED}[ERROR]${NC} $*"
 }

+summarize_junit_failures() {
+    local junit_path="$1"
+
+    if [ ! -f "${junit_path}" ]; then
+        log_warn "JUnit report not found: ${junit_path}"
+        return 0
+    fi
+
+    python3 - "${junit_path}" <<'PY'
+import sys
+import xml.etree.ElementTree as ET
+
+junit_path = sys.argv[1]
+try:
+    root = ET.parse(junit_path).getroot()
+except Exception as exc:
+    print(f"[WARN] Failed to parse JUnit report {junit_path}: {exc}")
+    raise SystemExit(0)
+
+failures = []
+for case in root.iter("testcase"):
+    failure = case.find("failure")
+    error = case.find("error")
+    node = failure if failure is not None else error
+    if node is None:
+        continue
+
+    classname = case.attrib.get("classname", "")
+    name = case.attrib.get("name", "")
+    duration = case.attrib.get("time", "0")
+    message = node.attrib.get("message") or (node.text or "").strip().splitlines()[0:1]
+    if isinstance(message, list):
+        message = message[0] if message else ""
+    failures.append((classname, name, duration, message))
+
+if not failures:
+    print("[INFO] No failed testcases found in JUnit report")
+    raise SystemExit(0)
+
+print("[ERROR] s3-tests failed testcase summary:")
+for classname, name, duration, message in failures[:20]:
+    nodeid = f"{classname}::{name}" if classname else name
+    print(f"[ERROR] - {nodeid} ({duration}s): {message}")
+
+if len(failures) > 20:
+    print(f"[ERROR] - ... {len(failures) - 20} additional failed testcases omitted")
+PY
+}
+
 # =============================================================================
 # Test Classification Files
 # =============================================================================
@@ -286,9 +322,6 @@ Environment Variables:
  MAXFAIL                - Stop after N failures, 0 = never stop (default: 1)
  XDIST                  - Enable parallel execution with N workers (default: 0)
  TEST_SCOPE             - "implemented" (whitelist, default) or "all" (entire upstream suite)
-  S3_SHARD_COUNT         - Number of deterministic exact-node-ID shards (default: 1)
-  S3_SHARD_INDEX         - Zero-based shard index (default: 0)
-  TEST_TIMEOUT           - Per-test timeout in seconds (default: 300)
  S3TESTS_REPO           - s3-tests repository URL (default: https://github.com/ceph/s3-tests.git)
  S3TESTS_REV            - Pinned s3-tests commit; bump deliberately and reclassify test lists
  MARKEXPR               - pytest marker expression (default: no marker filtering)
@@ -949,10 +982,9 @@ mkdir -p "${ARTIFACTS_DIR}"
 XDIST_ARGS=""
 if [ "${XDIST}" != "0" ]; then
    # Add pytest-xdist to requirements.txt so tox installs it inside its virtualenv
-    grep -qxF "pytest-xdist" requirements.txt || echo "pytest-xdist" >> requirements.txt
+    echo "pytest-xdist" >> requirements.txt
    XDIST_ARGS="-n ${XDIST} --dist=loadgroup"
 fi
-grep -qxF "pytest-timeout" requirements.txt || echo "pytest-timeout" >> requirements.txt

 # Resolve config path (absolute path for tox)
 if [[ "${S3TESTS_CONF}" = /* ]]; then
@@ -971,69 +1003,12 @@ else
    PYTEST_SELECTION_ARGS=("${S3_TEST_FILE}")
 fi

-collect_nodeids() {
-    local output_path="$1"
-    shift
-    local collect_log="${output_path%.txt}.log"
-    local collect_rc=0
-    local node_prefix="${S3_TEST_FILE//./\\.}::"
-
-    set +e
-    S3TEST_CONF="${CONF_OUTPUT_PATH}" tox -- -q --collect-only "$@" 2>&1 | tee "${collect_log}"
-    collect_rc=${PIPESTATUS[0]}
-    set -e
-    if [ "${collect_rc}" -ne 0 ]; then
-        log_error "pytest collection failed with exit code ${collect_rc}"
-        return "${collect_rc}"
-    fi
-    grep -E "^${node_prefix}" "${collect_log}" > "${output_path}" || true
-    if [ ! -s "${output_path}" ]; then
-        log_error "pytest collection produced no S3 test node IDs"
-        return 1
-    fi
-}
-
-ALL_COLLECTED_NODEIDS="${ARTIFACTS_DIR}/all-collected-nodeids.txt"
-UNSHARDED_SELECTED_NODEIDS="${ARTIFACTS_DIR}/unsharded-selected-nodeids.txt"
-SELECTED_NODEIDS="${ARTIFACTS_DIR}/selected-nodeids.txt"
-collect_nodeids "${ALL_COLLECTED_NODEIDS}" "${S3_TEST_FILE}" -m "not rustfs_never_marker"
-python3 "${SCRIPT_DIR}/report_compat.py" \
-    --lists-dir "${SCRIPT_DIR}" \
-    --collected-nodeids "${ALL_COLLECTED_NODEIDS}" \
-    --check-classifications-only || {
-    log_error "S3 test classifications do not match pinned revision ${S3TESTS_REV}"
-    exit 1
-}
-if [[ "${TEST_SCOPE}" == "all" && -z "${TESTEXPR}" && "${MARKEXPR}" == "not rustfs_never_marker" ]]; then
-    cp "${ALL_COLLECTED_NODEIDS}" "${UNSHARDED_SELECTED_NODEIDS}"
-else
-    collect_nodeids "${UNSHARDED_SELECTED_NODEIDS}" "${PYTEST_SELECTION_ARGS[@]}" -m "${MARKEXPR}"
-fi
-
-if (( S3_SHARD_COUNT > 1 )); then
-    awk -v count="${S3_SHARD_COUNT}" -v shard_index="${S3_SHARD_INDEX}" \
-        '((NR - 1) % count) == shard_index' \
-        "${UNSHARDED_SELECTED_NODEIDS}" > "${SELECTED_NODEIDS}"
-    if [[ ! -s "${SELECTED_NODEIDS}" ]]; then
-        log_error "Shard ${S3_SHARD_INDEX}/${S3_SHARD_COUNT} selected no tests"
-        exit 1
-    fi
-    PYTEST_SELECTION_ARGS=()
-    while IFS= read -r nodeid; do
-        PYTEST_SELECTION_ARGS+=("${nodeid}")
-    done < "${SELECTED_NODEIDS}"
-    log_info "Selected shard ${S3_SHARD_INDEX}/${S3_SHARD_COUNT}: ${#PYTEST_SELECTION_ARGS[@]} exact cases"
-else
-    cp "${UNSHARDED_SELECTED_NODEIDS}" "${SELECTED_NODEIDS}"
-fi
-
 # Run tests from s3tests/functional
 set +e
 S3TEST_CONF="${CONF_OUTPUT_PATH}" \
    tox -- \
    -vv -ra --showlocals --tb=long \
    --maxfail="${MAXFAIL}" \
-    --timeout="${TEST_TIMEOUT}" \
    --junitxml="${ARTIFACTS_DIR}/junit.xml" \
    ${XDIST_ARGS} \
    "${PYTEST_SELECTION_ARGS[@]}" \
@@ -1058,22 +1033,19 @@ elif [ "${DEPLOY_MODE}" = "existing" ]; then
    echo "{\"host\": \"${S3_HOST}\", \"port\": ${S3_PORT}, \"mode\": \"existing\"}" > "${ARTIFACTS_DIR}/rustfs-${TEST_MODE}/inspect.json" || true
 fi

-# Step 11: Classification report and gate
+# Step 11: Classification report (informational, never fails the run)
 REPORT_SCRIPT="${SCRIPT_DIR}/report_compat.py"
-REPORT_ARGS=(
-    --junit "${ARTIFACTS_DIR}/junit.xml"
-    --lists-dir "${SCRIPT_DIR}"
-    --collected-nodeids "${SELECTED_NODEIDS}"
-    --output "${ARTIFACTS_DIR}/compat-report.md"
-    --fail-on-regression
-)
-if [[ "${TEST_SCOPE}" == "all" ]]; then
-    REPORT_ARGS+=(--fail-on-unclassified)
+if [ -f "${REPORT_SCRIPT}" ] && [ -f "${ARTIFACTS_DIR}/junit.xml" ]; then
+    python3 "${REPORT_SCRIPT}" \
+        --junit "${ARTIFACTS_DIR}/junit.xml" \
+        --lists-dir "${SCRIPT_DIR}" \
+        --output "${ARTIFACTS_DIR}/compat-report.md" \
+        || log_warn "Compatibility report generation failed"
+fi
+
+if [ ${TEST_EXIT_CODE} -ne 0 ]; then
+    summarize_junit_failures "${ARTIFACTS_DIR}/junit.xml"
 fi
-set +e
-python3 "${REPORT_SCRIPT}" "${REPORT_ARGS[@]}"
-REPORT_EXIT_CODE=$?
-set -e

 # Summary
 if [ ${TEST_EXIT_CODE} -eq 0 ]; then
@@ -1087,10 +1059,4 @@ else
    log_info "Check RustFS logs: ${ARTIFACTS_DIR}/rustfs-${TEST_MODE}/rustfs.log"
 fi

-if [[ "${TEST_EXIT_CODE}" -ne 0 && "${TEST_EXIT_CODE}" -ne 1 ]]; then
-    exit "${TEST_EXIT_CODE}"
-fi
-if [[ "${TEST_SCOPE}" == "implemented" && "${TEST_EXIT_CODE}" -ne 0 ]]; then
-    exit "${TEST_EXIT_CODE}"
-fi
-exit "${REPORT_EXIT_CODE}"
+exit ${TEST_EXIT_CODE}
@@ -1,119 +0,0 @@
-#!/usr/bin/env python3
-"""Regression tests for the S3 compatibility report."""
-
-from __future__ import annotations
-
-import importlib.util
-import subprocess
-import sys
-import tempfile
-import unittest
-from pathlib import Path
-
-
-REPORT_PATH = Path(__file__).with_name("report_compat.py")
-SPEC = importlib.util.spec_from_file_location("report_compat", REPORT_PATH)
-assert SPEC and SPEC.loader
-REPORT = importlib.util.module_from_spec(SPEC)
-SPEC.loader.exec_module(REPORT)
-
-
-class ReportCompatTests(unittest.TestCase):
-    def test_upstream_names_expose_incomplete_junit(self) -> None:
-        with tempfile.TemporaryDirectory() as tmp:
-            directory = Path(tmp)
-            collected = directory / "collected.txt"
-            collected.write_text("s3tests/functional/test_s3.py::test_one[a]\ns3tests/functional/test_s3.py::test_one[b]\n")
-            junit = directory / "junit.xml"
-            junit.write_text('<testsuite><testcase name="test_one[a]" /></testsuite>')
-
-            expected = REPORT.load_collected_nodeids(collected)
-            results, _, _ = REPORT.parse_junit(junit)
-
-            self.assertEqual(expected - results.keys(), {"test_one[b]"})
-
-    def test_cli_fails_an_incomplete_sweep(self) -> None:
-        with tempfile.TemporaryDirectory() as tmp:
-            directory = Path(tmp)
-            collected = directory / "collected.txt"
-            collected.write_text("s3tests/functional/test_s3.py::test_one\ns3tests/functional/test_s3.py::test_two\n")
-            junit = directory / "junit.xml"
-            junit.write_text('<testsuite><testcase name="test_one" /></testsuite>')
-            for filename in REPORT.LIST_FILES.values():
-                (directory / filename).write_text("")
-            (directory / "implemented_tests.txt").write_text("test_one\n")
-
-            result = subprocess.run(
-                [
-                    sys.executable,
-                    str(REPORT_PATH),
-                    "--junit",
-                    str(junit),
-                    "--lists-dir",
-                    str(directory),
-                    "--collected-nodeids",
-                    str(collected),
-                    "--fail-on-regression",
-                    "--fail-on-unclassified",
-                ],
-                check=False,
-                capture_output=True,
-                text=True,
-            )
-
-            self.assertEqual(result.returncode, 1)
-            self.assertIn("1 missing result(s)", result.stdout)
-
-    def test_preflight_rejects_missing_and_stale_classifications(self) -> None:
-        with tempfile.TemporaryDirectory() as tmp:
-            directory = Path(tmp)
-            collected = directory / "collected.txt"
-            collected.write_text("s3tests/functional/test_s3.py::test_known[a]\ntest_new\n")
-            for filename in REPORT.LIST_FILES.values():
-                (directory / filename).write_text("")
-            (directory / "implemented_tests.txt").write_text("test_known\ntest_stale\ntest_stale\n")
-
-            result = subprocess.run(
-                [
-                    sys.executable,
-                    str(REPORT_PATH),
-                    "--lists-dir",
-                    str(directory),
-                    "--collected-nodeids",
-                    str(collected),
-                    "--check-classifications-only",
-                ],
-                check=False,
-                capture_output=True,
-                text=True,
-            )
-
-            self.assertEqual(result.returncode, 1)
-            self.assertIn("[UNCLASSIFIED] test_new", result.stdout)
-            self.assertIn("[STALE] test_stale", result.stdout)
-            self.assertIn("[INVALID] implemented_tests.txt has duplicates: test_stale", result.stdout)
-
-    def test_timeout_fails_even_when_test_is_excluded(self) -> None:
-        with tempfile.TemporaryDirectory() as tmp:
-            directory = Path(tmp)
-            junit = directory / "junit.xml"
-            junit.write_text(
-                '<testsuite><testcase name="test_slow"><failure message="Failed: Timeout (&gt;300.0s)" /></testcase></testsuite>'
-            )
-            for filename in REPORT.LIST_FILES.values():
-                (directory / filename).write_text("")
-            (directory / "excluded_tests.txt").write_text("test_slow\n")
-
-            result = subprocess.run(
-                [sys.executable, str(REPORT_PATH), "--junit", str(junit), "--lists-dir", str(directory)],
-                check=False,
-                capture_output=True,
-                text=True,
-            )
-
-            self.assertEqual(result.returncode, 1)
-            self.assertIn("1 timeout(s)", result.stdout)
-
-
-if __name__ == "__main__":
-    unittest.main()
@@ -11,12 +11,8 @@

 # Failed tests
 test_bucket_create_delete_bucket_ownership
-test_bucket_logging_request_id
 test_create_bucket_no_ownership_controls
 test_bucket_logging_owner
-test_head_object_404_with_policy_prefix
-test_multipart_reupload_checksum_and_etag
-test_multipart_upload_complete_without_create
 test_object_copy_not_owned_bucket
 test_bucket_policy_multipart
 test_post_object_upload_checksum
				`@@ -1 +0,0 @@`
				`sha256=9b9bc336b43b70d0e06e0adb5455bf035bb18945d85d60936eb6fe4d48e0e680`
				`@@ -1 +0,0 @@`
				`sha256=655a3f3c1d042e694339d15caba7580518320322d1bac0f09450b37e6c09e2e7`
				`@@ -1 +0,0 @@`
				`sha256=ec27cde6ce6400723c4b372bfbd2ac61709c744294e4810af765e8a808d8e31d`