Function spawn_cold_tier_task

nodedb/src/control/cold_tier.rs:28–70

Spawn the cold tiering background task. Scans `data_dir/segments/` every `tier_check_interval_secs` seconds and uploads segment files older than `tier_after_secs` to cold storage. The local file is removed after a successful upload. The task exits cleanly when `shutdown_rx` is set to `true`.

(
    shared: Arc<SharedState>,
    settings: ColdStorageSettings,
    data_dir: PathBuf,
    mut shutdown_rx: tokio::sync::watch::Receiver<bool>,
)

///
/// The task exits cleanly when `shutdown_rx` is set to `true`.
pub fn spawn_cold_tier_task(
    shared: Arc<SharedState>,
    settings: ColdStorageSettings,
    data_dir: PathBuf,
    mut shutdown_rx: tokio::sync::watch::Receiver<bool>,
) -> tokio::task::JoinHandle<()> {
    let check_interval = Duration::from_secs(settings.tier_check_interval_secs);
    let tier_after = Duration::from_secs(settings.tier_after_secs);
    let prefix = settings.prefix.clone();
    let segments_dir = data_dir.join("segments");

    tokio::spawn(async move {
        info!(
            check_interval_secs = settings.tier_check_interval_secs,
            tier_after_secs = settings.tier_after_secs,
            segments_dir = %segments_dir.display(),
            "cold tier task started"
        );

        loop {
            tokio::select! {
                _ = tokio::time::sleep(check_interval) => {}
                _ = shutdown_rx.changed() => {
                    if *shutdown_rx.borrow() {
                        info!("cold tier task stopping on shutdown signal");
                        return;
                    }
                }
            }

            let cold = match shared.cold_storage.as_ref() {
                Some(c) => Arc::clone(c),
                None => {
                    // Cold storage was removed from shared state; stop the task.
                    warn!("cold tier: cold_storage is None, stopping task");
                    return;
                }
            };

            run_tier_cycle_at(&cold, &segments_dir, tier_after, &prefix).await;
        }
    })
}

/// Run one tiering cycle against the given segments directory.
///