fix(scanner): reject duplicate usage updates (#6445)

* fix(scanner): reject duplicate usage updates

* style: format decommission test imports

* fix(ci): remove unused decommission and healing facades

* fix(ci): cfg-gate test-only usage overlay import

---------

Co-authored-by: houseme <housemecn@gmail.com>

crates/scanner/src/scanner_io/cache.rs

@@ -751,16 +751,29 @@ pub(super) async fn persist_and_publish_cache_snapshot(
 }
 
 pub(super) async fn send_data_usage_update(updates: &mpsc::Sender<DataUsageInfo>, data_usage_info: DataUsageInfo) -> Result<()> {
-    updates.send(data_usage_info).await.map_err(|e| {
+    match updates.try_send(data_usage_info) {
-        error!(
+        Ok(()) => Ok(()),
-            target: "rustfs::scanner::io",
+        Err(mpsc::error::TrySendError::Closed(_)) => {
-            event = EVENT_SCANNER_DATA_USAGE_STREAM,
+            error!(
-            component = LOG_COMPONENT_SCANNER,
+                target: "rustfs::scanner::io",
-            subsystem = LOG_SUBSYSTEM_IO,
+                event = EVENT_SCANNER_DATA_USAGE_STREAM,
-            state = "send_failed",
+                component = LOG_COMPONENT_SCANNER,
-            error = %e,
+                subsystem = LOG_SUBSYSTEM_IO,
-            "Scanner data usage publish failed"
+                state = "send_failed",
-        );
+                "Scanner data usage publish failed because the receiver is closed"
-        StorageError::other("scanner data usage receiver closed before update delivery")
+            );
-    })
+            Err(StorageError::other("scanner data usage receiver closed before update delivery"))
+        }
+        Err(mpsc::error::TrySendError::Full(_)) => {
+            error!(
+                target: "rustfs::scanner::io",
+                event = EVENT_SCANNER_DATA_USAGE_STREAM,
+                component = LOG_COMPONENT_SCANNER,
+                subsystem = LOG_SUBSYSTEM_IO,
+                state = "send_would_block",
+                "Scanner data usage publish rejected because an update is already queued"
+            );
+            Err(StorageError::other("scanner data usage update already queued"))
+        }
+    }
 }

crates/scanner/src/scanner_io/tests.rs

@@ -268,6 +268,36 @@ async fn data_usage_publish_fails_when_receiver_is_closed() {
     assert!(err.to_string().contains("receiver closed"));
 }
 
+#[tokio::test]
+async fn data_usage_publish_rejects_a_second_terminal_update_without_blocking() {
+    for (status, data_usage_info) in [
+        (ScannerCycleStatus::Complete, DataUsageInfo::default()),
+        (
+            ScannerCycleStatus::Superseded,
+            DataUsageInfo {
+                scanner_cycle: Some(7),
+                ..Default::default()
+            },
+        ),
+    ] {
+        let (updates, mut receiver) = mpsc::channel(1);
+        assert!(
+            publish_usage_snapshot(&updates, status, data_usage_info)
+                .await
+                .expect("first terminal update should be accepted")
+        );
+
+        let err =
+            tokio::time::timeout(Duration::from_secs(1), publish_usage_snapshot(&updates, status, DataUsageInfo::default()))
+                .await
+                .expect("a full terminal update must fail without waiting")
+                .expect_err("a second terminal update must be rejected");
+        assert!(err.to_string().contains("already queued"));
+        assert!(receiver.try_recv().is_ok(), "the first terminal update must remain owned by the receiver");
+        assert!(receiver.try_recv().is_err(), "the rejected second update must not enter the channel");
+    }
+}
+
 #[tokio::test]
 async fn multi_pool_scanner_cycle_publishes_combined_usage() {
     let (_temp_dir, store) = setup_two_pool_scanner_store().await;