Merge branch 'main' into overtrue/report-security-coverage-ratchet

test(kms): stabilize Vault failover validation (#6385 )
* test(kms): bound Vault failover progress wait * ci(nightly): honor manual dispatch ref * test(kms): preserve Vault worker failures * test(kms): validate Vault circuit recovery
2026-08-23 04:39:04 +00:00 · 2026-08-23 12:32:30 +08:00 · 2026-08-23 12:32:11 +08:00 · 2026-08-23 12:29:52 +08:00 · 2026-08-23 12:09:06 +08:00 · 2026-08-23 12:07:25 +08:00
33 changed files with 736 additions and 845 deletions
@@ -0,0 +1,20 @@
+# Report-only calibration baseline from https://github.com/rustfs/rustfs/actions/runs/29394996173.
+# Update counts only with a linked coverage run and a reviewed explanation.
+phase = "report-only"
+allowed_drop_percentage_points = 1.0
+
+[crates."crates/iam"]
+covered = 5149
+count = 8131
+
+[crates."crates/kms"]
+covered = 2950
+count = 4200
+
+[crates."crates/policy"]
+covered = 4636
+count = 5464
+
+[crates."crates/crypto"]
+covered = 469
+count = 494
@@ -36,6 +36,7 @@ script-tests: ## Run shell script tests
 	./scripts/test_manual_transition_runbooks.sh
 	./scripts/check_embedded_secrets.sh --self-test
 	python3 ./scripts/check_test_wiring.py --self-test
+	python3 ./scripts/check_security_coverage.py --self-test
 	python3 ./scripts/check_scheduled_validation_freshness.py --self-test
 	python3 ./scripts/s3-tests/test_report_compat.py
 	bash -n ./scripts/validate_object_data_cache_cold_stampede.sh
@@ -12,14 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

-# Weekly workspace line-coverage baseline (backlog#1153 infra-5).
+# Workspace line-coverage baseline and security-crate calibration
+# (backlog#1153 infra-5/infra-6).
 #
-# NON-BLOCKING by design: this workflow only runs on schedule and manual
-# dispatch, so it never attaches a status to a PR and must never be made a
-# required check. It exists to give coverage a visible baseline and trend
-# (per-crate table in the job summary, lcov artifact kept 90 days) — the
-# per-crate ratchet for the security-critical crates builds on it later
-# (backlog#1153 infra-6, report-only first per the ci-11 ladder).
+# NON-BLOCKING by design: the weekly job gives coverage a visible baseline and
+# trend, while relevant pull requests run a report-only security-crate
+# comparison. Neither job is a required check during calibration.
 #
 # Measurement scope matches the PR test gate (ci.yml "Run tests"):
 # `--workspace --exclude e2e_test` with the `ci` nextest profile. Doctests are
@@ -31,6 +29,17 @@
 name: coverage

 on:
+  pull_request:
+    branches: [main]
+    paths:
+      - "crates/iam/**"
+      - "crates/kms/**"
+      - "crates/policy/**"
+      - "crates/crypto/**"
+      - ".config/coverage-baselines.toml"
+      - "scripts/coverage_per_crate.py"
+      - "scripts/check_security_coverage.py"
+      - ".github/workflows/coverage.yml"
  workflow_dispatch:
  schedule:
    # 07:00 UTC Sunday — staggered clear of the other Sunday crons: ci (00:00),
@@ -39,6 +48,10 @@ on:
    # e2e-replication-nightly (04:00) and performance-ab (06:00) lanes.
    - cron: "43 7 * * 0"

+concurrency:
+  group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: ${{ github.event_name != 'schedule' }}
+
 # Only alert-on-failure needs more than read access; it declares its own
 # job-level `issues: write`.
 permissions:
@@ -46,12 +59,13 @@ permissions:

 jobs:
  coverage:
-    name: Workspace coverage (weekly)
+    name: Workspace line coverage
    runs-on: sm-standard-4
    # The instrumented build cannot reuse the regular CI cache (different
-    # RUSTFLAGS), so a cold week rebuilds the workspace before running the
-    # full suite; give it double the test job's 60-minute budget.
-    timeout-minutes: 120
+    # RUSTFLAGS), so a cold run rebuilds the workspace before running the
+    # full suite. Exact-head run 32573798257 needed 119m42s including reports
+    # and artifact upload, so keep a bounded 30-minute publication margin.
+    timeout-minutes: 150
    env:
      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
      # Match the PR gate's nextest semantics (ci.yml runs `--profile ci`):
@@ -91,7 +105,9 @@ jobs:
          cargo llvm-cov report --json --output-path target/llvm-cov/coverage.json

      - name: Write per-crate summary
-        run: python3 scripts/coverage_per_crate.py target/llvm-cov/coverage.json >> "$GITHUB_STEP_SUMMARY"
+        run: |
+          python3 scripts/coverage_per_crate.py target/llvm-cov/coverage.json >> "$GITHUB_STEP_SUMMARY"
+          python3 scripts/check_security_coverage.py target/llvm-cov/coverage.json >> "$GITHUB_STEP_SUMMARY"

      - name: Upload coverage artifact
        if: always()
@@ -39,11 +39,10 @@ jobs:
    env:
      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
    steps:
-      - name: Checkout main branch
+      - name: Checkout repository
        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
        with:
          persist-credentials: false
-          ref: main

      - name: Setup Rust environment
        uses: ./.github/actions/setup
@@ -89,11 +88,10 @@ jobs:
      # either casing.
      NO_PROXY: 127.0.0.1,localhost
    steps:
-      - name: Checkout main branch
+      - name: Checkout repository
        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
        with:
          persist-credentials: false
-          ref: main

      - name: Setup Rust environment
        uses: ./.github/actions/setup
@@ -178,11 +176,10 @@ jobs:
      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
      NO_PROXY: 127.0.0.1,localhost
    steps:
-      - name: Checkout main branch
+      - name: Checkout repository
        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
        with:
          persist-credentials: false
-          ref: main

      - name: Setup Rust environment
        uses: ./.github/actions/setup
@@ -30,7 +30,8 @@ make build-docker BUILD_OS=ubuntu22.04
 - Crate membership: `Cargo.toml` `[workspace].members`
 - Architecture, layering, crate map: [ARCHITECTURE.md](ARCHITECTURE.md)
 - Migration guardrails & readiness contracts: [docs/architecture/](docs/architecture/README.md)
- CI gates: `.github/workflows/ci.yml` (source of truth; never copy its steps into docs)
+- CI workflow steps: `.github/workflows/`; event, timeout, and required-status
+  matrix: [docs/testing/ci-gates.md](docs/testing/ci-gates.md)
 - Test-layer taxonomy, per-layer entry commands, serial/nextest rules, flake
  policy: [docs/testing/README.md](docs/testing/README.md)
 - Tier/ILM transition debugging (xl.meta inspection, versionId tracing):
@@ -70,6 +70,8 @@ make pre-pr

 > For the full test-layer taxonomy (unit / ecstore black-box / e2e / s3s-e2e / S3 compatibility / chaos / fuzz / bench), each layer's entry command, the naming conventions the migration gate depends on, and the serial/nextest rules, see [docs/testing/README.md](docs/testing/README.md).

+> For the event, timeout, required-status, and local reproduction matrix, see [docs/testing/ci-gates.md](docs/testing/ci-gates.md).
+
 ### 🔒 Automated Pre-commit Hooks
 #### What `make pre-commit` and `make pre-pr` actually run

@@ -1858,9 +1858,9 @@ dependencies = [

 [[package]]
 name = "cc"
-version = "1.4.3"
+version = "1.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d"
+checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273"
 dependencies = [
 "find-msvc-tools",
 "jobserver",
@@ -2522,12 +2522,6 @@ dependencies = [
 "subtle",
 ]

-[[package]]
-name = "cty"
-version = "0.2.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b365fabc795046672053e29c954733ec3b05e4be654ab130fe8f1f94d7051f35"
-
 [[package]]
 name = "curve25519-dalek"
 version = "4.1.3"
@@ -5988,15 +5982,6 @@ version = "0.2.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"

-[[package]]
-name = "libmimalloc-sys"
-version = "0.1.49"
-source = "git+https://github.com/xonatius/mimalloc_rust.git?rev=6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11#6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11"
-dependencies = [
- "cc",
- "cty",
-]
-
 [[package]]
 name = "libredox"
 version = "0.1.20"
@@ -6397,14 +6382,6 @@ dependencies = [
 "synstructure 0.13.2",
 ]

-[[package]]
-name = "mimalloc"
-version = "0.1.52"
-source = "git+https://github.com/xonatius/mimalloc_rust.git?rev=6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11#6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11"
-dependencies = [
- "libmimalloc-sys",
-]
-
 [[package]]
 name = "mime"
 version = "0.3.17"
@@ -9162,13 +9139,11 @@ dependencies = [
 "insta",
 "jiff",
 "libc",
- "libmimalloc-sys",
 "libsystemd",
 "matchit 0.9.2",
 "md-5 0.11.0",
 "metrics",
 "metrics-util",
- "mimalloc",
 "mime_guess",
 "opentelemetry",
 "opentelemetry_sdk",
@@ -9204,6 +9179,8 @@ dependencies = [
 "rustfs-lock",
 "rustfs-log-analyzer",
 "rustfs-madmin",
+ "rustfs-mimalloc",
+ "rustfs-mimalloc-sys",
 "rustfs-notify",
 "rustfs-object-capacity",
 "rustfs-object-data-cache",
@@ -9875,6 +9852,24 @@ dependencies = [
 "tokio",
 ]

+[[package]]
+name = "rustfs-mimalloc"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a406f4aa07084301d485beec873af6dccc8e3f8762da244743df92038b1db1a6"
+dependencies = [
+ "rustfs-mimalloc-sys",
+]
+
+[[package]]
+name = "rustfs-mimalloc-sys"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c3051b819175f58445d4c369a72f0ab88149f3885ba8bea2aff3be01f53fe7cd"
+dependencies = [
+ "cc",
+]
+
 [[package]]
 name = "rustfs-notify"
 version = "1.0.0-rc.3"
@@ -350,8 +350,8 @@ russh-sftp = "2.4.0"
 dav-server = "0.11.0"

 # Performance Analysis and Memory Profiling
-mimalloc = { version = "0.1.52", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11" }
-libmimalloc-sys = { version = "0.1.49", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11", features = ["extended"] }
+rustfs-mimalloc = { version = "0.5.0" }
+rustfs-mimalloc-sys = { version = "0.5.0" }
 hotpath = { version = "0.23.3", default-features = false }
 # Snapshot testing for output format regression detection
 insta = { version = "1.48" }
@@ -236,15 +236,6 @@ pub struct DataUsageInfo {
    /// without relying on synchronized clocks.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub usage_snapshot_authoritative_baseline: Option<DataUsageSnapshotIdentity>,
-    /// Per-set freshness for an observational aggregate.  A set entry is
-    /// never sufficient to make the aggregate authoritative; it only records
-    /// which last-known-good generation contributed to the view.
-    #[serde(default, skip_serializing_if = "Vec::is_empty")]
-    pub usage_snapshot_set_states: Vec<DataUsageSnapshotSetState>,
-    /// An observational view may contain only the sets that completed this
-    /// cycle (or retained a compatible last-known-good cache).
-    #[serde(default)]
-    pub usage_snapshot_partial: bool,
    /// Deprecated kept here for backward compatibility reasons
    pub bucket_sizes: HashMap<String, u64>,
    /// Per-disk snapshot information when available
@@ -261,22 +252,6 @@ pub struct DataUsageSnapshotIdentity {
    pub scanner_epoch: Option<u64>,
 }

-#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
-pub struct DataUsageSnapshotSetState {
-    pub pool_index: u64,
-    pub set_index: u64,
-    #[serde(default)]
-    pub scanner_cycle: Option<u64>,
-    #[serde(default)]
-    pub scanner_epoch: Option<u64>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub scan_plan_digest: Option<[u8; 32]>,
-    #[serde(default)]
-    pub complete: bool,
-    #[serde(default)]
-    pub tombstone: bool,
-}
-
 impl DataUsageInfo {
    pub fn snapshot_identity(&self) -> DataUsageSnapshotIdentity {
        DataUsageSnapshotIdentity {
@@ -316,7 +291,7 @@ pub fn data_usage_snapshot_is_newer(candidate: &DataUsageInfo, baseline: &DataUs
 /// rollback delete/recreate fences the previous bucket incarnation too.
 pub fn observed_data_usage_is_newer(observed: &DataUsageInfo, authoritative: &DataUsageInfo) -> bool {
    observed.usage_snapshot_converged == Some(false)
-        && (observed.is_complete_bucket_usage_snapshot() || observed.is_valid_partial_snapshot())
+        && observed.is_complete_bucket_usage_snapshot()
        && observed.usage_snapshot_authoritative_baseline.as_ref() == Some(&authoritative.snapshot_identity())
        && data_usage_snapshot_is_newer(observed, authoritative)
 }
@@ -1461,39 +1436,6 @@ impl DataUsageInfo {
            && u64::try_from(self.buckets_usage.len()).ok() == Some(self.buckets_count)
    }

-    /// Validate provenance before an observational view can be selected for
-    /// admin display. Partial data is accepted only with unique set states,
-    /// a plan digest for every state, and at least one usable generation.
-    pub fn is_valid_partial_snapshot(&self) -> bool {
-        if !self.usage_snapshot_partial
-            || self.usage_snapshot_converged != Some(false)
-            || self.last_update.is_none()
-            || self.scanner_cycle.is_none()
-            || self.scanner_epoch.is_none()
-            || self.usage_snapshot_set_states.is_empty()
-            || u64::try_from(self.buckets_usage.len()).ok() != Some(self.buckets_count)
-        {
-            return false;
-        }
-
-        let mut previous = None;
-        let mut plan_digest = None;
-        let mut has_source = false;
-        for state in &self.usage_snapshot_set_states {
-            if state.scan_plan_digest.is_none()
-                || plan_digest.is_some_and(|digest| Some(digest) != state.scan_plan_digest)
-                || state.scanner_cycle.is_some() != state.scanner_epoch.is_some()
-                || previous.is_some_and(|(pool, set)| (pool, set) >= (state.pool_index, state.set_index))
-            {
-                return false;
-            }
-            previous = Some((state.pool_index, state.set_index));
-            plan_digest = state.scan_plan_digest;
-            has_source |= state.scanner_cycle.is_some() && !state.tombstone;
-        }
-        has_source
-    }
-
    /// Add object metadata to data usage statistics
    pub fn add_object(&mut self, object_path: &str, meta_object: &rustfs_filemeta::MetaObject) {
        // This method is kept for backward compatibility
@@ -2321,55 +2263,6 @@ mod tests {
        assert!(!observed_data_usage_is_newer(&candidate(2, 9, Some(false), true), &authoritative));
        assert!(!observed_data_usage_is_newer(&candidate(2, 11, Some(true), true), &authoritative));
        assert!(!observed_data_usage_is_newer(&candidate(2, 11, Some(false), false), &authoritative));
-
-        let mut partial = candidate(2, 11, Some(false), false);
-        partial.usage_snapshot_partial = true;
-        partial.usage_snapshot_set_states = vec![DataUsageSnapshotSetState {
-            pool_index: 0,
-            set_index: 0,
-            scanner_cycle: Some(10),
-            scanner_epoch: Some(2),
-            scan_plan_digest: Some([1; 32]),
-            complete: false,
-            tombstone: false,
-        }];
-        assert!(observed_data_usage_is_newer(&partial, &authoritative));
-    }
-
-    #[test]
-    fn mixed_topology_snapshot_is_rejected() {
-        let mut partial = DataUsageInfo {
-            last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(2)),
-            scanner_cycle: Some(11),
-            scanner_epoch: Some(2),
-            buckets_count: 0,
-            usage_snapshot_converged: Some(false),
-            usage_snapshot_partial: true,
-            usage_snapshot_set_states: vec![
-                DataUsageSnapshotSetState {
-                    pool_index: 0,
-                    set_index: 0,
-                    scanner_cycle: Some(11),
-                    scanner_epoch: Some(2),
-                    scan_plan_digest: Some([1; 32]),
-                    complete: true,
-                    tombstone: false,
-                },
-                DataUsageSnapshotSetState {
-                    pool_index: 1,
-                    set_index: 0,
-                    scanner_cycle: Some(10),
-                    scanner_epoch: Some(2),
-                    scan_plan_digest: Some([2; 32]),
-                    complete: false,
-                    tombstone: false,
-                },
-            ],
-            ..Default::default()
-        };
-        assert!(!partial.is_valid_partial_snapshot());
-        partial.usage_snapshot_set_states[1].scan_plan_digest = Some([1; 32]);
-        assert!(partial.is_valid_partial_snapshot());
    }

    #[test]
@@ -73,16 +73,6 @@ struct CachedBucketUsage {
    // mutation. A strictly later generation is required before the mutation
    // evidence can be discarded.
    pending_scanner_position: Option<(u64, u64)>,
-    // Deletes are visible to admin immediately, but quota admission keeps
-    // them pending until a complete scanner generation reconciles the set.
-    // This marker intentionally remains process-local: the delete request
-    // updates this overlay before the scanner writes a durable snapshot. If
-    // the process restarts first, loading the persisted complete snapshot
-    // restores the pre-reconciliation (larger) baseline, which is
-    // conservative for quota admission. A persisted post-delete snapshot is
-    // necessarily a complete scanner reconciliation and therefore creates a
-    // fresh cache entry with no pending hold.
-    pending_negative_delta: u64,
 }

 type UsageMemoryCache = Arc<RwLock<HashMap<String, CachedBucketUsage>>>;
@@ -958,12 +948,7 @@ async fn load_observed_data_usage_snapshot(store: Arc<ECStore>) -> Option<DataUs
    };

    match parse_usage_snapshot(&data) {
-        Ok(info)
-            if info.usage_snapshot_converged == Some(false)
-                && (info.is_complete_bucket_usage_snapshot() || info.is_valid_partial_snapshot()) =>
-        {
-            Some(info)
-        }
+        Ok(info) if info.usage_snapshot_converged == Some(false) && info.is_complete_bucket_usage_snapshot() => Some(info),
        Ok(_) => {
            error!(
                event = "data_usage_snapshot_load_failed",
@@ -1008,7 +993,7 @@ async fn load_admin_data_usage_from_backend(store: Arc<ECStore>) -> Result<DataU
 }

 fn discard_incomplete_bucket_usage(data_usage_info: &mut DataUsageInfo) {
-    if !data_usage_info.is_complete_bucket_usage_snapshot() && !data_usage_info.usage_snapshot_partial {
+    if !data_usage_info.is_complete_bucket_usage_snapshot() {
        data_usage_info.usage_snapshot_complete = false;
        data_usage_info.buckets_usage.clear();
        data_usage_info.bucket_sizes.clear();
@@ -1658,7 +1643,6 @@ fn cached_bucket_usage_from_backend(usage: BucketUsageInfo, updated_at: SystemTi
        dirty: false,
        stale_snapshot_pending: false,
        pending_scanner_position: None,
-        pending_negative_delta: 0,
    }
 }

@@ -1672,7 +1656,6 @@ fn cached_bucket_usage_now(usage: BucketUsageInfo) -> CachedBucketUsage {
        dirty: false,
        stale_snapshot_pending: false,
        pending_scanner_position: None,
-        pending_negative_delta: 0,
    }
 }

@@ -1825,7 +1808,6 @@ pub async fn record_bucket_object_delete_memory(bucket: &str, deleted_size: u64,
        .or_insert_with(|| cached_bucket_usage_now(BucketUsageInfo::default()));

    entry.usage.size = entry.usage.size.saturating_sub(deleted_size);
-    entry.pending_negative_delta = entry.pending_negative_delta.saturating_add(deleted_size);
    if removed_current_object {
        entry.usage.objects_count = entry.usage.objects_count.saturating_sub(1);
        entry.usage.versions_count = entry.usage.versions_count.saturating_sub(1);
@@ -1881,7 +1863,7 @@ pub async fn get_bucket_usage_memory(bucket: &str) -> Option<u64> {
    cache
        .get(bucket)
        .filter(|cached| cached.authoritative)
-        .map(|cached| cached.usage.size.saturating_add(cached.pending_negative_delta))
+        .map(|cached| cached.usage.size)
 }

 async fn update_usage_cache_if_needed() {
@@ -2961,45 +2943,6 @@ mod tests {
        assert_eq!(selected.usage_snapshot_converged, Some(true));
    }

-    #[test]
-    fn persisted_authoritative_stalls_but_memory_overlay_remains_visible() {
-        let authoritative = DataUsageInfo {
-            last_update: Some(SystemTime::UNIX_EPOCH),
-            scanner_epoch: Some(4),
-            scanner_cycle: Some(10),
-            usage_snapshot_complete: true,
-            ..Default::default()
-        };
-        let mut partial = authoritative.clone();
-        partial.last_update = Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1));
-        partial.scanner_cycle = Some(11);
-        partial.usage_snapshot_complete = false;
-        partial.usage_snapshot_partial = true;
-        partial.usage_snapshot_converged = Some(false);
-        partial.usage_snapshot_authoritative_baseline = Some(authoritative.snapshot_identity());
-        partial.usage_snapshot_set_states = vec![rustfs_data_usage::DataUsageSnapshotSetState {
-            pool_index: 0,
-            set_index: 0,
-            scanner_cycle: Some(10),
-            scanner_epoch: Some(4),
-            scan_plan_digest: Some([1; 32]),
-            complete: false,
-            tombstone: false,
-        }];
-        partial.buckets_usage.insert(
-            "bucket".to_string(),
-            BucketUsageInfo {
-                size: 100,
-                ..Default::default()
-            },
-        );
-        partial.buckets_count = 1;
-
-        let (selected, _) = select_admin_data_usage_snapshot(authoritative, true, Some(partial));
-        assert!(selected.usage_snapshot_partial);
-        assert_eq!(selected.buckets_usage.get("bucket").map(|usage| usage.size), Some(100));
-    }
-
    #[tokio::test]
    async fn authoritative_save_cleanup_removes_observed_snapshot_best_effort() {
        let store = UsageCasStore::default();
@@ -4722,55 +4665,6 @@ mod tests {
        );
    }

-    #[tokio::test]
-    #[serial]
-    async fn partial_usage_is_observational_not_authoritative_for_quota() {
-        clear_usage_memory_cache_for_test().await;
-
-        let mut partial = data_usage_info_for_test("bucket-a", 10, 100, SystemTime::now());
-        partial.usage_snapshot_complete = false;
-        partial.usage_snapshot_partial = true;
-        replace_bucket_usage_memory_from_info(&partial).await;
-
-        assert_eq!(get_bucket_usage_memory("bucket-a").await, None);
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn stale_quota_uses_complete_baseline_plus_positive_deltas() {
-        clear_usage_memory_cache_for_test().await;
-
-        let baseline = data_usage_info_for_test("bucket-a", 1, 100, SystemTime::now());
-        replace_bucket_usage_memory_from_info(&baseline).await;
-        record_bucket_object_write_memory("bucket-a", None, 25).await;
-
-        assert_eq!(get_bucket_usage_memory("bucket-a").await, Some(125));
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn negative_delta_waits_for_set_reconciliation() {
-        clear_usage_memory_cache_for_test().await;
-
-        let baseline = data_usage_info_for_test("bucket-a", 1, 100, SystemTime::UNIX_EPOCH + Duration::from_secs(100));
-        replace_bucket_usage_memory_from_info(&baseline).await;
-        record_bucket_object_delete_memory("bucket-a", 25, true).await;
-
-        assert_eq!(get_bucket_usage_memory("bucket-a").await, Some(100));
-
-        // Simulate a process restart: the request-path overlay is gone, but
-        // the persisted authoritative snapshot is still the pre-reconciliation
-        // baseline. Quota must remain conservative until a complete scanner
-        // result proves the delete.
-        clear_usage_memory_cache_for_test().await;
-        replace_bucket_usage_memory_from_info(&baseline).await;
-        assert_eq!(get_bucket_usage_memory("bucket-a").await, Some(100));
-
-        let reconciled = data_usage_info_for_test("bucket-a", 0, 75, SystemTime::UNIX_EPOCH + Duration::from_secs(101));
-        replace_bucket_usage_memory_from_info(&reconciled).await;
-        assert_eq!(get_bucket_usage_memory("bucket-a").await, Some(75));
-    }
-
    #[tokio::test]
    #[serial]
    async fn memory_overlay_counts_versioned_overwrite_as_new_version() {
@@ -784,6 +784,24 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
 ///
 /// # Returns
 /// A Result containing the BitrotWriterWrapper or an error
+/// Size hint handed to `DiskAPI::create_file` for a bitrot-wrapped shard.
+///
+/// A known length is grown by one checksum per shard so the on-disk file size
+/// matches what the bitrot writer emits. A negative length is the
+/// unknown-size sentinel (`HashReader::SIZE_PRESERVE_LAYER`, used by SSE and
+/// compression) and must be preserved: `RemoteDisk::create_file` forwards it
+/// in the `put_file_stream` query, and the receiver only treats `size > 0` as
+/// a fixed body length when locating the authenticated trailer. Clamping it
+/// to `0` would claim an empty body and misframe the stream. `0` stays `0`
+/// because a genuinely empty object still means an empty body.
+fn bitrot_create_file_size(length: i64, shard_size: usize, checksum_algo: &HashAlgorithm) -> i64 {
+    if length <= 0 {
+        return length;
+    }
+    let length = length as usize;
+    (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
+}
+
 pub async fn create_bitrot_writer(
    is_inline_buffer: bool,
    disk: Option<&DiskStore>,
@@ -796,12 +814,7 @@ pub async fn create_bitrot_writer(
    let writer = if is_inline_buffer {
        CustomWriter::new_inline_buffer()
    } else if let Some(disk) = disk {
-        let length = if length > 0 {
-            let length = length as usize;
-            (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
-        } else {
-            0
-        };
+        let length = bitrot_create_file_size(length, shard_size, &checksum_algo);

        let file = disk.create_file("", volume, path, length).await?;
        #[cfg(feature = "hotpath")]
@@ -820,6 +833,25 @@ mod tests {
    use rustfs_rio::ChunkReader;
    use std::collections::VecDeque;

+    #[test]
+    fn bitrot_create_file_size_grows_known_length_by_checksums() {
+        // 10 bytes over 4-byte shards = 3 shards, each followed by a 32-byte hash.
+        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::HighwayHash256), 10 + 3 * 32);
+        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::None), 10);
+    }
+
+    #[test]
+    fn bitrot_create_file_size_keeps_empty_and_unknown_distinct() {
+        assert_eq!(bitrot_create_file_size(0, 4, &HashAlgorithm::HighwayHash256), 0);
+        // SSE/compression streams advertise SIZE_PRESERVE_LAYER (-1); the remote
+        // put_file_stream receiver relies on a non-positive size to parse the auth
+        // trailer from the stream tail, so the sentinel must survive untouched.
+        assert_eq!(
+            bitrot_create_file_size(rustfs_rio::HashReader::SIZE_PRESERVE_LAYER, 4, &HashAlgorithm::HighwayHash256),
+            rustfs_rio::HashReader::SIZE_PRESERVE_LAYER
+        );
+    }
+
    struct TestChunkReader {
        chunks: VecDeque<Bytes>,
    }
@@ -2124,26 +2124,13 @@ impl SetDisks {

            let put_object_size = known_put_object_storage_size(data.size());
            let shard_file_size_raw = erasure.shard_file_size(put_object_size);
-            let is_inline_buffer =
-                storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned);
+            let is_inline_buffer = storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned);

            let collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled();
            let shard_file_size = shard_file_size_raw;
            let shard_size = erasure.shard_size();
            let write_path = classify_put_write_path(is_inline_buffer, put_object_size, fi.erasure.block_size);
            let direct_inline_commit = matches!(write_path, SmallWritePath::Inline);
-            {
-                use std::io::Write;
-                let msg = format!(
-                    "INLINE_DEBUG: bucket={} obj={} size={} shard_fs={} ds={} bs={} inline={} direct={} path={} iblock={} ver={}\n",
-                    bucket, object, put_object_size, shard_file_size_raw, erasure.data_shards, fi.erasure.block_size,
-                    is_inline_buffer, direct_inline_commit, write_path.metric_label(), storage_class_config.inline_block(), opts.versioned
-                );
-                if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open("/tmp/rustfs_inline_debug.log") {
-                    let _ = f.write_all(msg.as_bytes());
-                }
-                let _ = std::io::stderr().write_all(msg.as_bytes());
-            }
            rustfs_io_metrics::record_put_object_path(write_path.metric_label());
            let writer_setup_stage_start = collect_stage_timing.then(Instant::now);
            let (mut writers, errors) = if direct_inline_commit {
@@ -3194,7 +3194,7 @@ impl ECStore {

        // Default return value
        let mut del_objects = vec![DeletedObject::default(); objects.len()];
-        let mut accounting = vec![None; objects.len()];
+        let accounting = vec![None; objects.len()];

        let mut del_errs = Vec::with_capacity(objects.len());
        for _ in 0..objects.len() {
@@ -271,7 +271,7 @@ pub(super) fn resolve_latest_object_info_candidates(
            .filter(|candidate| latest_candidate_mod_time(candidate) == Some(latest_mod_time))
            .collect::<Vec<_>>();

-        latest_candidates.sort_by(|left, right| right.idx.cmp(&left.idx));
+        latest_candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.idx));

        let Some(winner) = latest_candidates.first() else {
            return Err(Error::ErasureReadQuorum);
@@ -16,14 +16,14 @@
 //!
 //! `scripts/test/vault_ha_kms_live.sh` owns the official Vault containers and
 //! kills the active node while this test continuously decrypts through a
-//! surviving standby. KV2 and Transit requests must remain successful, use a
-//! bounded number of attempts, and leave the circuit and in-flight gauges at
-//! zero after a new leader is elected.
+//! surviving standby. KV2 and Transit must recover after the bounded circuit
+//! interval, use a bounded number of attempts, and leave the circuit and
+//! in-flight gauges at zero after a new leader is elected.

 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
-use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
 use std::time::Duration;

 use metrics_util::MetricKind;
@@ -43,6 +43,11 @@ const OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
 const IN_FLIGHT: &str = "rustfs_kms_backend_in_flight";
 const CIRCUIT_OPEN: &str = "rustfs_kms_backend_circuit_open";
 const MAX_ATTEMPTS: u32 = 10;
+const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(2);
+const HEALTHY_PROGRESS_TIMEOUT: Duration = Duration::from_secs(20);
+// The circuit remains open for 30s after five failed attempts.
+const POST_FAILOVER_PROGRESS_TIMEOUT: Duration = Duration::from_secs(35);
+const FAILOVER_ERROR_POLL_INTERVAL: Duration = Duration::from_millis(100);

 type MetricEntry = (
    metrics_util::CompositeKey,
@@ -64,7 +69,7 @@ fn config(backend: KmsBackend, backend_config: BackendConfig) -> KmsConfig {
        backend,
        backend_config,
        allow_insecure_dev_defaults: true,
-        timeout: Duration::from_secs(2),
+        timeout: ATTEMPT_TIMEOUT,
        retry_attempts: MAX_ATTEMPTS,
        enable_cache: false,
        ..KmsConfig::default()
@@ -164,14 +169,31 @@ fn retryable_failures(snapshot: &[MetricEntry], operation: &str) -> u64 {
        .sum()
 }

-async fn wait_for_count(counter: &AtomicU64, minimum: u64, description: &str) {
-    tokio::time::timeout(Duration::from_secs(20), async {
+async fn wait_for_count(
+    counter: &AtomicU64,
+    failure: &Mutex<Option<String>>,
+    minimum: u64,
+    description: &str,
+    timeout: Duration,
+) {
+    tokio::time::timeout(timeout, async {
        while counter.load(Ordering::SeqCst) < minimum {
+            if let Some(error) = failure.lock().expect("decrypt failure lock poisoned").as_ref() {
+                panic!(
+                    "{description} worker failed after {} successful decrypts: {error}",
+                    counter.load(Ordering::SeqCst)
+                );
+            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
    })
    .await
-    .unwrap_or_else(|_| panic!("timed out waiting for {description}"));
+    .unwrap_or_else(|_| {
+        panic!(
+            "timed out after {timeout:?} waiting for {description}: completed {}, expected {minimum}",
+            counter.load(Ordering::SeqCst)
+        )
+    });
 }

 async fn wait_for_file(path: &Path, description: &str) {
@@ -189,7 +211,8 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
    request: DecryptRequest,
    expected: Vec<u8>,
    completed: Arc<AtomicU64>,
-    failed: Arc<AtomicBool>,
+    allow_failover_errors: Arc<AtomicBool>,
+    failure: Arc<Mutex<Option<String>>>,
    stop: CancellationToken,
 ) {
    while !stop.is_cancelled() {
@@ -197,8 +220,18 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
            Ok(response) if response.plaintext == expected => {
                completed.fetch_add(1, Ordering::SeqCst);
            }
-            Ok(_) | Err(_) => {
-                failed.store(true, Ordering::SeqCst);
+            Ok(_) => {
+                *failure.lock().expect("decrypt failure lock poisoned") =
+                    Some("decrypt returned unexpected plaintext".to_string());
+                return;
+            }
+            Err(rustfs_kms::KmsError::BackendError { .. } | rustfs_kms::KmsError::OperationTimedOut { .. })
+                if allow_failover_errors.load(Ordering::SeqCst) =>
+            {
+                tokio::time::sleep(FAILOVER_ERROR_POLL_INTERVAL).await;
+            }
+            Err(error) => {
+                *failure.lock().expect("decrypt failure lock poisoned") = Some(error.to_string());
                return;
            }
        }
@@ -296,7 +329,9 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
    );

    let stop = CancellationToken::new();
-    let failed = Arc::new(AtomicBool::new(false));
+    let allow_failover_errors = Arc::new(AtomicBool::new(false));
+    let kv2_failure = Arc::new(Mutex::new(None));
+    let transit_failure = Arc::new(Mutex::new(None));
    let kv2_completed = Arc::new(AtomicU64::new(0));
    let transit_completed = Arc::new(AtomicU64::new(0));
    let kv2_worker = tokio::spawn(decrypt_loop(
@@ -304,7 +339,8 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
        kv2_request,
        kv2_data_key.plaintext_key,
        Arc::clone(&kv2_completed),
-        Arc::clone(&failed),
+        Arc::clone(&allow_failover_errors),
+        Arc::clone(&kv2_failure),
        stop.clone(),
    ));
    let transit_worker = tokio::spawn(decrypt_loop(
@@ -312,12 +348,21 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
        transit_request,
        transit_data_key.plaintext_key,
        Arc::clone(&transit_completed),
-        Arc::clone(&failed),
+        Arc::clone(&allow_failover_errors),
+        Arc::clone(&transit_failure),
        stop.clone(),
    ));

-    wait_for_count(&kv2_completed, 2, "two healthy KV2 decrypts").await;
-    wait_for_count(&transit_completed, 2, "two healthy Transit decrypts").await;
+    wait_for_count(&kv2_completed, &kv2_failure, 2, "two healthy KV2 decrypts", HEALTHY_PROGRESS_TIMEOUT).await;
+    wait_for_count(
+        &transit_completed,
+        &transit_failure,
+        2,
+        "two healthy Transit decrypts",
+        HEALTHY_PROGRESS_TIMEOUT,
+    )
+    .await;
+    allow_failover_errors.store(true, Ordering::SeqCst);
    std::fs::write(&marker, b"ready").expect("publish failover readiness marker");

    wait_for_file(&elected, "the replacement Vault leader").await;
@@ -326,18 +371,39 @@ async fn exercise_failover(snapshotter: &Snapshotter) {

    let kv2_after_election = kv2_completed.load(Ordering::SeqCst) + 2;
    let transit_after_election = transit_completed.load(Ordering::SeqCst) + 2;
-    wait_for_count(&kv2_completed, kv2_after_election, "post-failover KV2 decrypts").await;
-    wait_for_count(&transit_completed, transit_after_election, "post-failover Transit decrypts").await;
+    wait_for_count(
+        &kv2_completed,
+        &kv2_failure,
+        kv2_after_election,
+        "post-failover KV2 decrypts",
+        POST_FAILOVER_PROGRESS_TIMEOUT,
+    )
+    .await;
+    wait_for_count(
+        &transit_completed,
+        &transit_failure,
+        transit_after_election,
+        "post-failover Transit decrypts",
+        POST_FAILOVER_PROGRESS_TIMEOUT,
+    )
+    .await;

    stop.cancel();
    kv2_worker.await.expect("KV2 decrypt worker must join");
    transit_worker.await.expect("Transit decrypt worker must join");
-    assert!(!failed.load(Ordering::SeqCst), "no decrypt may fail or return different plaintext");
+    assert!(
+        kv2_failure.lock().expect("KV2 failure lock poisoned").is_none(),
+        "no KV2 decrypt may fail or return different plaintext"
+    );
+    assert!(
+        transit_failure.lock().expect("Transit failure lock poisoned").is_none(),
+        "no Transit decrypt may fail or return different plaintext"
+    );
 }

 #[test]
 #[ignore = "requires a real three-node Vault Raft cluster; run scripts/test/vault_ha_kms_live.sh"]
-fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
+fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
    let recorder = DebuggingRecorder::new();
    let snapshotter = recorder.snapshotter();
    metrics::with_local_recorder(&recorder, || {
@@ -349,11 +415,6 @@ fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
    });
    let snapshot = snapshotter.snapshot().into_vec();

-    assert_eq!(
-        counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "circuit_open")]),
-        0,
-        "a bounded leader election must not open the circuit"
-    );
    assert_eq!(
        counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]),
        0,
@@ -28,9 +28,8 @@ use rustfs_common::heal_channel::HealScanMode;
 use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
 pub use rustfs_data_usage::{
    AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
-    DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, DataUsageSnapshotSetState, LEGACY_DATA_USAGE_OBJECT_NAME,
-    PrefixUsageEntry, PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, hash_path,
-    prefix_usage_in_cache,
+    DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry,
+    PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, hash_path, prefix_usage_in_cache,
 };
 use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
 use tokio::time::{Duration, Instant, sleep, timeout};
@@ -345,18 +344,6 @@ pub struct DataUsageCacheInfo {
    pub scan_plan_digest: Option<DataUsageScanPlanDigest>,
    #[serde(default)]
    pub cache_key_format: u16,
-    /// Whether the entries retained while a set scan was incomplete come
-    /// from a prior complete set snapshot.  This is observational input only.
-    #[serde(default)]
-    pub lkg_snapshot_complete: bool,
-    #[serde(default)]
-    pub lkg_next_cycle: Option<u64>,
-    #[serde(default)]
-    pub lkg_last_update: Option<SystemTime>,
-    #[serde(default)]
-    pub lkg_leader_epoch: Option<u64>,
-    #[serde(default)]
-    pub lkg_scan_plan_digest: Option<DataUsageScanPlanDigest>,
 }

 impl Serialize for DataUsageCacheInfo {
@@ -366,7 +353,7 @@ impl Serialize for DataUsageCacheInfo {
    {
        // Keep this metadata map-encoded so older readers can ignore fields
        // appended by newer scanner versions during rolling upgrades.
-        let mut state = serializer.serialize_map(Some(21))?;
+        let mut state = serializer.serialize_map(Some(16))?;
        state.serialize_entry("name", &self.name)?;
        state.serialize_entry("next_cycle", &self.next_cycle)?;
        state.serialize_entry("leader_epoch", &self.leader_epoch)?;
@@ -383,11 +370,6 @@ impl Serialize for DataUsageCacheInfo {
        state.serialize_entry("snapshot_complete", &self.snapshot_complete)?;
        state.serialize_entry("scan_plan_digest", &self.scan_plan_digest)?;
        state.serialize_entry("cache_key_format", &self.cache_key_format)?;
-        state.serialize_entry("lkg_snapshot_complete", &self.lkg_snapshot_complete)?;
-        state.serialize_entry("lkg_next_cycle", &self.lkg_next_cycle)?;
-        state.serialize_entry("lkg_last_update", &self.lkg_last_update)?;
-        state.serialize_entry("lkg_leader_epoch", &self.lkg_leader_epoch)?;
-        state.serialize_entry("lkg_scan_plan_digest", &self.lkg_scan_plan_digest)?;
        state.end()
    }
 }
@@ -2274,8 +2274,9 @@ async fn final_data_usage_publication_defer_reason(
            }
        }
        ScannerCycleStatus::Deferred(reason) => Some(reason),
-        // Incomplete cycles may publish a non-authoritative observational
-        // snapshot when at least one set has a usable current/LKG view.
+        // Incomplete cycles do not publish a usage snapshot. Keep the
+        // decision permissive so existing partial-cycle handling remains
+        // unchanged if a future scanner path emits a bookkeeping update.
        ScannerCycleStatus::Incomplete => None,
    }
 }
@@ -198,7 +198,7 @@ where
            data_usage_info.usage_snapshot_authoritative_baseline = Some(authoritative.snapshot_identity());
        }

-        if !data_usage_info.is_complete_bucket_usage_snapshot() && !data_usage_info.usage_snapshot_partial {
+        if !data_usage_info.is_complete_bucket_usage_snapshot() {
            error!(
                target: "rustfs::scanner",
                event = EVENT_SCANNER_PERSIST_STATE,
@@ -18,8 +18,8 @@ use crate::scanner_folder::{ScannerItem, scan_data_folder};
 use crate::sleeper::SCANNER_SLEEPER;
 use crate::{
    DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, DataUsageCache, DataUsageCacheInfo, DataUsageCachePrepareOutcome,
-    DataUsageCacheSource, DataUsageEntry, DataUsageEntryInfo, DataUsageInfo, DataUsageScanPlanDigest, DataUsageSnapshotSetState,
-    ScannerError, SizeSummary, TierStats,
+    DataUsageCacheSource, DataUsageEntry, DataUsageEntryInfo, DataUsageInfo, DataUsageScanPlanDigest, ScannerError, SizeSummary,
+    TierStats,
 };
 use futures::future::join_all;
 use metrics::counter;
@@ -278,17 +278,6 @@ async fn publish_usage_snapshot(
    Ok(true)
 }

-async fn publish_observational_snapshot(
-    updates: &mpsc::Sender<DataUsageInfo>,
-    mut data_usage_info: DataUsageInfo,
-) -> Result<bool> {
-    data_usage_info.usage_snapshot_complete = false;
-    data_usage_info.usage_snapshot_partial = true;
-    data_usage_info.usage_snapshot_converged = Some(false);
-    send_data_usage_update(updates, data_usage_info).await?;
-    Ok(true)
-}
-
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum ScannerCycleActivityStatus {
    Unchanged,
@@ -188,7 +188,7 @@ pub(super) fn completed_data_usage_info(
    }

    let mut total = DataUsageEntry::default();
-    let mut bucket_entries = HashMap::with_capacity(all_buckets.len());
+    let mut buckets_usage = HashMap::with_capacity(all_buckets.len());
    for bucket in all_buckets {
        let mut merged = DataUsageEntry::default();
        for result in results {
@@ -200,14 +200,10 @@ pub(super) fn completed_data_usage_info(
        if !total.checked_merge(&merged) {
            return None;
        }
-        bucket_entries.insert(bucket.clone(), merged);
+        buckets_usage.insert(bucket.clone(), checked_bucket_usage_info(&merged)?);
    }

    let merged_last_update = results.iter().filter_map(|result| result.info.last_update).max()?;
-    let buckets_usage = bucket_entries
-        .iter()
-        .map(|(bucket, entry)| Some((bucket.clone(), checked_bucket_usage_info(entry)?)))
-        .collect::<Option<HashMap<_, _>>>()?;
    let bucket_sizes = buckets_usage
        .iter()
        .map(|(bucket, usage)| (bucket.clone(), usage.size))
@@ -229,145 +225,6 @@ pub(super) fn completed_data_usage_info(
    Some((data_usage_info, merged_last_update))
 }

-/// Build a non-authoritative view from the set snapshots that completed this
-/// cycle plus compatible per-set last-known-good caches.  The caller must
-/// persist this result only on the observational object; a missing set is
-/// intentionally represented by an incomplete state and is never treated as
-/// an empty set.
-pub(super) fn observational_data_usage_info(
-    results: &[DataUsageCache],
-    expected_sources: &HashSet<DataUsageCacheSource>,
-    all_buckets: &[String],
-    expected_plan_digest: DataUsageScanPlanDigest,
-    scanner_cycle: u64,
-    leader_epoch: u64,
-) -> Option<(DataUsageInfo, SystemTime)> {
-    let mut by_source = HashMap::with_capacity(results.len());
-    for result in results {
-        let source = result.info.source?;
-        if !expected_sources.contains(&source) || by_source.insert(source, result).is_some() {
-            return None;
-        }
-    }
-
-    let mut usable = Vec::new();
-    let mut set_states = Vec::with_capacity(expected_sources.len());
-    let mut sources = expected_sources.iter().copied().collect::<Vec<_>>();
-    sources.sort_by_key(|source| (source.pool_index, source.set_index));
-    for source in sources {
-        let result = by_source.get(&source).copied();
-        let current = result.filter(|result| {
-            result.info.snapshot_complete
-                && result.info.next_cycle == scanner_cycle
-                && result.info.leader_epoch == leader_epoch
-                && result.info.scan_plan_digest == Some(expected_plan_digest)
-        });
-        let lkg = result.filter(|result| {
-            !result.info.snapshot_complete
-                && result.info.lkg_snapshot_complete
-                && result.info.lkg_scan_plan_digest == Some(expected_plan_digest)
-                && result.info.lkg_leader_epoch.is_some_and(|epoch| {
-                    epoch < leader_epoch
-                        || (epoch == leader_epoch && result.info.lkg_next_cycle.is_some_and(|cycle| cycle <= scanner_cycle))
-                })
-        });
-        let current_snapshot = current.is_some();
-        let selected = current.or(lkg);
-        if let Some(selected) = selected {
-            let (cycle, epoch, digest, last_update, complete) = if current_snapshot {
-                (
-                    Some(selected.info.next_cycle),
-                    Some(selected.info.leader_epoch),
-                    selected.info.scan_plan_digest.map(|digest| digest.0),
-                    selected.info.last_update,
-                    true,
-                )
-            } else {
-                (
-                    selected.info.lkg_next_cycle,
-                    selected.info.lkg_leader_epoch,
-                    selected.info.lkg_scan_plan_digest.map(|digest| digest.0),
-                    selected.info.lkg_last_update,
-                    false,
-                )
-            };
-            set_states.push(DataUsageSnapshotSetState {
-                pool_index: u64::try_from(source.pool_index).ok()?,
-                set_index: u64::try_from(source.set_index).ok()?,
-                scanner_cycle: cycle,
-                scanner_epoch: epoch,
-                scan_plan_digest: digest,
-                complete,
-                tombstone: false,
-            });
-            usable.push((selected, last_update));
-        } else {
-            set_states.push(DataUsageSnapshotSetState {
-                pool_index: u64::try_from(source.pool_index).ok()?,
-                set_index: u64::try_from(source.set_index).ok()?,
-                scanner_cycle: None,
-                scanner_epoch: None,
-                scan_plan_digest: Some(expected_plan_digest.0),
-                complete: false,
-                tombstone: false,
-            });
-        }
-    }
-    if usable.is_empty() {
-        return None;
-    }
-
-    let mut total = DataUsageEntry::default();
-    let mut bucket_entries = HashMap::with_capacity(all_buckets.len());
-    let mut merged_last_update = None;
-    for (result, last_update) in usable {
-        if let Some(update) = last_update {
-            merged_last_update = Some(merged_last_update.map_or(update, |current: SystemTime| current.max(update)));
-        }
-        for bucket in all_buckets {
-            let Some(entry) = result.checked_flatten(bucket) else {
-                continue;
-            };
-            let bucket_entry = bucket_entries.entry(bucket.clone()).or_insert_with(DataUsageEntry::default);
-            if !bucket_entry.checked_merge(&entry) {
-                return None;
-            }
-            if !total.checked_merge(&entry) {
-                return None;
-            }
-        }
-    }
-    let merged_last_update = merged_last_update?;
-    let buckets_usage = bucket_entries
-        .iter()
-        .map(|(bucket, entry)| Some((bucket.clone(), checked_bucket_usage_info(entry)?)))
-        .collect::<Option<HashMap<_, _>>>()?;
-    Some((
-        DataUsageInfo {
-            last_update: Some(merged_last_update),
-            scanner_cycle: Some(scanner_cycle),
-            scanner_epoch: Some(leader_epoch),
-            objects_total_count: u64::try_from(total.objects).ok()?,
-            versions_total_count: u64::try_from(total.versions).ok()?,
-            delete_markers_total_count: u64::try_from(total.delete_markers).ok()?,
-            objects_total_size: u64::try_from(total.size).ok()?,
-            tier_stats: total.all_tier_stats.filter(|tiers| !tiers.is_empty()),
-            buckets_count: u64::try_from(buckets_usage.len()).ok()?,
-            bucket_sizes: buckets_usage
-                .iter()
-                .map(|(bucket, usage)| (bucket.clone(), usage.size))
-                .collect(),
-            buckets_usage,
-            usage_snapshot_complete: false,
-            usage_snapshot_partial: true,
-            usage_snapshot_converged: Some(false),
-            usage_snapshot_set_states: set_states,
-            ..Default::default()
-        },
-        merged_last_update,
-    ))
-}
-
 pub(super) async fn send_cache_root_entry_info(
    bucket_result_tx: &mpsc::Sender<DataUsageEntryInfo>,
    cache: &DataUsageCache,
@@ -40,21 +40,6 @@ impl ScannerIOCache for SetDisks {
        let set_label = self.set_index.to_string();

        let source = DataUsageCacheSource::new(self.pool_index, self.set_index);
-        let mut old_cache = DataUsageCache::default();
-        if let Err(e) = old_cache.load(self.clone(), DATA_USAGE_CACHE_NAME).await {
-            warn!(
-                target: "rustfs::scanner::io",
-                event = EVENT_SCANNER_CACHE_PERSIST_STATE,
-                component = LOG_COMPONENT_SCANNER,
-                subsystem = LOG_SUBSYSTEM_IO,
-                pool = self.pool_index,
-                set = self.set_index,
-                cache_name = DATA_USAGE_CACHE_NAME,
-                state = "old_cache_load_failed",
-                error = %e,
-                "Scanner old data usage cache load failed; rebuilding from bucket caches"
-            );
-        }
        if buckets.is_empty() {
            let now = SystemTime::now();
            let mut cache = DataUsageCache {
@@ -95,24 +80,6 @@ impl ScannerIOCache for SetDisks {
                "Scanner set state found no online disks"
            );
            reset_disk_bucket_scan_gauges(&pool_label, &set_label);
-            let lkg = old_cache.info.snapshot_complete.then(|| old_cache.clone());
-            let mut incomplete_scope = lkg.clone().unwrap_or_default();
-            incomplete_scope.info.name = DATA_USAGE_ROOT.to_string();
-            incomplete_scope.info.next_cycle = want_cycle;
-            incomplete_scope.info.last_update = None;
-            incomplete_scope.info.leader_epoch = leader_epoch;
-            incomplete_scope.info.source = Some(source);
-            incomplete_scope.info.snapshot_complete = false;
-            incomplete_scope.info.scan_plan_digest = Some(scan_plan_digest);
-            incomplete_scope.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT;
-            if let Some(lkg) = lkg {
-                incomplete_scope.info.lkg_snapshot_complete = true;
-                incomplete_scope.info.lkg_next_cycle = Some(lkg.info.next_cycle);
-                incomplete_scope.info.lkg_last_update = lkg.info.last_update;
-                incomplete_scope.info.lkg_leader_epoch = Some(lkg.info.leader_epoch);
-                incomplete_scope.info.lkg_scan_plan_digest = lkg.info.scan_plan_digest;
-            }
-            let _ = updates.send(incomplete_scope).await;
            return Ok(());
        }
        // Preserve the original set topology across capability filtering. During
@@ -195,24 +162,6 @@ impl ScannerIOCache for SetDisks {
                "Scanner set state found no usable namespace scanner disks"
            );
            reset_disk_bucket_scan_gauges(&pool_label, &set_label);
-            let lkg = old_cache.info.snapshot_complete.then(|| old_cache.clone());
-            let mut incomplete_scope = lkg.clone().unwrap_or_default();
-            incomplete_scope.info.name = DATA_USAGE_ROOT.to_string();
-            incomplete_scope.info.next_cycle = want_cycle;
-            incomplete_scope.info.last_update = None;
-            incomplete_scope.info.leader_epoch = leader_epoch;
-            incomplete_scope.info.source = Some(source);
-            incomplete_scope.info.snapshot_complete = false;
-            incomplete_scope.info.scan_plan_digest = Some(scan_plan_digest);
-            incomplete_scope.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT;
-            if let Some(lkg) = lkg {
-                incomplete_scope.info.lkg_snapshot_complete = true;
-                incomplete_scope.info.lkg_next_cycle = Some(lkg.info.next_cycle);
-                incomplete_scope.info.lkg_last_update = lkg.info.last_update;
-                incomplete_scope.info.lkg_leader_epoch = Some(lkg.info.leader_epoch);
-                incomplete_scope.info.lkg_scan_plan_digest = lkg.info.scan_plan_digest;
-            }
-            let _ = updates.send(incomplete_scope).await;
            return Ok(());
        }
        let set_disk_inventory = Arc::new(scanner_set_disk_inventory(self.as_ref()).await);
@@ -254,15 +203,22 @@ impl ScannerIOCache for SetDisks {
        record_disk_bucket_scans_active(0, &pool_label, &set_label);
        let _reset_disk_bucket_scan_gauges = DiskBucketScanGaugeReset::new(pool_label.clone(), set_label.clone());

-        let old_lkg = old_cache.info.snapshot_complete.then(|| {
-            (
-                old_cache.info.next_cycle,
-                old_cache.info.last_update,
-                old_cache.info.leader_epoch,
-                old_cache.info.scan_plan_digest,
-            )
-        });
-        let prepare_outcome = match old_cache.prepare_for_scan(
+        let mut old_cache = DataUsageCache::default();
+        if let Err(e) = old_cache.load(self.clone(), DATA_USAGE_CACHE_NAME).await {
+            warn!(
+                target: "rustfs::scanner::io",
+                event = EVENT_SCANNER_CACHE_PERSIST_STATE,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_IO,
+                pool = self.pool_index,
+                set = self.set_index,
+                cache_name = DATA_USAGE_CACHE_NAME,
+                state = "old_cache_load_failed",
+                error = %e,
+                "Scanner old data usage cache load failed; rebuilding from bucket caches"
+            );
+        }
+        match old_cache.prepare_for_scan(
            DATA_USAGE_ROOT,
            want_cycle,
            leader_epoch,
@@ -303,16 +259,7 @@ impl ScannerIOCache for SetDisks {
                );
                return Ok(());
            }
-            outcome => outcome,
-        };
-        if matches!(prepare_outcome, DataUsageCachePrepareOutcome::Reused)
-            && let Some((cycle, last_update, epoch, digest)) = old_lkg
-        {
-            old_cache.info.lkg_snapshot_complete = true;
-            old_cache.info.lkg_next_cycle = Some(cycle);
-            old_cache.info.lkg_last_update = last_update;
-            old_cache.info.lkg_leader_epoch = Some(epoch);
-            old_cache.info.lkg_scan_plan_digest = digest;
+            DataUsageCachePrepareOutcome::Reused | DataUsageCachePrepareOutcome::Reset => {}
        }

        let mut cache = DataUsageCache {
@@ -1152,29 +1099,23 @@ impl ScannerIOCache for SetDisks {
                cache.info.next_cycle = want_cycle;
                cache.info.last_update.get_or_insert_with(SystemTime::now);
                cache.info.snapshot_complete = true;
-                cache.info.lkg_snapshot_complete = false;
-                cache.info.lkg_next_cycle = None;
-                cache.info.lkg_last_update = None;
-                cache.info.lkg_leader_epoch = None;
-                cache.info.lkg_scan_plan_digest = None;
                cache.clone()
            };
            let _ = persist_and_publish_cache_snapshot(self.clone(), &updates, cache_snapshot, cache_cycle_floor.as_ref()).await;
        } else {
-            let mut incomplete_scope = cache_mutex.lock().await.clone();
-            incomplete_scope.info.name = DATA_USAGE_ROOT.to_string();
-            incomplete_scope.info.next_cycle = want_cycle;
-            incomplete_scope.info.last_update = None;
-            incomplete_scope.info.leader_epoch = leader_epoch;
-            incomplete_scope.info.source = Some(source);
-            incomplete_scope.info.snapshot_complete = false;
-            incomplete_scope.info.scan_plan_digest = Some(scan_plan_digest);
-            incomplete_scope.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT;
-            incomplete_scope.info.lkg_snapshot_complete = old_cache.info.lkg_snapshot_complete;
-            incomplete_scope.info.lkg_next_cycle = old_cache.info.lkg_next_cycle;
-            incomplete_scope.info.lkg_last_update = old_cache.info.lkg_last_update;
-            incomplete_scope.info.lkg_leader_epoch = old_cache.info.lkg_leader_epoch;
-            incomplete_scope.info.lkg_scan_plan_digest = old_cache.info.lkg_scan_plan_digest;
+            let incomplete_scope = DataUsageCache {
+                info: DataUsageCacheInfo {
+                    name: DATA_USAGE_ROOT.to_string(),
+                    next_cycle: want_cycle,
+                    leader_epoch,
+                    source: Some(source),
+                    snapshot_complete: false,
+                    scan_plan_digest: Some(scan_plan_digest),
+                    cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+                    ..Default::default()
+                },
+                cache: HashMap::new(),
+            };
            if let Err(e) = updates.send(incomplete_scope).await {
                error!(
                    target: "rustfs::scanner::io",
@@ -234,7 +234,6 @@ impl ScannerIOCycle for ECStore {
                let active_set_scans_clone = active_set_scans.clone();

                let (tx, mut rx) = mpsc::channel::<DataUsageCache>(1);
-                let failed_scope_tx = tx.clone();

                // Spawn task to receive and store results
                let receiver_fut = tokio::spawn(async move {
@@ -315,21 +314,6 @@ impl ScannerIOCycle for ECStore {
                            state = "set_scan_failed",
                            "Scanner set scan failed; continuing cycle"
                        );
-                        let _ = failed_scope_tx
-                            .send(DataUsageCache {
-                                info: DataUsageCacheInfo {
-                                    name: DATA_USAGE_ROOT.to_string(),
-                                    next_cycle: want_cycle_clone,
-                                    leader_epoch,
-                                    source: Some(source),
-                                    snapshot_complete: false,
-                                    scan_plan_digest: Some(scan_plan_digest),
-                                    cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
-                                    ..Default::default()
-                                },
-                                cache: HashMap::new(),
-                            })
-                            .await;
                        let mut first_err = first_err_mutex_clone.lock().await;
                        record_set_scan_failure(&mut first_err, e);
                    }
@@ -386,19 +370,6 @@ impl ScannerIOCycle for ECStore {
            budget_elapsed,
            ctx.is_cancelled(),
        );
-        let observational_usage = completed_usage
-            .is_none()
-            .then(|| {
-                observational_data_usage_info(
-                    &results,
-                    &expected_sources,
-                    &all_bucket_names,
-                    scan_plan_digest,
-                    want_cycle,
-                    leader_epoch,
-                )
-            })
-            .flatten();
        let structurally_complete_snapshot = result.is_ok() && completed_all_sets && completed_usage.is_some();
        let cycle_status = classify_nsscanner_cycle(
            structurally_complete_snapshot,
@@ -410,10 +381,6 @@ impl ScannerIOCycle for ECStore {
        );
        if let Some((data_usage_info, _)) = completed_usage {
            publish_usage_snapshot(&updates, cycle_status, data_usage_info).await?;
-        } else if !ctx.is_cancelled()
-            && let Some((data_usage_info, _)) = observational_usage
-        {
-            publish_observational_snapshot(&updates, data_usage_info).await?;
        }
        let dirty_usage_clear = should_clear_dirty_usage_snapshot(
            result.is_ok(),
@@ -105,160 +105,6 @@ fn completed_data_usage_info_for_test(
    completed_data_usage_info(results, &expected_sources, all_buckets, true, budget_elapsed, cancelled)
 }

-fn lkg_root_cache(bucket: &str, objects: usize, source: DataUsageCacheSource) -> DataUsageCache {
-    let mut cache = completed_root_cache(bucket, objects, 10, source);
-    cache.info.snapshot_complete = false;
-    cache.info.next_cycle = 8;
-    cache.info.leader_epoch = 3;
-    cache.info.lkg_snapshot_complete = true;
-    cache.info.lkg_next_cycle = Some(7);
-    cache.info.lkg_last_update = cache.info.last_update;
-    cache.info.lkg_leader_epoch = Some(3);
-    cache.info.lkg_scan_plan_digest = Some(TEST_PLAN_DIGEST);
-    cache
-}
-
-#[test]
-fn partial_usage_is_observational_not_authoritative_for_quota() {
-    let all_buckets = vec!["bucket".to_string()];
-    let current_source = DataUsageCacheSource::new(0, 0);
-    let stalled_source = DataUsageCacheSource::new(1, 0);
-    let mut current = completed_root_cache("bucket", 2, 20, current_source);
-    current.info.next_cycle = 8;
-    current.info.leader_epoch = 3;
-    let stalled = lkg_root_cache("bucket", 1, stalled_source);
-    let expected = HashSet::from([current_source, stalled_source]);
-
-    assert!(
-        completed_data_usage_info(&[current.clone(), stalled.clone()], &expected, &all_buckets, true, false, false).is_none()
-    );
-    let (observed, _) = observational_data_usage_info(&[current, stalled], &expected, &all_buckets, TEST_PLAN_DIGEST, 8, 3)
-        .expect("a completed set should produce an observational view");
-    assert!(observed.usage_snapshot_partial);
-    assert!(!observed.usage_snapshot_complete);
-    assert_eq!(observed.usage_snapshot_converged, Some(false));
-    assert_eq!(observed.usage_snapshot_set_states.len(), 2);
-}
-
-#[test]
-fn lkg_scope_does_not_count_as_current_cycle_completion() {
-    let source = DataUsageCacheSource::new(0, 0);
-    let mut lkg = lkg_root_cache("bucket", 1, source);
-    lkg.info.last_update = None;
-    let expected = HashSet::from([source]);
-    assert!(!scanner_results_form_complete_snapshot(&[lkg], &expected));
-}
-
-#[test]
-fn stale_quota_uses_complete_baseline_plus_positive_deltas() {
-    let all_buckets = vec!["bucket".to_string()];
-    let source = DataUsageCacheSource::new(0, 0);
-    let mut current = completed_root_cache("bucket", 3, 20, source);
-    current.info.next_cycle = 8;
-    current.info.leader_epoch = 3;
-    let expected = HashSet::from([source]);
-    let (observed, _) = observational_data_usage_info(&[current], &expected, &all_buckets, TEST_PLAN_DIGEST, 8, 3)
-        .expect("complete set data is a valid observational baseline");
-    assert_eq!(observed.objects_total_size, 30);
-    assert_eq!(observed.usage_snapshot_set_states[0].complete, true);
-}
-
-#[test]
-fn negative_delta_waits_for_set_reconciliation() {
-    let all_buckets = vec!["bucket".to_string()];
-    let source = DataUsageCacheSource::new(0, 0);
-    let mut stalled = lkg_root_cache("bucket", 4, source);
-    stalled.info.lkg_scan_plan_digest = Some(DataUsageScanPlanDigest([9; 32]));
-    let expected = HashSet::from([source]);
-    assert!(observational_data_usage_info(&[stalled], &expected, &all_buckets, TEST_PLAN_DIGEST, 8, 3).is_none());
-}
-
-#[test]
-fn set_membership_add_remove_uses_generation_and_tombstone() {
-    let state = DataUsageSnapshotSetState {
-        pool_index: 1,
-        set_index: 2,
-        scanner_cycle: Some(9),
-        scanner_epoch: Some(4),
-        scan_plan_digest: Some(TEST_PLAN_DIGEST.0),
-        complete: false,
-        tombstone: true,
-    };
-    let encoded = serde_json::to_vec(&state).expect("set state should serialize");
-    let decoded: DataUsageSnapshotSetState = serde_json::from_slice(&encoded).expect("set state should deserialize");
-    assert_eq!(decoded, state);
-
-    let snapshot = DataUsageInfo {
-        last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(10)),
-        scanner_cycle: Some(9),
-        scanner_epoch: Some(4),
-        buckets_count: 0,
-        usage_snapshot_converged: Some(false),
-        usage_snapshot_partial: true,
-        usage_snapshot_set_states: vec![
-            DataUsageSnapshotSetState {
-                pool_index: 0,
-                set_index: 0,
-                scanner_cycle: Some(9),
-                scanner_epoch: Some(4),
-                scan_plan_digest: Some(TEST_PLAN_DIGEST.0),
-                complete: true,
-                tombstone: false,
-            },
-            state,
-        ],
-        ..Default::default()
-    };
-    assert!(snapshot.is_valid_partial_snapshot());
-}
-
-#[test]
-fn old_set_completion_cannot_overwrite_new_aggregate() {
-    let all_buckets = vec!["bucket".to_string()];
-    let source = DataUsageCacheSource::new(0, 0);
-    let mut old = completed_root_cache("bucket", 1, 20, source);
-    old.info.next_cycle = 7;
-    old.info.leader_epoch = 2;
-    let expected = HashSet::from([source]);
-    assert!(observational_data_usage_info(&[old], &expected, &all_buckets, TEST_PLAN_DIGEST, 8, 3).is_none());
-}
-
-#[test]
-fn usage_aggregate_survives_restart_and_leader_failover() {
-    let all_buckets = vec!["bucket".to_string()];
-    let source = DataUsageCacheSource::new(0, 0);
-    let mut lkg = lkg_root_cache("bucket", 5, source);
-    lkg.info.lkg_leader_epoch = Some(4);
-    lkg.info.lkg_next_cycle = Some(9);
-    let expected = HashSet::from([source]);
-    let (observed, _) = observational_data_usage_info(&[lkg], &expected, &all_buckets, TEST_PLAN_DIGEST, 10, 5)
-        .expect("compatible LKG should survive a leader change");
-    assert_eq!(observed.usage_snapshot_set_states[0].scanner_epoch, Some(4));
-    assert_eq!(observed.objects_total_size, 50);
-}
-
-#[test]
-fn usage_aggregate_cost_is_linear_in_set_count() {
-    let all_buckets = vec!["bucket".to_string()];
-    let mut results = Vec::new();
-    let mut expected = HashSet::new();
-    for index in 0..32 {
-        let source = DataUsageCacheSource::new(index, 0);
-        expected.insert(source);
-        let mut cache = completed_root_cache("bucket", 1, 20, source);
-        cache.info.next_cycle = 8;
-        cache.info.leader_epoch = 3;
-        results.push(cache);
-    }
-    let (observed, _) = observational_data_usage_info(&results, &expected, &all_buckets, TEST_PLAN_DIGEST, 8, 3)
-        .expect("all set snapshots should aggregate");
-    assert_eq!(observed.objects_total_count, 32);
-    let reversed = results.iter().rev().cloned().collect::<Vec<_>>();
-    let (reversed_observed, _) = observational_data_usage_info(&reversed, &expected, &all_buckets, TEST_PLAN_DIGEST, 8, 3)
-        .expect("reordered set snapshots should aggregate");
-    assert_eq!(observed.usage_snapshot_set_states, reversed_observed.usage_snapshot_set_states);
-}
-
 #[test]
 fn completed_data_usage_info_publishes_tier_stats_across_sets() {
    let all_buckets = vec!["bucket-a".to_string(), "bucket-b".to_string()];
@@ -43,9 +43,6 @@ allow-git = [
    # RustFS fork carrying presigned expiry and constant-time authentication fixes.
    # owner: rustfs-maintainers  review: 2026-10
    "https://github.com/rustfs/s3s.git",
-    # MiMalloc fork pinned for hotpath allocation counting support.
-    # owner: houseme  review: 2026-10
-    "https://github.com/xonatius/mimalloc_rust.git",
 ]

 [bans]
@@ -158,10 +158,11 @@ added by backlog#1153 infra-4.

 ## Coverage

-Line coverage is measured **weekly, not per-PR**, and is non-blocking: it
-exists for visibility and trend, never as a required check. Per-crate ratchets
-for the security-critical crates (iam / kms / policy / crypto) build on this
-baseline later (backlog#1153 infra-6, report-only first).
+Workspace line coverage is measured weekly. Pull requests that touch iam, kms,
+policy, or crypto also run a non-required, report-only comparison against
+`.config/coverage-baselines.toml`. During calibration, a regression is recorded
+in the job summary without failing the job; missing or malformed coverage
+evidence still fails closed (backlog#1153 infra-6).

 - **CI**: `.github/workflows/coverage.yml` runs every Sunday and on manual
  dispatch: `cargo llvm-cov nextest --workspace --exclude e2e_test` under the
@@ -174,6 +175,12 @@ baseline later (backlog#1153 infra-6, report-only first).
  plus the full suite). It prints the same per-crate table via
  `scripts/coverage_per_crate.py` and writes `target/llvm-cov/lcov.info` and
  `coverage.json`.
+- **Security-critical ratchet**: relevant pull requests compare iam / kms /
+  policy / crypto line coverage with the versioned baseline. Drops greater than
+  the configured one-percentage-point calibration threshold are marked
+  `REGRESSION (report-only)`. The weekly summary runs the same comparison so
+  calibration continues even when no relevant pull request is open. Baseline
+  changes require a linked coverage run and a reviewed explanation.
 - **Trend comparison**: each run's job summary is the weekly per-crate
  snapshot — open two runs from the Actions history (workflow "coverage") and
  compare their tables. For line-level diffs, download the two runs'
@@ -0,0 +1,149 @@
+# CI gate matrix
+
+This file is the source of truth for which validation runs on each event, its
+configured wall-clock budget, and whether it can block a merge. Test taxonomy,
+naming, and nextest serialization rules remain in [README.md](README.md); e2e
+membership and counts remain in
+[e2e-suite-inventory.md](e2e-suite-inventory.md).
+
+The distinction between **required** and **report-only** is load-bearing:
+a failing job blocks a merge only when its exact check name is present in the
+live `main` ruleset. A workflow name, a `merge_group` trigger, or a red PR check
+does not make a job required by itself.
+
+## Required merge checks
+
+The live `main` ruleset (`6436880`) currently requires exactly these contexts:
+
+| Required context | Producer | Validation |
+|---|---|---|
+| `CLA Check` | `.github/workflows/cla.yml` | Contributor agreement |
+| `Quick Checks` | `.github/workflows/ci.yml` | Formatting and repository guard scripts |
+| `Test and Lint` | `.github/workflows/ci.yml` | Clippy, workspace nextest excluding `e2e_test`, doctests, and migration proofs |
+
+For pull requests limited to the paths excluded by the main CI workflow,
+`.github/workflows/ci-docs-only.yml` reports `Quick Checks` and
+`Test and Lint` under the same names. It runs the real quick checks and the
+planning-document guard; it does not claim that Rust compilation or runtime
+tests ran. Despite the workflow name, these paths also include selected deploy,
+workflow, and lock files.
+
+Verify the live rule rather than trusting this snapshot before changing merge
+policy:
+
+```bash
+gh api repos/rustfs/rustfs/rulesets/6436880 \
+  --jq '.rules[] | select(.type == "required_status_checks") | .parameters'
+```
+
+The ruleset currently has `strict_required_status_checks_policy=false`.
+`Continuous Integration` accepts `merge_group` events and runs `e2e-full` for
+them, but `End-to-End Tests (full merge gate)` is not currently a required
+context. Therefore the repository is prepared to test a merge-queue SHA, but
+the workflow alone does not prove that every merge passed that lane.
+
+## Pull request and merge matrix
+
+Budgets below are job `timeout-minutes`, not typical runtimes. “Report-only”
+means the result is visible and actionable but is not in the live required
+context list.
+
+| Event | Validation | Budget | Merge status | Reproduction |
+|---|---|---:|---|---|
+| PR, non-doc change | `Quick Checks` | 10 min | Required | `make pre-commit` (broader local umbrella) |
+| PR, non-doc change | `Test and Lint` | 90 min | Required | `cargo nextest run --profile ci --all --exclude e2e_test` |
+| PR, non-doc change | `Typos` | 10 min | Report-only | `typos` |
+| PR, non-doc change | `ILM Integration (serial)` | 90 min | Report-only | Use the exact command in `.github/workflows/ci.yml` |
+| PR, non-doc change | rio-v2 / swift / sftp test-and-lint variants | 90 min each | Report-only | `cargo nextest run` with the workflow's feature set |
+| PR, non-doc change | `Build RustFS Debug Binary` | 30 min | Report-only; prerequisite for black-box lanes | `cargo build -p rustfs --bins` |
+| PR, non-doc change | `io_uring Integration (real)` | 30 min | Report-only | `cargo test -p rustfs-ecstore --lib uring_ -- --test-threads=1 --nocapture` |
+| PR, non-doc change | `End-to-End Tests` (`e2e-smoke` plus `s3s-e2e`) | 30 min | Report-only | `cargo nextest run --profile e2e-smoke -p e2e_test`; then `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>` |
+| PR, non-doc change | `S3 Implemented Tests` | 60 min | Report-only | Build `rustfs`, then run `scripts/s3-tests/run.sh` with `DEPLOY_MODE=binary`, `TEST_MODE=single`, and `MAXFAIL=0` |
+| PR, non-doc change | `S3 Lifecycle Behavior Tests` | 30 min | Report-only | Use the accelerated scanner environment in `.github/workflows/ci.yml` with `scripts/s3-tests/run.sh` |
+| PR touching dependency or workflow inputs | Cargo Deny / Workflow Pin Report / Dependency Review | 20 / 5 / 30 min | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
+| PR touching architecture rules or architecture docs | `Architecture Migration Rules` | 10 min | Report-only | `scripts/check_architecture_migration_rules.sh` |
+| PR touching Nix or workspace manifests | `Nix Build & Check` | 60 min | Report-only | `nix flake check` |
+| PR limited to main-CI-excluded paths | companion `Quick Checks` and `Test and Lint` | 10 min each | Required | `git diff --check`; `make doc-paths-check` when documentation paths changed |
+| `merge_group` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Standard required contexts only; `e2e-full` report-only | `cargo nextest run --profile e2e-full -p e2e_test` |
+| Push to `main` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Post-merge detection | Same as `merge_group` |
+| PR touching fuzz inputs or harness paths | Build plus five 60-second fuzz smoke targets | 60 min build; 30 min per target | Report-only | `MAX_TOTAL_TIME=60 ./scripts/fuzz/run.sh` |
+| PR touching selected ecstore disk/format paths | `Rename Safety` on Windows | 60 min | Report-only | Run the four `cargo test -p rustfs-ecstore --lib <filter>` commands in `windows-filesystem.yml` on Windows |
+
+The authoritative e2e filters live in `.config/nextest.toml`; extend a profile
+instead of adding a second ad-hoc selector. Before a profile runs,
+`scripts/check_test_wiring.py` compares its exact membership to the committed
+digest so a silent test drop fails closed.
+
+## Scheduled and manual validation
+
+Scheduled lanes are independent fault domains. They do not block a pull
+request, but their workflow-local gate can fail the run and scheduled failures
+are routed to the shared failure-issue action. The scheduled-validation
+watchdog and freshness workflow separately detect incomplete runs and missing
+schedules.
+
+| Cadence (UTC unless noted) | Workflow / validation | Budget | Verdict and artifacts | Reproduction |
+|---|---|---:|---|---|
+| Daily 02:17 | Fuzz: five nightly corpus targets | 60 min build; 60 min per target | Gate; corpus/crash artifacts, scheduled failure alert | `MAX_TOTAL_TIME=<seconds> ./scripts/fuzz/run.sh` |
+| Daily 03:17 | MinIO interop (EC + SSE read parity) | 40 min | Gate; scheduled failure alert | Dispatch `minio-interop.yml` or follow its pinned Docker fixture steps |
+| Daily 04:29 | Replication / cluster-fault / protocol e2e | 45 / 90 / 90 min | Three independent gates; JUnit, membership, and server logs | `cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--profile e2e-nightly`; `-j 1 --profile e2e-protocols` |
+| Daily 06:31 | Warp performance A/B | 180 min | Regression budget gate; A/B summaries and server logs | `bash scripts/run_hotpath_warp_abba.sh --help` |
+| Daily 00:07 Asia/Shanghai (16:07 UTC previous day) | Nightly GNU build and Vault lanes | 150 / 90 / 60 min | Build, live Vault, and HA failover gates | Use the commands and pinned Vault images in `nightly-gnu.yml` |
+| Daily 03:23 | Security Audit | 20 / 5 min, plus 30 min on PR dependency review | Cargo Deny and workflow-pin gates; scheduled failure alert | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
+| Daily 23:47 | Scheduled Validation Freshness | 10 min | Fails when a critical schedule was never created or is stale | Dispatch `scheduled-validation-freshness.yml` |
+| Sunday 00:11 | Full `Continuous Integration` matrix | Per-job budgets above | Weekly variant coverage, including dormant rio-v2 binary/e2e lanes | Dispatch `ci.yml` |
+| Sunday 01:13 | Seven-platform build matrix | 150 min per platform | Build/package integrity; scheduled failure alert | Dispatch `build.yml` with an exact platform set |
+| Sunday 02:19 | Ceph s3-tests full sweep: single and real four-node, four shards each | 180 min per shard | Compatibility gate; report, JUnit, exact node IDs, and server logs | `scripts/s3-tests/run.sh` against an existing single or distributed target |
+| Sunday 06:41 | Mint | 120 min | **Report-only by design**; per-suite PASS/FAIL/NA and raw `log.json` | Reproduce the pinned Docker sequence in `mint.yml` or dispatch it |
+| Sunday 07:43 | Workspace line coverage | 120 min | Report-only trend; lcov and JSON retained 90 days | `make coverage` |
+| Monthly, day 1 06:37 | Runner Hygiene | 15 min | Validates runner ephemerality; scheduled failure alert | Dispatch `runner-hygiene.yml` |
+
+Manual `workflow_dispatch` exists for the scheduled workflows above. Manual
+runs are debugging evidence and intentionally do not open scheduled-failure
+issues. A manual performance run may explicitly allow a known regression; that
+override must not be treated as an ordinary passing baseline.
+
+## Release validation
+
+Release validation is post-merge and tag-driven; it does not substitute for a
+pull-request gate.
+
+| Event | Validation | Budget | Result |
+|---|---|---:|---|
+| Push to `main` or weekly schedule | `Build and Release` platform matrix | 150 min per platform | Build artifacts for all selected targets; no release publication on a main push |
+| Valid release or preview tag | `Build and Release` plus asset checks | 150 min per platform | Draft release, checksummed assets, and publish step |
+| Successful non-preview release-tag build | Docker image build and image scan | 60 min build; 30 min scan | Multi-architecture images plus vulnerability report |
+| Successful release-tag build | DEB/RPM packaging | 30 min per architecture | Packages and checksum files uploaded to the release |
+| Successful non-preview release-tag build | Helm template test and package | 30 min build; 30 min publish | Versioned chart and repository index |
+
+Use an exact preview tag for end-to-end release rehearsal. Manual dispatches
+are backfill/debug paths and do not prove the automatic `workflow_run` chain.
+
+## Evidence requirements
+
+A green check is useful only when it proves the intended behavior ran:
+
+- Record the exact commit SHA and run URL.
+- Separate product failure from runner prerequisites, service readiness, and
+  cancellation. Repair the precondition, then rerun the exact workload.
+- Preserve membership manifests, JUnit, raw compatibility logs, seeds, and
+  server logs where the workflow provides them.
+- For a bug fix or a new fault checker, provide sensitivity evidence: the old
+  behavior or an intentional mutation must fail the new oracle, and the fixed
+  behavior must pass it.
+- Never promote a report-only lane to required from one green run. Require at
+  least 14 days and 30 representative pull requests with at least 99% complete
+  execution, then update the ruleset and this table together.
+
+## Change checklist
+
+Update this file in the same pull request when any of these change:
+
+- workflow triggers, job names, timeouts, or nextest profile ownership;
+- required status contexts or strict/merge-queue policy;
+- scheduled cadence, alert routing, artifact contract, or local reproduction;
+- report-only versus gating semantics.
+
+Do not copy per-module test counts here. Update
+[e2e-suite-inventory.md](e2e-suite-inventory.md) and its enforced membership
+digest instead.
@@ -336,13 +336,13 @@ opentelemetry = { workspace = true }
 tracing-opentelemetry = { workspace = true }
 # Data structures
 hashbrown = { workspace = true, features = ["serde", "rayon"] }
-mimalloc = { workspace = true }
+rustfs-mimalloc = { workspace = true }

 [target.'cfg(target_os = "linux")'.dependencies]
 libsystemd.workspace = true

 [target.'cfg(not(target_os = "windows"))'.dependencies]
-libmimalloc-sys.workspace = true
+rustfs-mimalloc-sys.workspace = true

 [dev-dependencies]
 uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
@@ -369,14 +369,8 @@ pub fn allocator_reclaim_controller_snapshot(ctx: &CancellationToken) -> Allocat
 }

 #[cfg(not(target_os = "windows"))]
-#[allow(unsafe_code)]
 fn collect_allocator_memory(force: bool) -> Result<(), String> {
-    // SAFETY: `mi_collect` is provided by the active global allocator backend
-    // on this target family. It is explicitly intended to reclaim retained
-    // pages/segments and does not require additional invariants from the caller.
-    unsafe {
-        libmimalloc_sys::mi_collect(force);
-    }
+    rustfs_mimalloc::MiMalloc::collect(force);
    Ok(())
 }

@@ -26,22 +26,22 @@ struct MiMallocAllocator;
 unsafe impl GlobalAlloc for MiMallocAllocator {
    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
        // SAFETY: the caller upholds GlobalAlloc's contract for layout.
-        unsafe { mimalloc::MiMalloc.alloc(layout) }
+        unsafe { rustfs_mimalloc::MiMalloc.alloc(layout) }
    }

    unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
        // SAFETY: the caller upholds GlobalAlloc's contract for layout.
-        unsafe { mimalloc::MiMalloc.alloc_zeroed(layout) }
+        unsafe { rustfs_mimalloc::MiMalloc.alloc_zeroed(layout) }
    }

    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
        // SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
-        unsafe { mimalloc::MiMalloc.dealloc(ptr, layout) }
+        unsafe { rustfs_mimalloc::MiMalloc.dealloc(ptr, layout) }
    }

    unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
        // SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
-        unsafe { mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
+        unsafe { rustfs_mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
    }
 }

@@ -51,7 +51,7 @@ static GLOBAL: hotpath::CountingAllocator<MiMallocAllocator> = hotpath::Counting

 #[cfg(not(all(feature = "hotpath", feature = "hotpath-alloc")))]
 #[global_allocator]
-static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
+static GLOBAL: rustfs_mimalloc::MiMalloc = rustfs_mimalloc::MiMalloc;

 fn main() {
    let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main").build();
@@ -71,8 +71,9 @@ mod tests {
        allocation.extend_from_slice(&[7_u8; 64]);

        assert_eq!(allocation.len(), 64);
+        let heap = rustfs_mimalloc::heap::Heap::main();
        // SAFETY: the live Vec pointer is valid to inspect for heap ownership.
-        assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(allocation.as_ptr().cast()) });
+        assert!(unsafe { heap.contains(allocation.as_ptr()) });
    }

    #[test]
@@ -85,12 +86,13 @@ mod tests {
        let layout = Layout::from_size_align(32, 8).expect("valid test allocation layout");
        let grown_layout = Layout::from_size_align(64, 8).expect("valid grown test allocation layout");
        let allocator = super::MiMallocAllocator;
+        let heap = rustfs_mimalloc::heap::Heap::main();

        // SAFETY: The pointer is checked for null before use and later released
        // through the same allocator with the corresponding layout.
        let ptr = unsafe { allocator.alloc_zeroed(layout) };
        assert!(!ptr.is_null());
-        assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(ptr.cast()) });
+        assert!(unsafe { heap.contains(ptr) });
        assert!(unsafe { std::slice::from_raw_parts(ptr, 32).iter().all(|byte| *byte == 0) });

        // SAFETY: `ptr` was allocated by `allocator` with `layout`; on failure
@@ -102,7 +104,7 @@ mod tests {
            panic!("mimalloc realloc failed in allocator smoke test");
        }

-        assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(grown_ptr.cast()) });
+        assert!(unsafe { heap.contains(grown_ptr) });
        // SAFETY: `grown_ptr` was reallocated by `allocator` and is released
        // with the matching grown layout.
        unsafe { allocator.dealloc(grown_ptr, grown_layout) };
@@ -17,10 +17,7 @@ use rustfs_io_metrics::{
    record_cpu_usage, record_memory_usage, record_process_memory_split,
 };
 use serde::Serialize;
-#[cfg(any(test, not(target_os = "windows")))]
 use serde_json::Value;
-#[cfg(not(target_os = "windows"))]
-use std::ffi::CStr;
 use std::path::Path;
 use std::sync::{Arc, Mutex, OnceLock};
 use std::time::Duration;
@@ -231,7 +228,18 @@ fn read_cgroup_memory_snapshot() -> Option<CgroupMemorySnapshot> {
    read_cgroup_v2().or_else(read_cgroup_v1)
 }

-#[cfg(any(test, not(target_os = "windows")))]
+fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
+    let json = rustfs_mimalloc::MiMalloc::stats_json();
+    if json.is_empty() {
+        return None;
+    }
+    let observation = parse_mimalloc_stats_json(&json)?;
+    Some(AllocatorMemorySnapshot {
+        backend: crate::allocator_reclaim::allocator_backend(),
+        observation,
+    })
+}
+
 fn numeric_json_value(value: &Value) -> Option<u64> {
    match value {
        Value::Number(number) => number
@@ -242,7 +250,6 @@ fn numeric_json_value(value: &Value) -> Option<u64> {
    }
 }

-#[cfg(any(test, not(target_os = "windows")))]
 fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
    match value {
        Value::Object(fields) => fields
@@ -254,7 +261,6 @@ fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
    }
 }

-#[cfg(any(test, not(target_os = "windows")))]
 fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64> {
    match value {
        Value::Object(fields) => {
@@ -271,12 +277,10 @@ fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64>
    }
 }

-#[cfg(any(test, not(target_os = "windows")))]
 fn mimalloc_stat_current(value: &Value, metric: &str) -> Option<u64> {
    mimalloc_stat_field(value, metric, "current")
 }

-#[cfg(any(test, not(target_os = "windows")))]
 fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64> {
    metrics
        .iter()
@@ -285,7 +289,6 @@ fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64
        .filter(|value| *value > 0)
 }

-#[cfg(any(test, not(target_os = "windows")))]
 fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservation> {
    let value = serde_json::from_str::<Value>(stats_json).ok()?;
    let malloc_metrics = ["malloc_normal", "malloc_huge"];
@@ -312,33 +315,6 @@ fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservat
    }
 }

-#[cfg(not(target_os = "windows"))]
-#[allow(unsafe_code)]
-fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
-    // SAFETY: `mi_stats_get_json` returns a null-terminated JSON buffer owned by
-    // mimalloc when called with a null input buffer. The mimalloc API requires
-    // freeing that buffer with `mi_free`; parsing finishes before the buffer is freed.
-    let observation = unsafe {
-        let stats_ptr = libmimalloc_sys::mi_stats_get_json(0, std::ptr::null_mut());
-        if stats_ptr.is_null() {
-            return None;
-        }
-
-        let observation = CStr::from_ptr(stats_ptr).to_str().ok().and_then(parse_mimalloc_stats_json);
-        libmimalloc_sys::mi_free(stats_ptr.cast());
-        observation?
-    };
-    Some(AllocatorMemorySnapshot {
-        backend: crate::allocator_reclaim::allocator_backend(),
-        observation,
-    })
-}
-
-#[cfg(target_os = "windows")]
-fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
-    None
-}
-
 fn configured_memory_observability_interval_secs() -> u64 {
    rustfs_utils::get_env_u64(ENV_MEMORY_OBSERVABILITY_INTERVAL_SECS, DEFAULT_MEMORY_OBSERVABILITY_INTERVAL_SECS).max(1)
 }
@@ -566,6 +542,13 @@ mod tests {
        assert_eq!(parse_mimalloc_stats_json(r#"{ "allocator": "unknown" }"#), None);
    }

+    #[test]
+    fn read_allocator_memory_snapshot_uses_mimalloc_stats_json() {
+        let snapshot = super::read_allocator_memory_snapshot();
+        #[cfg(not(target_os = "windows"))]
+        assert!(snapshot.is_some(), "allocator snapshot should be available on non-Windows");
+    }
+
    #[test]
    fn memory_observability_snapshot_reports_disabled_when_metrics_are_disabled() {
        let snapshot = build_memory_observability_status_snapshot(false, 15, false);
@@ -0,0 +1,264 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Compare security-critical crate line coverage with the report-only baseline."""
+
+import argparse
+import json
+import math
+import os
+import sys
+import tempfile
+import tomllib
+from pathlib import Path
+
+from coverage_per_crate import fmt_pct, load_coverage
+
+
+SECURITY_CRATES = ("crates/iam", "crates/kms", "crates/policy", "crates/crypto")
+
+
+def load_baselines(path: str) -> tuple[float, dict[str, tuple[int, int]]]:
+    with open(path, "rb") as fh:
+        config = tomllib.load(fh)
+
+    if config.get("phase") != "report-only":
+        raise ValueError("coverage baseline phase must be report-only")
+
+    raw_allowed_drop = config["allowed_drop_percentage_points"]
+    if isinstance(raw_allowed_drop, bool) or not isinstance(raw_allowed_drop, (int, float)):
+        raise ValueError("allowed_drop_percentage_points must be a number")
+    allowed_drop = float(raw_allowed_drop)
+    if not math.isfinite(allowed_drop) or allowed_drop < 0:
+        raise ValueError("allowed_drop_percentage_points must be finite and non-negative")
+
+    baselines: dict[str, tuple[int, int]] = {}
+    for crate, values in config["crates"].items():
+        covered = values["covered"]
+        count = values["count"]
+        if type(covered) is not int or type(count) is not int:
+            raise ValueError(f"invalid baseline for {crate}: covered and count must be integers")
+        if covered < 0 or count <= 0 or covered > count:
+            raise ValueError(f"invalid baseline for {crate}: {covered}/{count}")
+        baselines[crate] = (covered, count)
+    missing = [crate for crate in SECURITY_CRATES if crate not in baselines]
+    unexpected = sorted(set(baselines).difference(SECURITY_CRATES))
+    if missing or unexpected:
+        raise ValueError(f"coverage baseline crate set mismatch: missing={missing}, unexpected={unexpected}")
+    return allowed_drop, baselines
+
+
+def compare(
+    current: dict[str, list[int]],
+    baselines: dict[str, tuple[int, int]],
+    allowed_drop: float,
+) -> list[tuple[str, int, int, int, int, float, bool]]:
+    rows = []
+    for crate, (baseline_covered, baseline_count) in baselines.items():
+        if crate not in current:
+            raise ValueError(f"coverage report is missing {crate}")
+        covered, count = current[crate]
+        if type(covered) is not int or type(count) is not int:
+            raise ValueError(f"invalid coverage for {crate}: covered and count must be integers")
+        if covered < 0 or count <= 0 or covered > count:
+            raise ValueError(f"invalid coverage for {crate}: {covered}/{count}")
+        current_pct = 100.0 * covered / count
+        baseline_pct = 100.0 * baseline_covered / baseline_count
+        delta = current_pct - baseline_pct
+        rows.append((crate, covered, count, baseline_covered, baseline_count, delta, delta < -allowed_drop))
+    return rows
+
+
+def print_report(rows: list[tuple[str, int, int, int, int, float, bool]], allowed_drop: float) -> None:
+    print("## Security-critical coverage ratchet (report-only)")
+    print()
+    print(f"Calibration threshold: a drop greater than {allowed_drop:.2f} percentage points is reported as a regression.")
+    print()
+    print("| Crate | Current | Baseline | Delta | Status |")
+    print("|---|---:|---:|---:|---|")
+    for crate, covered, count, baseline_covered, baseline_count, delta, regressed in rows:
+        status = "REGRESSION (report-only)" if regressed else "OK"
+        print(
+            f"| `{crate}` | {fmt_pct(covered, count)} ({covered}/{count}) "
+            f"| {fmt_pct(baseline_covered, baseline_count)} ({baseline_covered}/{baseline_count}) "
+            f"| {delta:+.2f} pp | {status} |"
+        )
+    print()
+    print("This calibration phase records regressions without failing the job; malformed or incomplete evidence still fails closed.")
+
+
+def self_test() -> None:
+    with tempfile.TemporaryDirectory() as tmp:
+        root = Path(tmp)
+        coverage = root / "coverage.json"
+        baseline = root / "baseline.toml"
+        coverage_data = {
+            "data": [
+                {
+                    "files": [
+                        {
+                            "filename": str(root / "crates/iam/src/lib.rs"),
+                            "summary": {"lines": {"covered": 80, "count": 100}},
+                        },
+                        {
+                            "filename": str(root / "crates/kms/src/lib.rs"),
+                            "summary": {"lines": {"covered": 90, "count": 100}},
+                        },
+                        {
+                            "filename": str(root / "crates/policy/src/lib.rs"),
+                            "summary": {"lines": {"covered": 90, "count": 100}},
+                        },
+                        {
+                            "filename": str(root / "crates/crypto/src/lib.rs"),
+                            "summary": {"lines": {"covered": 90, "count": 100}},
+                        },
+                    ],
+                    "totals": {"lines": {"covered": 350, "count": 400}},
+                }
+            ]
+        }
+        coverage.write_text(json.dumps(coverage_data), encoding="utf-8")
+        baseline_text = """phase = "report-only"
+allowed_drop_percentage_points = 1.0
+[crates."crates/iam"]
+covered = 90
+count = 100
+[crates."crates/kms"]
+covered = 85
+count = 100
+[crates."crates/policy"]
+covered = 90
+count = 100
+[crates."crates/crypto"]
+covered = 90
+count = 100
+"""
+        baseline.write_text(baseline_text, encoding="utf-8")
+        current, _ = load_coverage(str(coverage), str(root))
+        allowed_drop, baselines = load_baselines(str(baseline))
+        rows = compare(current, baselines, allowed_drop)
+        assert [row[-1] for row in rows] == [True, False, False, False]
+        try:
+            compare({"crates/iam": current["crates/iam"]}, baselines, allowed_drop)
+        except ValueError as error:
+            assert str(error) == "coverage report is missing crates/kms"
+        else:
+            raise AssertionError("missing crate must fail closed")
+        try:
+            compare({**current, "crates/iam": [101, 100]}, baselines, allowed_drop)
+        except ValueError as error:
+            assert str(error) == "invalid coverage for crates/iam: 101/100"
+        else:
+            raise AssertionError("invalid coverage must fail closed")
+        for invalid_threshold in ("true", '"1.0"', "nan", "inf", "-inf"):
+            baseline.write_text(
+                baseline_text.replace("allowed_drop_percentage_points = 1.0", f"allowed_drop_percentage_points = {invalid_threshold}"),
+                encoding="utf-8",
+            )
+            try:
+                load_baselines(str(baseline))
+            except ValueError:
+                pass
+            else:
+                raise AssertionError(f"non-finite threshold {invalid_threshold} must fail closed")
+        for field, invalid_values in (
+            ("covered", ("true", '"90"', "90.0", "90.5")),
+            ("count", ("true", '"100"', "100.0", "100.5")),
+        ):
+            for invalid_value in invalid_values:
+                baseline.write_text(
+                    baseline_text.replace(f"{field} = {90 if field == 'covered' else 100}", f"{field} = {invalid_value}", 1),
+                    encoding="utf-8",
+                )
+                try:
+                    load_baselines(str(baseline))
+                except ValueError:
+                    pass
+                else:
+                    raise AssertionError(f"non-integer baseline {field} {invalid_value} must fail closed")
+        for covered, count in (
+            (True, 100),
+            (80, True),
+            (80.0, 100),
+            (80, 100.0),
+            (float("nan"), 100),
+            (80, float("inf")),
+        ):
+            try:
+                compare({**current, "crates/iam": [covered, count]}, baselines, allowed_drop)
+            except ValueError:
+                pass
+            else:
+                raise AssertionError(f"invalid aggregate coverage {covered}/{count} must fail closed")
+        lines = coverage_data["data"][0]["files"][0]["summary"]["lines"]
+        for field, invalid_values in (
+            ("covered", (True, "80", 80.0, 80.5, float("nan"), float("inf"), float("-inf"))),
+            ("count", (True, "100", 100.0, 100.5, float("nan"), float("inf"), float("-inf"))),
+        ):
+            original = lines[field]
+            for invalid_value in invalid_values:
+                lines[field] = invalid_value
+                coverage.write_text(json.dumps(coverage_data), encoding="utf-8")
+                try:
+                    load_coverage(str(coverage), str(root))
+                except ValueError:
+                    pass
+                else:
+                    raise AssertionError(f"invalid raw coverage {field} {invalid_value} must fail closed")
+            lines[field] = original
+        baseline.write_text(
+            baseline_text.replace(
+                '[crates."crates/crypto"]\ncovered = 90\ncount = 100\n',
+                "",
+            ),
+            encoding="utf-8",
+        )
+        try:
+            load_baselines(str(baseline))
+        except ValueError:
+            pass
+        else:
+            raise AssertionError("missing security-crate baseline must fail closed")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("coverage_json", nargs="?")
+    parser.add_argument("--baseline", default=".config/coverage-baselines.toml")
+    parser.add_argument("--repo-root", default=os.getcwd())
+    parser.add_argument("--self-test", action="store_true")
+    args = parser.parse_args()
+
+    if args.self_test:
+        self_test()
+        print("security coverage self-test passed")
+        return 0
+    if not args.coverage_json:
+        parser.error("coverage_json is required unless --self-test is used")
+
+    try:
+        current, _ = load_coverage(args.coverage_json, os.path.abspath(args.repo_root))
+        allowed_drop, baselines = load_baselines(args.baseline)
+        rows = compare(current, baselines, allowed_drop)
+    except (OSError, ValueError, KeyError, IndexError, json.JSONDecodeError, tomllib.TOMLDecodeError) as error:
+        print(f"error: {error}", file=sys.stderr)
+        return 1
+
+    print_report(rows, allowed_drop)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -47,6 +47,31 @@ def fmt_pct(covered: int, count: int) -> str:
    return f"{100.0 * covered / count:.2f}%" if count else "—"


+def _line_counts(lines: dict[str, int], source: str) -> tuple[int, int]:
+    covered = lines["covered"]
+    count = lines["count"]
+    if type(covered) is not int or type(count) is not int or covered < 0 or count < 0 or covered > count:
+        raise ValueError(f"invalid line coverage for {source}: {covered}/{count}")
+    return covered, count
+
+
+def load_coverage(path: str, root: str) -> tuple[dict[str, list[int]], dict[str, int]]:
+    with open(path, encoding="utf-8") as fh:
+        export = json.load(fh)
+
+    data = export["data"][0]
+    files = data["files"]
+    total_covered, total_count = _line_counts(data["totals"]["lines"], "totals")
+
+    crates: dict[str, list[int]] = {}
+    for f in files:
+        covered, count = _line_counts(f["summary"]["lines"], f["filename"])
+        acc = crates.setdefault(crate_label(f["filename"], root), [0, 0])
+        acc[0] += covered
+        acc[1] += count
+    return crates, {"covered": total_covered, "count": total_count}
+
+
 def main() -> int:
    if len(sys.argv) < 2 or len(sys.argv) > 3:
        print(__doc__.strip(), file=sys.stderr)
@@ -54,24 +79,12 @@ def main() -> int:
    path = sys.argv[1]
    root = os.path.abspath(sys.argv[2] if len(sys.argv) == 3 else os.getcwd())

-    with open(path, encoding="utf-8") as fh:
-        export = json.load(fh)
-
    try:
-        data = export["data"][0]
-        files = data["files"]
-        totals = data["totals"]["lines"]
-    except (KeyError, IndexError) as exc:
+        crates, totals = load_coverage(path, root)
+    except (KeyError, IndexError, ValueError) as exc:
        print(f"error: unexpected llvm-cov JSON shape ({exc})", file=sys.stderr)
        return 1

-    crates: dict[str, list[int]] = {}
-    for f in files:
-        lines = f["summary"]["lines"]
-        acc = crates.setdefault(crate_label(f["filename"], root), [0, 0])
-        acc[0] += lines["covered"]
-        acc[1] += lines["count"]
-
    rows = sorted(
        crates.items(),
        key=lambda kv: (100.0 * kv[1][0] / kv[1][1]) if kv[1][1] else 101.0,
@@ -241,7 +241,7 @@ env \
  RUSTFS_TEST_VAULT_FAILOVER_MARKER="$MARKER" \
  RUSTFS_TEST_VAULT_OLD_LEADER="$OLD_LEADER" \
  cargo test -p rustfs-kms --test vault_ha_failover_live \
-    vault_raft_leader_failure_preserves_kv2_and_transit_decrypts -- \
+    vault_raft_leader_failure_recovers_kv2_and_transit_decrypts -- \
    --ignored --nocapture --test-threads=1 &
 TEST_PID=$!
Author	SHA1	Message	Date
houseme	6c99df80f2	Merge branch 'main' into overtrue/report-security-coverage-ratchet	2026-08-23 12:32:30 +08:00
Zhengchao An	23a2c7d776	test(kms): stabilize Vault failover validation (#6385 ) * test(kms): bound Vault failover progress wait * ci(nightly): honor manual dispatch ref * test(kms): preserve Vault worker failures * test(kms): validate Vault circuit recovery	2026-08-23 12:32:11 +08:00
唐小鸭	5f72209446	fix(ecstore): keep unknown-size sentinel in create_bitrot_writer (#6380 ) SSE and compression wrap the payload so its length is unknown and advertise HashReader::SIZE_PRESERVE_LAYER (-1). Every layer preserved that sentinel except create_bitrot_writer, which clamped it to 0 before calling DiskAPI::create_file. RemoteDisk forwards that size verbatim in the put_file_stream query, so remote peers were told the body was empty. Since the authenticated put-file trailer (#5868) the receiver used the declared size to split body from trailer, turning the clamp into a fatal "auth trailer has trailing data" failure for every SSE PUT on multi-node deployments (rc.2). #6320 relaxed the receiver to only trust size > 0; this change fixes the sender so the sentinel survives end to end and the wire no longer conflates empty objects with unknown-length streams. Refs #6331	2026-08-23 12:29:52 +08:00
Zhengchao An	b6ba89d9e4	docs(testing): document CI gate matrix (#6412 )	2026-08-23 12:09:06 +08:00
houseme	648d5166e2	feat(allocator): replace mimalloc/libmimalloc-sys with rustfs-mimalloc/rustfs-mimalloc-sys (#6404 ) Replace the upstream xonatius/mimalloc_rust.git fork (mimalloc + libmimalloc-sys) with the published rustfs-mimalloc (v0.5.0) and rustfs-mimalloc-sys (v0.5.0) crates from crates.io. The new crates are based on mimalloc V3 (v3.5.0) and provide: - MiMalloc global allocator with safe API (collect, stats_json, process_info) - Heap management and arena operations (heap module) - Full FFI bindings to mimalloc V3 Changes: - Workspace deps: mimalloc + libmimalloc-sys (git) → rustfs-mimalloc + rustfs-mimalloc-sys (crates.io) - allocator_reclaim.rs: libmimalloc_sys::mi_collect → rustfs_mimalloc::MiMalloc::collect - memory_observability.rs: raw FFI mi_stats_get_json → MiMalloc::stats_json() - main.rs: heap ownership tests use Heap::contains() (V3 API) - deny.toml: remove xonatius/mimalloc_rust.git from allow-git Co-authored-by: heihutu <heihutu@gmail.com>	2026-08-23 12:07:25 +08:00
houseme	84eb5aebef	fix(ecstore): remove inline write debug noise (#6408 ) * fix(ecstore): remove inline write debug noise Co-Authored-By: heihutu <heihutu@gmail.com> * fix(ecstore): satisfy warning-as-error lints Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>	2026-08-23 12:07:20 +08:00
overtrue	2786c6813a	ci(coverage): cancel superseded PR runs	2026-08-23 04:31:40 +08:00
overtrue	d3fe16442e	fix(ci): reject malformed coverage counts	2026-08-23 04:09:21 +08:00
overtrue	60a05220d2	fix(ci): reject incomplete coverage baselines	2026-08-23 03:19:23 +08:00
overtrue	d37746fcc3	chore: merge main into coverage ratchet	2026-08-23 02:41:10 +08:00
overtrue	d614fdbe34	ci(coverage): preserve report publication margin	2026-08-23 00:34:18 +08:00
overtrue	09de44df08	ci(coverage): add security ratchet calibration	2026-08-22 20:43:36 +08:00