fix: reset stream stats with limit (openobserve#6995)

Author: hengfeiyang

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!(

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() {