diff --git a/src/compaction/src/stats.rs b/src/compaction/src/stats.rs
index 72d878ed94..f2c912d1c6 100644
--- a/src/compaction/src/stats.rs
+++ b/src/compaction/src/stats.rs
@@ -13,7 +13,7 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see .
-use std::collections::HashSet;
+use std::{collections::HashSet, time::Duration};
use config::{
cluster::LOCAL_NODE,
@@ -25,6 +25,9 @@ use config::{
use db;
use infra::{cluster::get_node_by_uuid, dist_lock, file_list as infra_file_list};
+const STATS_SCAN_MAX_ATTEMPTS: u32 = 3;
+const STATS_SCAN_RETRY_DELAY: Duration = Duration::from_secs(2);
+
pub async fn update_stats_from_file_list() -> Result<(), anyhow::Error> {
let latest_updated_at = infra_file_list::get_max_update_at()
.await
@@ -89,7 +92,6 @@ pub async fn update_stats_from_file_list() -> Result<(), anyhow::Error> {
continue;
}
total_streams += streams.len();
- let stream_type_str = stream_type.to_string();
for stream_name in streams {
let stream_key = format!("{org_id}/{stream_type}/{stream_name}");
if !updated_streams.is_empty() && !updated_streams.contains(&stream_key) {
@@ -101,37 +103,14 @@ pub async fn update_stats_from_file_list() -> Result<(), anyhow::Error> {
if !is_recent && no_need_update_old_stats {
continue;
}
- let start = std::time::Instant::now();
- let result = update_stats_from_file_list_for_stream(
+ update_stream_stats_with_retry(
&org_id,
stream_type,
&stream_name,
date_range.clone(),
*is_recent,
)
- .await;
-
- // Record metrics
- let duration = start.elapsed().as_secs_f64();
- let scan_type = if *is_recent { "recent" } else { "historical" }.to_string();
- metrics::STREAM_STATS_SCAN_DURATION
- .with_label_values(&[&org_id, &stream_type_str, &scan_type])
- .observe(duration);
-
- metrics::STREAM_STATS_SCAN_TOTAL
- .with_label_values(&[&org_id, &stream_type_str, &scan_type])
- .inc();
-
- if let Err(e) = result {
- metrics::STREAM_STATS_SCAN_ERRORS_TOTAL
- .with_label_values(&[&org_id, &stream_type_str, &scan_type])
- .inc();
-
- log::error!(
- "[STATS] update stats for {org_id}/{stream_type}/{stream_name} error: {e}"
- );
- return Err(e);
- }
+ .await?;
}
log::info!(
@@ -208,6 +187,55 @@ async fn update_stats_lock_node() -> Result