docs(operations): split scanner excess alerts into English and Chinese pages

The page shipped Chinese-only; keep it as scanner-excess-alerts_zh.md and
add a faithful English translation at the original path, cross-linked at
the top of both.

Co-Authored-By: heihutu <heihutu@gmail.com>

Cargo.lock

@@ -10251,6 +10251,7 @@ dependencies = [
  "rustfs-ecstore",
  "rustfs-filemeta",
  "rustfs-lock",
+ "rustfs-s3-types",
  "rustfs-storage-api",
  "rustfs-utils",
  "s3s",

crates/ecstore/src/api/mod.rs

@@ -380,7 +380,7 @@ pub mod erasure {
 
 pub mod event {
     pub use crate::event::name::EventName;
-    pub use crate::services::event_notification::{EventArgs, register_event_dispatch_hook};
+    pub use crate::services::event_notification::{EventArgs, register_event_dispatch_hook, send_event};
 }
 
 pub mod global {

crates/scanner/Cargo.toml

@@ -108,6 +108,9 @@ temp-env = { workspace = true }
 tempfile = { workspace = true }
 uuid = { workspace = true, features = ["v4", "serde", "fast-rng", "macro-diagnostics"] }
 tokio = { workspace = true, features = ["test-util", "fs", "rt-multi-thread"] }
+# Test-only: pins the emitted scanner alert wire names against the canonical
+# EventName string forms subscribers configure (rustfs/backlog#1868).
+rustfs-s3-types.workspace = true
 # Enables the shared MockWarmBackend / xl.meta assertion helpers exposed via
 # the ecstore `api::tier::test_util` facade module (rustfs/backlog#1148 ilm-6).
 rustfs-ecstore = { workspace = true, features = ["test-util"] }

crates/scanner/src/scanner_folder.rs

@@ -12,10 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::fs::FileType;
 use std::io::ErrorKind;
-use std::sync::{Arc, Once};
+use std::sync::{Arc, Mutex, Once};
 use std::time::{Duration, Instant, SystemTime};
 
 use crate::ReplTargetSizeSummary;
@@ -32,6 +32,7 @@ use crate::scanner_io::{
     SCANNER_SKIP_FILE_ERROR, ScannerIODisk as _, is_scanner_metadata_corrupt_error, is_scanner_metadata_transient_error,
 };
 use crate::sleeper::DynamicSleeper;
+use crate::storage_api::owner::{EcstoreEventArgs, ecstore_send_event};
 use metrics::{counter, describe_counter};
 use rustfs_common::heal_channel::{
     HEAL_DELETE_DANGLING, HealAdmissionDropReason, HealAdmissionResult, HealChannelPriority, HealChannelRequest,
@@ -97,6 +98,101 @@ const METRIC_SCANNER_EXCESS_FOLDERS_TOTAL: &str = "rustfs_scanner_excess_folders
 const METRIC_SCANNER_PENDING_HEAL_PRUNE_TOTAL: &str = "rustfs_scanner_pending_heal_prune_total";
 const METRIC_SCANNER_PENDING_HEAL_MALFORMED_TOTAL: &str = "rustfs_scanner_pending_heal_malformed_total";
 const MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET: usize = 128;
+
+// --- scanner excess alerts as S3 notification events (rustfs/backlog#1868) --
+//
+// The excess-versions / excess-version-size / excess-folders alerts were
+// metrics-and-logs only; subscribers (consoles, external auditors) had no way
+// to hear them. MinIO emits s3:ObjectManyVersions / s3:ObjectLargeVersions /
+// s3:PrefixManyFolders for the same conditions — RustFS carries those as
+// EventName::Scanner* with the wire names below. Without a cooldown a single
+// over-threshold object would re-emit on every scan cycle (~a minute), so
+// emissions are edge-held per (kind, bucket, object) for 24h.
+
+/// `s3:Scanner:ManyVersions` (MinIO `s3:ObjectManyVersions`).
+pub const EVENT_SCANNER_MANY_VERSIONS: &str = "s3:Scanner:ManyVersions";
+/// `s3:Scanner:LargeVersions` (MinIO `s3:ObjectLargeVersions`).
+pub const EVENT_SCANNER_LARGE_VERSIONS: &str = "s3:Scanner:LargeVersions";
+/// `s3:Scanner:BigPrefix` (MinIO `s3:PrefixManyFolders`).
+pub const EVENT_SCANNER_BIG_PREFIX: &str = "s3:Scanner:BigPrefix";
+const ENV_SCANNER_ALERT_COOLDOWN_SECS: &str = "RUSTFS_SCANNER_ALERT_COOLDOWN_SECS";
+const DEFAULT_SCANNER_ALERT_COOLDOWN_SECS: u64 = 86_400;
+/// Hard cap on distinct cooldown keys; a pathological number of over-threshold
+/// objects clears the map wholesale instead of growing without bound (the
+/// worst case is one re-emission per still-hot key per scan cycle).
+const MAX_SCANNER_ALERT_COOLDOWN_KEYS: usize = 4096;
+
+/// Distinct alert kinds sharing one cooldown map.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+enum ScannerAlertKind {
+    ManyVersions,
+    LargeVersions,
+    BigPrefix,
+}
+
+type ScannerAlertCooldownKey = (ScannerAlertKind, String, String);
+type ScannerAlertCooldownMap = HashMap<ScannerAlertCooldownKey, Instant>;
+
+static SCANNER_ALERT_EMISSION_COOLDOWN: Mutex<Option<ScannerAlertCooldownMap>> = Mutex::new(None);
+
+fn scanner_alert_cooldown() -> Duration {
+    let raw = std::env::var(ENV_SCANNER_ALERT_COOLDOWN_SECS)
+        .ok()
+        .and_then(|v| v.parse::<u64>().ok());
+    Duration::from_secs(raw.unwrap_or(DEFAULT_SCANNER_ALERT_COOLDOWN_SECS))
+}
+
+/// Edge-held emission gate: returns `true` (and records the cooldown) only
+/// when this (kind, bucket, object) last fired longer than the cooldown ago —
+/// or never. Metrics and logs stay level-triggered every cycle; only the
+/// notification events are held back.
+fn scanner_alert_emission_allows(kind: ScannerAlertKind, bucket: &str, object: &str, cooldown: Duration) -> bool {
+    let key = (kind, bucket.to_string(), object.to_string());
+    let mut guard = SCANNER_ALERT_EMISSION_COOLDOWN
+        .lock()
+        .unwrap_or_else(|poison| poison.into_inner());
+    let guard = guard.get_or_insert_with(ScannerAlertCooldownMap::new);
+    let now = Instant::now();
+    // Expired entries leave first; the cap is still exceeded only when live
+    // keys alone overflow it, in which case a wholesale clear trades one
+    // extra emission per hot key for a hard memory bound.
+    if guard.len() >= MAX_SCANNER_ALERT_COOLDOWN_KEYS {
+        guard.retain(|_, fired_at| now.duration_since(*fired_at) < cooldown);
+        if guard.len() >= MAX_SCANNER_ALERT_COOLDOWN_KEYS {
+            guard.clear();
+        }
+    }
+    match guard.get(&key) {
+        Some(fired_at) if now.duration_since(*fired_at) < cooldown => false,
+        _ => {
+            guard.insert(key, now);
+            true
+        }
+    }
+}
+
+/// Emit a scanner alert as an S3 notification event through the standard
+/// dispatch pipeline. Fire-and-forget: the notify layer owns delivery,
+/// retry, and target filtering; the scanner never waits on it.
+fn emit_scanner_alert_event(event_name: &str, bucket: &str, object: &str, size: i64, details: &[(&str, String)]) {
+    let mut req_params = HashMap::with_capacity(details.len());
+    for (key, value) in details {
+        req_params.insert((*key).to_string(), value.clone());
+    }
+    ecstore_send_event(EcstoreEventArgs {
+        event_name: event_name.to_string(),
+        bucket_name: bucket.to_string(),
+        object: crate::ScannerObjectInfo {
+            bucket: bucket.to_string(),
+            name: object.to_string(),
+            size,
+            ..Default::default()
+        },
+        req_params,
+        user_agent: "Scanner".to_string(),
+        ..Default::default()
+    });
+}
 const MAX_PENDING_SCANNER_HEALS_PER_BUCKET: usize = 10_000;
 
 static SCANNER_INLINE_HEAL_WARN_ONCE: Once = Once::new();
@@ -1197,6 +1293,7 @@ impl ScannerItem {
     fn alert_excessive_versions(&self, remaining_versions: usize, cumulative_size: i64) {
         ensure_scanner_alert_metrics_registered();
         let (too_many_versions, too_large_versions) = should_alert_excessive_versions(remaining_versions, cumulative_size);
+        let object_path = self.object_path();
         if too_many_versions {
             global_metrics().record_scanner_source_executed(ScannerWorkSource::Alerts, 1);
             counter!(
@@ -1204,13 +1301,26 @@ impl ScannerItem {
                 "bucket" => self.bucket.clone()
             )
             .increment(1);
+            if scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, &self.bucket, &object_path, scanner_alert_cooldown())
+            {
+                emit_scanner_alert_event(
+                    EVENT_SCANNER_MANY_VERSIONS,
+                    &self.bucket,
+                    &object_path,
+                    cumulative_size,
+                    &[
+                        ("versions", remaining_versions.to_string()),
+                        ("threshold", scanner_excess_versions_threshold().to_string()),
+                    ],
+                );
+            }
             warn!(
                 target: "rustfs::scanner::folder",
                 event = EVENT_SCANNER_ALERT_STATE,
                 component = LOG_COMPONENT_SCANNER,
                 subsystem = LOG_SUBSYSTEM_FOLDER,
                 bucket = %self.bucket,
-                object = %self.object_path(),
+                object = %object_path,
                 versions = remaining_versions,
                 threshold = scanner_excess_versions_threshold(),
                 state = "excess_versions",
@@ -1224,13 +1334,31 @@ impl ScannerItem {
                 "bucket" => self.bucket.clone()
             )
             .increment(1);
+            if scanner_alert_emission_allows(
+                ScannerAlertKind::LargeVersions,
+                &self.bucket,
+                &object_path,
+                scanner_alert_cooldown(),
+            ) {
+                emit_scanner_alert_event(
+                    EVENT_SCANNER_LARGE_VERSIONS,
+                    &self.bucket,
+                    &object_path,
+                    cumulative_size,
+                    &[
+                        ("versions", remaining_versions.to_string()),
+                        ("cumulativeSize", cumulative_size.to_string()),
+                        ("threshold", scanner_excess_version_size_threshold().to_string()),
+                    ],
+                );
+            }
             warn!(
                 target: "rustfs::scanner::folder",
                 event = EVENT_SCANNER_ALERT_STATE,
                 component = LOG_COMPONENT_SCANNER,
                 subsystem = LOG_SUBSYSTEM_FOLDER,
                 bucket = %self.bucket,
-                object = %self.object_path(),
+                object = %object_path,
                 versions = remaining_versions,
                 cumulative_size,
                 threshold = scanner_excess_version_size_threshold(),
@@ -1611,6 +1739,15 @@ impl FolderScanner {
             "root" => self.root.clone()
         )
         .increment(1);
+        if scanner_alert_emission_allows(ScannerAlertKind::BigPrefix, &self.root, folder, scanner_alert_cooldown()) {
+            emit_scanner_alert_event(
+                EVENT_SCANNER_BIG_PREFIX,
+                &self.root,
+                folder,
+                0,
+                &[("folders", total_folders.to_string()), ("threshold", threshold.to_string())],
+            );
+        }
         warn!(
             target: "rustfs::scanner::folder",
             event = EVENT_SCANNER_ALERT_STATE,
@@ -3076,6 +3213,90 @@ mod tests {
     #[cfg(unix)]
     use std::os::unix::fs::{PermissionsExt, symlink};
     use std::sync::Mutex;
+
+    /// Reset the process-global alert cooldown map; test-only.
+    fn reset_alert_cooldowns() {
+        *SCANNER_ALERT_EMISSION_COOLDOWN
+            .lock()
+            .unwrap_or_else(|poison| poison.into_inner()) = Some(ScannerAlertCooldownMap::new());
+    }
+
+    /// The emitted event-name strings must be exactly what `EventName`
+    /// serializes, or a bucket notification subscribed to the documented name
+    /// would silently never match (rustfs/backlog#1868).
+    #[test]
+    fn scanner_alert_wire_names_match_canonical_event_names() {
+        use rustfs_s3_types::EventName;
+        assert_eq!(EVENT_SCANNER_MANY_VERSIONS, EventName::ScannerManyVersions.to_string());
+        assert_eq!(EVENT_SCANNER_LARGE_VERSIONS, EventName::ScannerLargeVersions.to_string());
+        assert_eq!(EVENT_SCANNER_BIG_PREFIX, EventName::ScannerBigPrefix.to_string());
+    }
+
+    fn cooldown_map_len() -> usize {
+        SCANNER_ALERT_EMISSION_COOLDOWN
+            .lock()
+            .unwrap_or_else(|poison| poison.into_inner())
+            .as_ref()
+            .map(|map| map.len())
+            .unwrap_or(0)
+    }
+
+    /// Backdate every recorded cooldown so the next check fires again.
+    fn expire_all_alert_cooldowns(cooldown: Duration) {
+        let now = Instant::now();
+        let mut guard = SCANNER_ALERT_EMISSION_COOLDOWN
+            .lock()
+            .unwrap_or_else(|poison| poison.into_inner());
+        if let Some(map) = guard.as_mut() {
+            for fired_at in map.values_mut() {
+                if let Some(expired) = now.checked_sub(cooldown + Duration::from_secs(1)) {
+                    *fired_at = expired;
+                }
+            }
+        }
+    }
+
+    /// The emission gate is the only thing standing between an over-threshold
+    /// object and one S3 event per scan cycle, so its edge semantics get
+    /// pinned directly. All scenarios share one #[test] because the cooldown
+    /// map is process-global and parallel tests would read each other's
+    /// firings.
+    #[test]
+    fn scanner_alert_emission_is_edge_held_per_key_and_bounded() {
+        reset_alert_cooldowns();
+        let cooldown = Duration::from_secs(3600);
+
+        // First firing allows, an immediate re-check is held.
+        assert!(scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "bkt", "obj", cooldown));
+        assert!(!scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "bkt", "obj", cooldown));
+
+        // Different kind, object, and bucket are independent keys.
+        assert!(scanner_alert_emission_allows(ScannerAlertKind::LargeVersions, "bkt", "obj", cooldown));
+        assert!(scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "bkt", "other", cooldown));
+        assert!(scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "other", "obj", cooldown));
+        assert_eq!(cooldown_map_len(), 4);
+
+        // After the cooldown elapses the same key fires again.
+        expire_all_alert_cooldowns(cooldown);
+        assert!(scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "bkt", "obj", cooldown));
+
+        // A zero cooldown degenerates to always-emit (operators may want that).
+        assert!(scanner_alert_emission_allows(ScannerAlertKind::BigPrefix, "bkt", "dir", Duration::ZERO));
+        assert!(scanner_alert_emission_allows(ScannerAlertKind::BigPrefix, "bkt", "dir", Duration::ZERO));
+
+        // Hard bound: overflow the cap with zero-cooldown keys and confirm the
+        // map clears rather than growing past it.
+        reset_alert_cooldowns();
+        for index in 0..=(MAX_SCANNER_ALERT_COOLDOWN_KEYS + 8) {
+            let _ = scanner_alert_emission_allows(ScannerAlertKind::BigPrefix, "bkt", &format!("dir-{index}"), Duration::ZERO);
+        }
+        assert!(
+            cooldown_map_len() <= MAX_SCANNER_ALERT_COOLDOWN_KEYS,
+            "cooldown map must stay bounded, got {}",
+            cooldown_map_len()
+        );
+    }
+
     use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
     use temp_env::{with_var, with_var_unset};
     use tracing_subscriber::fmt::MakeWriter;

crates/scanner/src/storage_api.rs

@@ -78,6 +78,7 @@ pub(crate) use rustfs_ecstore::api::disk::{
 pub(crate) use rustfs_ecstore::api::error::{
     Error as EcstoreErrorType, Result as EcstoreResultType, StorageError as EcstoreStorageError,
 };
+pub(crate) use rustfs_ecstore::api::event::{EventArgs as EcstoreEventArgs, send_event as ecstore_send_event};
 #[cfg(test)]
 pub(crate) use rustfs_ecstore::api::layout::{
     EndpointServerPools as EcstoreEndpointServerPools, Endpoints as EcstoreEndpoints, PoolEndpoints as EcstorePoolEndpoints,
@@ -110,8 +111,8 @@ pub(crate) mod owner {
         ECSTORE_BUCKET_META_PREFIX, ECSTORE_RUSTFS_META_BUCKET, ECSTORE_STORAGE_FORMAT_FILE, ECSTORE_STORAGECLASS_RRS,
         ECSTORE_STORAGECLASS_STANDARD, ECSTORE_TRANSITION_COMPLETE, EcstoreBucketTargetSys, EcstoreBucketVersioningSys,
         EcstoreDisk, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskInfo, EcstoreDiskInfoOptions,
-        EcstoreDiskLocation, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreLcEventSrc,
-        EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts,
+        EcstoreDiskLocation, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreEventArgs,
+        EcstoreLcEventSrc, EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts,
         EcstoreReplicationConfigurationExt, EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard,
         EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreTierConfig, EcstoreVersioningApi,
         ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule,
@@ -121,7 +122,7 @@ pub(crate) mod owner {
         ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw,
         ecstore_object_opts_from_object_info, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path,
         ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle,
-        ecstore_save_config, scanner_replication_config_for_lifecycle_eval,
+        ecstore_save_config, ecstore_send_event, scanner_replication_config_for_lifecycle_eval,
     };
 
     #[cfg(test)]

docs/operations/scanner-excess-alerts.md

@@ -0,0 +1,37 @@
+# Scanner Excess Alerts: Metrics, S3 Events, and Thresholds
+
+> 中文版：[scanner-excess-alerts_zh.md](scanner-excess-alerts_zh.md)
+
+Date: 2026-08-18 (rustfs/backlog#1868 / HS-04; includes the HS-15 threshold-delta notes)
+
+The background scanner detects three classes of "excess" conditions while it walks buckets and surfaces them as alerts. This page documents each alert's trigger condition, the subscribable S3 event, the cooldown semantics, and the threshold differences versus MinIO — for operators debugging alerts and for event consumers wiring up subscriptions.
+
+## The three alerts
+
+| Alert | Trigger (per scan cycle) | Metric | S3 event (RustFS wire name) | MinIO event name |
+|---|---|---|---|---|
+| Excess versions | Retained versions of one object ≥ `scanner:alert_excess_versions` | `rustfs_scanner_excess_object_versions_total{bucket}` | `s3:Scanner:ManyVersions` | `s3:ObjectManyVersions` |
+| Excess version size | Cumulative bytes of all versions of one object ≥ `scanner:alert_excess_version_size` | `rustfs_scanner_excess_object_version_size_total{bucket}` | `s3:Scanner:LargeVersions` | `s3:ObjectLargeVersions` |
+| Excess folders | Direct subfolders of one directory > `scanner:alert_excess_folders` | `rustfs_scanner_excess_folders_total{root}` | `s3:Scanner:BigPrefix` | `s3:PrefixManyFolders` |
+
+Subscribe like any bucket notification: configure a notification on the target bucket with the RustFS wire name above (or the `s3:Scanner:*` wildcard). Events carry `UserAgent: Scanner` as their origin marker, and `req_params` holds the observed value and the threshold (`versions` / `cumulativeSize` / `folders` / `threshold`), so consumers can judge severity directly.
+
+## Metrics and events fire on different cadences
+
+- **Metrics and structured logs are level-triggered**: as long as the object stays over the threshold, every scan cycle counts and logs it (default cycle ≈ 60s; see `scanner:speed`).
+- **S3 events are edge-triggered with a cooldown**: the same (alert kind, bucket, object) emits at most once per cooldown window — 24 hours by default (`RUSTFS_SCANNER_ALERT_COOLDOWN_SECS`; set it to 0 to emit every cycle). When the window lapses and the object is still over the threshold, the event fires again. The cooldown table lives in process memory with a 4096-entry hard cap; on overflow it is cleared and rebuilt (worst case: one extra emission per still-hot key).
+- A process restart resets the cooldown (every still-over-threshold object emits once more after a restart) — deliberately: restarts usually accompany incident response, and the re-emission buys visibility.
+
+## Threshold defaults and the MinIO deltas (HS-15)
+
+| Config key | ENV | RustFS default | MinIO default | Notes |
+|---|---|---|---|---|
+| `scanner:alert_excess_versions` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSIONS` | 100 | 100 | Identical |
+| `scanner:alert_excess_version_size` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSION_SIZE` | 1 TiB | 1 TB | Same order of magnitude; different unit basis (TiB vs TB) |
+| `scanner:alert_excess_folders` | `RUSTFS_SCANNER_ALERT_EXCESS_FOLDERS` | 65538 | 50000 | **Deliberate divergence**: 65538 tolerates the Proxmox Backup Server chunk layout (65536 chunks per directory plus the directory's own entries); MinIO's 50000 would fire continuously for PBS users. Set it to 50000 explicitly to match MinIO behavior |
+
+All three keys accept both env and admin config (`PUT /rustfs/admin/v3/config`, `scanner` subsystem); hot updates take effect immediately.
+
+## Why the event names are mapped
+
+RustFS's event enum (`rustfs_s3_types::EventName::ScannerManyVersions/LargeVersions/BigPrefix`) keeps the repo's established `s3:Scanner:*` wire names (literally different from MinIO's `s3:ObjectManyVersions`; the enum comments preserve the mapping). Subscribers should use the RustFS wire names in this page. If you need MinIO-literal compatibility, map the names on the console/consumer side — do not change the published wire names.

docs/operations/scanner-excess-alerts_zh.md

@@ -0,0 +1,37 @@
+# Scanner 超限告警：指标、S3 事件与阈值
+
+> English version: [scanner-excess-alerts.md](scanner-excess-alerts.md)
+
+日期：2026-08-18（rustfs/backlog#1868 / HS-04，含 HS-15 阈值差异说明）
+
+后台 scanner 在扫描过程中检测三类"超限"状态并对外告警。本文说明每类告警的触发条件、可订阅的 S3 事件、冷却语义，以及与 MinIO 的阈值差异，供运维排障与事件消费方对接。
+
+## 三类告警
+
+| 告警 | 触发条件（任一扫描周期） | 指标 | S3 事件（RustFS wire 名） | MinIO 对应事件名 |
+|---|---|---|---|---|
+| 版本数超限 | 单对象保留版本数 ≥ `scanner:alert_excess_versions` | `rustfs_scanner_excess_object_versions_total{bucket}` | `s3:Scanner:ManyVersions` | `s3:ObjectManyVersions` |
+| 版本总大小超限 | 单对象全部版本累计字节 ≥ `scanner:alert_excess_version_size` | `rustfs_scanner_excess_object_version_size_total{bucket}` | `s3:Scanner:LargeVersions` | `s3:ObjectLargeVersions` |
+| 子目录数超限 | 单目录直接子目录数 > `scanner:alert_excess_folders` | `rustfs_scanner_excess_folders_total{root}` | `s3:Scanner:BigPrefix` | `s3:PrefixManyFolders` |
+
+订阅方式与普通桶通知一致：对目标桶配置 notification，事件名填上表 RustFS wire 名（或通配 `s3:Scanner:*`）。事件以 `UserAgent: Scanner` 标记来源，`req_params` 携带实际值与阈值（`versions` / `cumulativeSize` / `folders` / `threshold`），便于消费方直接判断严重程度。
+
+## 指标与事件的触发节奏不同
+
+- **指标与结构化日志是电平触发**：只要对象仍在阈值之上，每个扫描周期都会计数/打日志（默认周期约 60s，见 `scanner:speed`）。
+- **S3 事件是边沿触发 + 冷却**：同一 (告警类型， 桶, 对象) 在冷却窗口内只发一次，默认 24 小时（`RUSTFS_SCANNER_ALERT_COOLDOWN_SECS`，设 0 表示每周期都发）。窗口过后对象仍超限会再次发出。冷却表在进程内有 4096 条硬顶，超限清空重建（最坏情况是每个仍超限的 key 多发一次）。
+- 进程重启会重置冷却（重启后每个仍超限的对象会再发一次）——这是有意为之：重启常伴随排障，重发提供可见性。
+
+## 阈值默认值与 MinIO 差异（HS-15）
+
+| 配置键 | ENV | RustFS 默认 | MinIO 默认 | 差异说明 |
+|---|---|---|---|---|
+| `scanner:alert_excess_versions` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSIONS` | 100 | 100 | 一致 |
+| `scanner:alert_excess_version_size` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSION_SIZE` | 1 TiB | 1 TB | 语义同量级，单位口径不同（TiB vs TB） |
+| `scanner:alert_excess_folders` | `RUSTFS_SCANNER_ALERT_EXCESS_FOLDERS` | 65538 | 50000 | **有意差异**：65538 兼容 Proxmox Backup Server 的 chunk 布局（每目录 65536 个 chunk + 目录自身条目），按 MinIO 的 50000 会对 PBS 用户持续误报。如需与 MinIO 行为一致可显式配置为 50000 |
+
+三个键均支持 env 与 admin config（`PUT /rustfs/admin/v3/config` 的 `scanner` 子系统）双通道，热更新即时生效。
+
+## 事件名映射的由来
+
+RustFS 的事件枚举（`rustfs_s3_types::EventName::ScannerManyVersions/LargeVersions/BigPrefix`）沿用仓库既有 wire 名 `s3:Scanner:*`（与 MinIO 的 `s3:ObjectManyVersions` 字面不同，枚举注释中保留了映射关系）。订阅方应以本文的 RustFS wire 名为准；如需 MinIO 字面兼容，请在 console/消费侧做名称映射，不要修改已发布的 wire 名。

rustfs/src/admin/handlers/site_replication.rs

@@ -6023,7 +6023,10 @@ fn edit_generation_wall_clock() -> u64 {
 /// node's clock behind the clock that fed the previous lifetime) mints
 /// below the stale mark and the origin stays fenced — but only until real
 /// time passes the previous lifetime's last allocation, because every later
-/// allocation takes the wall-clock floor again. Bounded by the skew,
+/// allocation takes the wall-clock floor again (and never longer than
+/// [`PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS`]: a regression past the window
+/// leaves the mark implausibly distant and the origin runs unfenced
+/// immediately). Bounded by the skew,
 /// self-healing, and no rollback window beyond the plain counter's: a
 /// delivery applies only at or above the receiver's mark, so the one
 /// cross-lifetime interleaving that can apply stale content — a
@@ -6063,6 +6066,52 @@ fn peer_edit_fence(queries: &HashMap<String, String>) -> Option<(String, u64)> {
     Some((origin.clone(), generation))
 }
 
+/// How far below the recorded high-water mark a delivery may sit and still
+/// be fenced as stale. The distance a GENUINE superseded delivery can trail
+/// its origin's mark is small: retransmissions re-run the sender flow and
+/// mint a fresh generation (the retry queue keys on the bare path and never
+/// replays a fenced URL), so only an in-flight straggler of the losing
+/// fan-out race trails the mark, by delivery latency — minutes at the
+/// outside. A mark further above than this window cannot be explained by
+/// any genuine race, only by a forged fence (the shared service account
+/// lets any peer stamp any origin) or by a persisted clock excursion the
+/// origin has since left behind — and fencing on it would silently drop the
+/// origin's real edits, so the stale check ignores it instead.
+const PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS: u64 = 24 * 60 * 60 * 1_000_000_000;
+
+/// Whether an incoming fence may be honoured, as far as this site can vouch
+/// for it. The sender's identity is unverifiable (shared service account),
+/// so the check runs over what the receiving state knows: the claimed origin
+/// must be a site this state currently replicates with — the same membership
+/// rule the load-time mark pruning applies, so every mark recorded behind
+/// this check is one a reload would keep — and not this site itself, which
+/// never delivers edits to itself. The caller IGNORES an inadmissible fence
+/// rather than failing the request: the delivery applies exactly as an
+/// unstamped (pre-fence) delivery would, no high-water mark is read or
+/// written, and the worst a forged fence achieves is forfeiting an ordering
+/// guarantee its sender was never owed. The generation itself is NOT
+/// bounded here: a genuine origin whose hybrid clock persisted a wall-clock
+/// excursion allocates arbitrarily far in the future, and refusing to
+/// record its marks would strip the ordering fence from exactly the
+/// deliveries that still race — the staleness window on the read side is
+/// what defuses forged marks instead.
+fn peer_edit_fence_is_admissible(state: &SiteReplicationState, local_deployment_id: &str, fence: &(String, u64)) -> bool {
+    let (origin, generation) = fence;
+    if origin != local_deployment_id && state.peers.contains_key(origin) {
+        return true;
+    }
+    warn!(
+        event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+        component = LOG_COMPONENT_ADMIN,
+        subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+        result = "fence_origin_not_a_remote_peer",
+        origin = %origin,
+        generation = *generation,
+        "ignoring inadmissible peer-edit fence"
+    );
+    false
+}
+
 /// True when a strictly newer edit from the same origin site already landed
 /// here. No lock on the sending side can order deliveries issued by two
 /// nodes of that site, so ordering is decided here, on the generation the
@@ -6070,11 +6119,42 @@ fn peer_edit_fence(queries: &HashMap<String, String>) -> Option<(String, u64)> {
 /// stale: one edit legitimately fans out several deliveries under a single
 /// generation (the ILM-expiry edit sends every peer's record), and a replay of
 /// an applied delivery re-applies the same edit idempotently.
+///
+/// A mark more than [`PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS`] above the
+/// delivery is implausible and does NOT fence: the shared service account
+/// means any peer can stamp any origin, so a forged `u64::MAX`-scale mark
+/// would otherwise silently swallow the origin's genuine edits for good.
+/// Bounding the fence by distance instead of by an absolute ceiling keeps
+/// ordering intact wherever the origin's clock actually operates — two
+/// racing deliveries trail each other by seconds whether the hybrid clock
+/// tracks wall time or persists a long-gone excursion far ahead of it —
+/// while a mark no genuine race can explain merely downgrades the origin to
+/// unfenced (pre-fence) delivery instead of dropping its edits. (One genuine
+/// shape does land out here: a plain-counter straggler arriving after its
+/// origin's first hybrid-clock edit. It gets the same downgrade — applied
+/// unfenced — once, at upgrade time; fencing it instead would silence the
+/// mirror case, a hybrid-clock origin downgraded back to the plain counter.)
 fn peer_edit_delivery_is_stale(state: &SiteReplicationState, origin: &str, generation: u64) -> bool {
-    state
-        .applied_edit_generations
-        .get(origin)
-        .is_some_and(|applied| *applied > generation)
+    let Some(applied) = state.applied_edit_generations.get(origin) else {
+        return false;
+    };
+    if *applied <= generation {
+        return false;
+    }
+    if *applied - generation > PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS {
+        warn!(
+            event = EVENT_ADMIN_SITE_REPLICATION_STATE,
+            component = LOG_COMPONENT_ADMIN,
+            subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
+            result = "fence_mark_beyond_staleness_window",
+            origin,
+            generation,
+            applied_mark = *applied,
+            "ignoring implausibly distant peer-edit high-water mark"
+        );
+        return false;
+    }
+    true
 }
 
 fn record_applied_peer_edit_generation(state: &mut SiteReplicationState, origin: &str, generation: u64) {
@@ -10698,6 +10778,11 @@ impl Operation for SRPeerEditHandler {
         let outcome = update_site_replication_state_when_changed(move |state| {
             let mut incoming = incoming;
             let local_peer = local_peer_at_endpoint(commit_endpoint, state);
+            // The fence is self-reported — the shared service account means
+            // the sender cannot be identified — so it is honoured only after
+            // the admissibility check, against the same state it will gate.
+            let commit_fence =
+                commit_fence.filter(|fence| peer_edit_fence_is_admissible(state, &local_peer.deployment_id, fence));
             // Ordering fence: the sending site allocates the generation under
             // its state-object lock, so a delivery that lost the race carries
             // a generation this site has already passed. Applying it would
@@ -13393,6 +13478,15 @@ mod tests {
             handler_block.contains("record_applied_peer_edit_generation(state, origin, *generation);"),
             "SRPeerEditHandler must record the applied generation so later stale deliveries are recognised"
         );
+        // Fence hardening: origin and generation are self-reported by a
+        // caller the shared service account cannot identify, so the handler
+        // must pass the fence through the admissibility check — against the
+        // same state the fence gates, i.e. inside the transaction — before
+        // reading or raising any high-water mark.
+        assert!(
+            handler_block.contains(".filter(|fence| peer_edit_fence_is_admissible(state, &local_peer.deployment_id, fence))"),
+            "SRPeerEditHandler must admit a fence only through peer_edit_fence_is_admissible inside the state transaction"
+        );
         // P1-15 PR2: both halves of the fence and the edit they fence share
         // ONE transaction. Checking the fence against a state read outside the
         // lock would let the check pass on one snapshot and the write land on
@@ -14769,6 +14863,121 @@ mod tests {
         assert!(peer_edit_delivery_is_stale(&state, origin, generation - 1));
     }
 
+    /// A fence is self-reported: every site authenticates peer traffic with
+    /// the same site-replicator credential, so a compromised peer can stamp
+    /// ANY origin with ANY generation. An origin the receiver does not
+    /// replicate with — or the receiver itself — is ignored and plants no
+    /// mark; a mark a compromised peer plants for a CURRENT origin cannot
+    /// silence that origin, because the staleness window refuses to fence on
+    /// a mark implausibly far above the genuine deliveries.
+    #[test]
+    fn forged_peer_edit_fences_cannot_poison_the_high_water_marks() {
+        let mut state = SiteReplicationState {
+            peers: BTreeMap::from([
+                (
+                    "site-local".to_string(),
+                    PeerInfo {
+                        deployment_id: "site-local".to_string(),
+                        ..peer("local", "https://local.example:9000")
+                    },
+                ),
+                (
+                    "site-victim".to_string(),
+                    PeerInfo {
+                        deployment_id: "site-victim".to_string(),
+                        ..peer("victim", "https://victim.example:9000")
+                    },
+                ),
+            ]),
+            ..Default::default()
+        };
+        // An origin outside the current membership is refused outright...
+        let unknown = ("site-unknown".to_string(), 4u64);
+        assert!(!peer_edit_fence_is_admissible(&state, "site-local", &unknown));
+
+        // No site delivers edits to itself: a fence claiming the receiver as
+        // its origin is forged by construction, current peer or not.
+        let own = ("site-local".to_string(), 4u64);
+        assert!(!peer_edit_fence_is_admissible(&state, "site-local", &own));
+
+        // A current remote peer's fence is admitted and works end to end.
+        let genuine = ("site-victim".to_string(), 1u64);
+        assert!(peer_edit_fence_is_admissible(&state, "site-local", &genuine));
+        assert!(!peer_edit_delivery_is_stale(&state, &genuine.0, genuine.1));
+        record_applied_peer_edit_generation(&mut state, &genuine.0, genuine.1);
+        assert_eq!(state.applied_edit_generations.get("site-victim"), Some(&1));
+
+        // A forged u64::MAX-scale mark CAN be recorded — the shared service
+        // account means the receiver cannot tell the stamp was forged — but
+        // it is inert: the victim's genuine hybrid-clock deliveries sit far
+        // more than the staleness window below it, so they keep applying
+        // instead of being silently acked-and-dropped.
+        record_applied_peer_edit_generation(&mut state, "site-victim", u64::MAX);
+        assert!(!peer_edit_delivery_is_stale(&state, "site-victim", edit_generation_wall_clock()));
+    }
+
+    /// The staleness window bounds the fence by DISTANCE from the mark, not
+    /// by an absolute clock ceiling, so ordering must hold wherever the
+    /// origin's hybrid clock actually operates. The regression that matters:
+    /// a temporary wall-clock excursion far in the future is persisted by
+    /// `next_peer_edit_generation` (`max(now, prev + 1)` never comes back
+    /// down), and two later edits g+1 then g can arrive in reverse order —
+    /// g must still be fenced, even though both generations dwarf the
+    /// receiver's clock. Conversely a mark further above a delivery than any
+    /// genuine race can explain must not fence it.
+    #[test]
+    fn peer_edit_fence_orders_a_persisted_future_clock_and_defuses_distant_marks() {
+        let mut state = SiteReplicationState {
+            peers: BTreeMap::from([(
+                "site-origin".to_string(),
+                PeerInfo {
+                    deployment_id: "site-origin".to_string(),
+                    ..peer("origin", "https://origin.example:9000")
+                },
+            )]),
+            ..Default::default()
+        };
+
+        // The origin's clock once jumped ten years ahead; the hybrid clock
+        // keeps allocating from there long after the clock was corrected.
+        let excursion = edit_generation_wall_clock() + 10 * 365 * 24 * 60 * 60 * 1_000_000_000;
+        let fence = ("site-origin".to_string(), excursion + 1);
+        assert!(peer_edit_fence_is_admissible(&state, "site-local", &fence));
+        record_applied_peer_edit_generation(&mut state, &fence.0, fence.1);
+
+        // The reverse delivery of the race: g arrives after g+1 landed.
+        // Without the fence it would commit last and roll g+1 back.
+        assert!(peer_edit_delivery_is_stale(&state, "site-origin", excursion));
+        // Equal generation (same edit's fan-out or a replay) still applies,
+        // as does the next edit.
+        assert!(!peer_edit_delivery_is_stale(&state, "site-origin", excursion + 1));
+        assert!(!peer_edit_delivery_is_stale(&state, "site-origin", excursion + 2));
+
+        // The window's exact boundary: a delivery trailing the mark by the
+        // full window is still fenced; one nanosecond further is not — that
+        // distance is no longer explicable by a genuine race, only by a
+        // forged mark or an excursion the origin has left behind.
+        let mark = fence.1;
+        // A straggler trailing by a concrete hour must still be fenced —
+        // pins the window's real magnitude, not just its symbolic boundary.
+        assert!(peer_edit_delivery_is_stale(&state, "site-origin", mark - 60 * 60 * 1_000_000_000));
+        assert!(peer_edit_delivery_is_stale(
+            &state,
+            "site-origin",
+            mark - PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS
+        ));
+        assert!(!peer_edit_delivery_is_stale(
+            &state,
+            "site-origin",
+            mark - PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS - 1
+        ));
+
+        // A pre-hybrid plain-counter origin trails such a mark by eons: it
+        // is not fenced (the rc.2-era downgrade case), it just runs
+        // unfenced until its counter regime catches up.
+        assert!(!peer_edit_delivery_is_stale(&state, "site-origin", 3));
+    }
+
     /// P1-15 review follow-up: a site that leaves the mesh drops below two
     /// peers, which clears its state object and restarts its generation
     /// counter at zero. A mark left over from its previous membership would