Repository: risingwavelabs/risingwave
File: src/meta/src/hummock/mod.rs (Rust)

Code before:
    (join_handle, shutdown_tx)
}

pub fn start_vacuum_time_travel_metadata_loop(
    hummock_manager: HummockManagerRef,
    interval: Duration,
) -> (JoinHandle<()>, Sender<()>) {
    let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel();
    let join_handle = tokio::spawn(async move {
        let mut min_trigger_interval = tokio::time::interval(interval);
        min_trigger_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        loop {
            tokio::select! {
                // Wait for interval
                _ = min_trigger_interval.tick() => {},
                // Shutdown vacuum
                _ = &mut shutdown_rx => {
                    tracing::info!("Vacuum time travel metadata loop is stopped");
                    return;
                }
            }
            if let Err(err) = hummock_manager.delete_time_travel_metadata().await {
                tracing::warn!(error = %err.as_report(), "Vacuum time travel metadata error");
            }
        }
    });
    (join_handle, shutdown_tx)
}

pub fn start_checkpoint_loop(
    hummock_manager: HummockManagerRef,
    backup_manager: BackupManagerRef,
    interval: Duration,
    min_delta_log_num: u64,
) -> (JoinHandle<()>, Sender<()>) {
    let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel();
    let join_handle = tokio::spawn(async move {
        let mut min_trigger_interval = tokio::time::interval(interval);
        min_trigger_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        loop {
            tokio::select! {
                // Wait for interval
                _ = min_trigger_interval.tick() => {},
                // Shutdown checkpoint
                _ = &mut shutdown_rx => {
                    tracing::info!("Hummock version checkpoint is stopped");
                    return;
                }
            }
            if hummock_manager.is_version_checkpoint_paused()
                || hummock_manager.env.opts.compaction_deterministic_test

Review comment (suggestion):
cc @hzxa21 @zwang28 

If I remember correctly, time-travel causes a surge in data volume, this was also the reason we controlled the retention period for time-travel. If snapshot-backfill runs for a long time, the increase in data volume will also be significant.

Code after:
            }
        }
    });
    (join_handle, shutdown_tx)
}

pub fn start_vacuum_time_travel_metadata_loop(
    hummock_manager: HummockManagerRef,
    interval: Duration,
    should_pause_vacuum_time_travel: Box<dyn Fn() -> BoxFuture<'static, bool> + Send>,
) -> (JoinHandle<()>, Sender<()>) {
    let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel();
    let join_handle = tokio::spawn(async move {
        let mut min_trigger_interval = tokio::time::interval(interval);
        min_trigger_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        loop {
            tokio::select! {
                // Wait for interval
                _ = min_trigger_interval.tick() => {},
                // Shutdown vacuum
                _ = &mut shutdown_rx => {
                    tracing::info!("Vacuum time travel metadata loop is stopped");
                    return;
                }
            }
            if should_pause_vacuum_time_travel().await {
                tracing::warn!("time travel vacuum paused");
                continue;
            }
            if let Err(err) = hummock_manager.delete_time_travel_metadata().await {
                tracing::warn!(error = %err.as_report(), "Vacuum time travel metadata error");
            }
        }
    });
    (join_handle, shutdown_tx)
}

pub fn start_checkpoint_loop(
    hummock_manager: HummockManagerRef,
    backup_manager: BackupManagerRef,
    interval: Duration,
    min_delta_log_num: u64,
) -> (JoinHandle<()>, Sender<()>) {
    let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel();
    let join_handle = tokio::spawn(async move {
        let mut min_trigger_interval = tokio::time::interval(interval);
        min_trigger_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        loop {
            tokio::select! {
                // Wait for interval
                _ = min_trigger_interval.tick() => {},