From 2cb55bb6b5fbb02420bb07db2b240aa42a30fc61 Mon Sep 17 00:00:00 2001
From: Hengfei Yang <hengfei.yang@gmail.com>
Date: Mon, 2 Jun 2025 16:01:43 +0800
Subject: [PATCH] fix: reset stream stats with limit (#6995)

---
 src/cli/basic/cli.rs         | 17 +++--------------
 src/service/compact/stats.rs | 14 ++++++++++++++
 2 files changed, 17 insertions(+), 14 deletions(-)

diff --git a/src/cli/basic/cli.rs b/src/cli/basic/cli.rs
index 51e10b1309..ccf66ac6a4 100644
--- a/src/cli/basic/cli.rs
+++ b/src/cli/basic/cli.rs
@@ -327,20 +327,9 @@ pub async fn cli() -> Result<bool, anyhow::Error> {
                     // load stream list
                     db::schema::cache().await?;
                     // update stats from file list
-                    loop {
-                        let ret = compact::stats::update_stats_from_file_list()
-                            .await
-                            .expect("file list remote calculate stats failed");
-                        let Some(offset) = ret else {
-                            break;
-                        };
-                        log::info!(
-                            "keep updating stats from file list, offset: {:?} ...",
-                            offset
-                        );
-                        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
-                    }
-                    log::info!("update stats from file list success");
+                    compact::stats::update_stats_from_file_list()
+                        .await
+                        .expect("file list remote calculate stats failed");
                 }
                 _ => {
                     return Err(anyhow::anyhow!(
diff --git a/src/service/compact/stats.rs b/src/service/compact/stats.rs
index d5434b5ff2..e60f9c1083 100644
--- a/src/service/compact/stats.rs
+++ b/src/service/compact/stats.rs
@@ -23,6 +23,20 @@ use crate::{
 };
 
 pub async fn update_stats_from_file_list() -> Result<Option<(i64, i64)>, anyhow::Error> {
+    loop {
+        let Some(offset) = update_stats_from_file_list_inner().await? else {
+            break;
+        };
+        log::info!(
+            "keep updating stream stats from file list, offset: {:?} ...",
+            offset
+        );
+        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
+    }
+    Ok(None)
+}
+
+async fn update_stats_from_file_list_inner() -> Result<Option<(i64, i64)>, anyhow::Error> {
     // get last offset
     let (mut offset, node) = db::compact::stats::get_offset().await;
     if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() {