rpki/src/rtr/runtime.rs

use std::path::PathBuf;
use std::time::Instant;

use anyhow::Result;
use chrono::Utc;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tracing::{info, warn};

use crate::rtr::admin::{RuntimeConfigHandle, SourceReloadCommand, SourceReloadResult};
use crate::rtr::cache::{SharedRtrCache, Snapshot};
use crate::rtr::config::{AppConfig, RuntimeConfig};
use crate::rtr::report::{ReportContext, current_rss_mib};
use crate::rtr::server::{RtrNotifier, RtrServiceStats};
use crate::rtr::store::RtrStore;
use crate::source::pipeline::{
    PayloadLoadConfig, SourceFingerprint, latest_sources_fingerprint,
    load_payloads_from_latest_sources_with_report,
};

pub fn spawn_refresh_task(
    config: &AppConfig,
    runtime_config: RuntimeConfigHandle,
    mut source_reload_rx: mpsc::Receiver<SourceReloadCommand>,
    shared_cache: SharedRtrCache,
    store: RtrStore,
    notifier: RtrNotifier,
    service_stats: RtrServiceStats,
    report_context: ReportContext,
) -> JoinHandle<()> {
    let report_dir = PathBuf::from(&config.report_dir);
    let payload_load_config = PayloadLoadConfig {
        ccr_dir: config.ccr_dir.clone(),
        slurm_dir: config.slurm_dir.clone(),
    };
    let initial_runtime_config = runtime_config.current();

    tokio::spawn(async move {
        let mut active_runtime_config = initial_runtime_config;
        let mut config_rx = runtime_config.subscribe();
        let mut interval = tokio::time::interval(std::time::Duration::from_secs(
            active_runtime_config.source_refresh_interval_seconds,
        ));
        let mut last_fingerprint: Option<SourceFingerprint> = None;
        report_context.write_source_or_warn(
            &report_dir,
            "startup",
            &shared_cache,
            &notifier,
            &service_stats,
        );
        report_context.write_clients_or_warn(
            &report_dir,
            "startup",
            &shared_cache,
            &notifier,
            &service_stats,
        );
        report_context.write_runtime_or_warn(
            &report_dir,
            "startup",
            &shared_cache,
            &notifier,
            &service_stats,
        );
        let mut runtime_interval = tokio::time::interval_at(
            tokio::time::Instant::now()
                + std::time::Duration::from_secs(
                    active_runtime_config.runtime_report_interval_seconds,
                ),
            std::time::Duration::from_secs(active_runtime_config.runtime_report_interval_seconds),
        );
        runtime_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
        let mut client_change_rx = service_stats.subscribe_connection_changes();

        loop {
            tokio::select! {
                changed = config_rx.changed() => {
                    match changed {
                        Ok(()) => {
                            let next_config = config_rx.borrow().clone();
                            apply_runtime_config_update(
                                &mut active_runtime_config,
                                next_config,
                                &shared_cache,
                                &store,
                                &report_context,
                                &mut interval,
                                &mut runtime_interval,
                            );
                            report_context.write_runtime_or_warn(&report_dir, "admin_config_changed", &shared_cache, &notifier, &service_stats);
                        }
                        Err(_) => {
                            warn!("RTR runtime config change channel closed");
                        }
                    }
                    continue;
                }
                changed = client_change_rx.changed() => {
                    match changed {
                        Ok(()) => {
                            report_context.write_clients_or_warn(&report_dir, "clients_changed", &shared_cache, &notifier, &service_stats);
                        }
                        Err(_) => {
                            warn!("RTR client connection change channel closed");
                        }
                    }
                    continue;
                }
                _ = runtime_interval.tick() => {
                    log_cache_memory_stats("periodic_observe", &shared_cache, &notifier);
                    report_context.write_runtime_or_warn(&report_dir, "runtime_periodic", &shared_cache, &notifier, &service_stats);
                    continue;
                }
                _ = interval.tick() => {}
                command = source_reload_rx.recv() => {
                    let Some(command) = command else {
                        warn!("RTR source reload admin channel closed");
                        continue;
                    };
                    let result = perform_source_refresh(
                        command.phase,
                        command.force,
                        &payload_load_config,
                        &shared_cache,
                        &store,
                        &notifier,
                        &service_stats,
                        &report_context,
                        &report_dir,
                        &mut last_fingerprint,
                    );
                    let _ = command.respond_to.send(result.map_err(|err| err.to_string()));
                    continue;
                }
            }
            let _ = perform_source_refresh(
                "refresh_complete",
                false,
                &payload_load_config,
                &shared_cache,
                &store,
                &notifier,
                &service_stats,
                &report_context,
                &report_dir,
                &mut last_fingerprint,
            );
        }
    })
}

#[allow(clippy::too_many_arguments)]
fn perform_source_refresh(
    phase: &'static str,
    force: bool,
    payload_load_config: &PayloadLoadConfig,
    shared_cache: &SharedRtrCache,
    store: &RtrStore,
    notifier: &RtrNotifier,
    service_stats: &RtrServiceStats,
    report_context: &ReportContext,
    report_dir: &PathBuf,
    last_fingerprint: &mut Option<SourceFingerprint>,
) -> Result<SourceReloadResult> {
    let source_to_delta_started = Instant::now();
    let attempted_at = Utc::now();

    let current_fingerprint = match latest_sources_fingerprint(payload_load_config) {
        Ok(fp) => fp,
        Err(err) => {
            report_context.record_refresh_failure(
                attempted_at,
                source_to_delta_started.elapsed().as_millis(),
                &err,
            );
            warn!(
                "failed to fingerprint CCR/SLURM sources from {}: {:?} (source_to_delta_elapsed_ms={})",
                payload_load_config.ccr_dir,
                err,
                source_to_delta_started.elapsed().as_millis()
            );
            report_context.write_source_or_warn(
                report_dir,
                "refresh_failed",
                shared_cache,
                notifier,
                service_stats,
            );
            return Err(err);
        }
    };
    report_context.record_source_fingerprint(current_fingerprint.clone());

    if !force && last_fingerprint.as_ref() == Some(&current_fingerprint) {
        report_context
            .record_refresh_unchanged(attempted_at, source_to_delta_started.elapsed().as_millis());
        info!(
            "RTR source refresh skipped: source files unchanged (ccr_path={}, slurm_file_count={}, elapsed_ms={})",
            current_fingerprint.ccr.path,
            current_fingerprint.slurm_files.len(),
            source_to_delta_started.elapsed().as_millis()
        );
        log_cache_memory_stats("refresh_skipped_unchanged", shared_cache, notifier);
        report_context.write_source_or_warn(
            report_dir,
            "refresh_skipped_unchanged",
            shared_cache,
            notifier,
            service_stats,
        );
        return Ok(SourceReloadResult {
            phase,
            changed: false,
            skipped_unchanged: true,
            payload_count: None,
            serials: shared_cache.load_full().serials(),
        });
    }

    let load = match load_payloads_from_latest_sources_with_report(payload_load_config) {
        Ok(load) => load,
        Err(err) => {
            report_context.record_refresh_failure(
                attempted_at,
                source_to_delta_started.elapsed().as_millis(),
                &err,
            );
            warn!(
                "failed to reload CCR/SLURM payloads from {}: {:?} (source_to_delta_elapsed_ms={})",
                payload_load_config.ccr_dir,
                err,
                source_to_delta_started.elapsed().as_millis()
            );
            report_context.write_source_or_warn(
                report_dir,
                "refresh_failed",
                shared_cache,
                notifier,
                service_stats,
            );
            return Err(err);
        }
    };

    let source = load.source;
    let quality = load.quality;
    let payloads = load.payloads;
    let payload_count = payloads.len();
    let source_snapshot = Snapshot::from_payloads(payloads);
    let old_cache = shared_cache.load_full();
    let old_serial = old_cache.serial_for_version(2);
    let mut next_cache = old_cache.as_ref().clone();
    let updated = match next_cache.update_with_snapshot(source_snapshot, store) {
        Ok(()) => {
            let new_serial = next_cache.serial_for_version(2);
            shared_cache.store(std::sync::Arc::new(next_cache));
            if new_serial != old_serial {
                info!(
                    "RTR cache refresh applied: ccr_dir={}, payload_count={}, old_serial={}, new_serial={}",
                    payload_load_config.ccr_dir, payload_count, old_serial, new_serial
                );
                true
            } else {
                info!(
                    "RTR cache refresh found no change: ccr_dir={}, payload_count={}, serial={}",
                    payload_load_config.ccr_dir, payload_count, old_serial
                );
                false
            }
        }
        Err(err) => {
            report_context.record_refresh_failure(
                attempted_at,
                source_to_delta_started.elapsed().as_millis(),
                &err,
            );
            warn!("RTR cache update failed: {:?}", err);
            report_context.write_source_or_warn(
                report_dir,
                "refresh_failed",
                shared_cache,
                notifier,
                service_stats,
            );
            return Err(err);
        }
    };
    report_context.record_refresh_success(
        attempted_at,
        source_to_delta_started.elapsed().as_millis(),
        updated,
        source,
        quality,
    );
    info!(
        "RTR source-to-delta timing: phase={}, ccr_dir={}, payload_count={}, changed={}, elapsed_ms={}",
        phase,
        payload_load_config.ccr_dir,
        payload_count,
        updated,
        source_to_delta_started.elapsed().as_millis()
    );

    if updated {
        let listener_count = notifier.notify_cache_updated();
        info!(
            "RTR cache updated, notify signal emitted to session listeners: listener_count={}",
            listener_count
        );
    }
    log_cache_memory_stats(phase, shared_cache, notifier);
    report_context.write_source_or_warn(report_dir, phase, shared_cache, notifier, service_stats);
    last_fingerprint.replace(current_fingerprint);

    Ok(SourceReloadResult {
        phase,
        changed: updated,
        skipped_unchanged: false,
        payload_count: Some(payload_count),
        serials: shared_cache.load_full().serials(),
    })
}

fn apply_runtime_config_update(
    active: &mut RuntimeConfig,
    next: RuntimeConfig,
    shared_cache: &SharedRtrCache,
    store: &RtrStore,
    report_context: &ReportContext,
    refresh_interval: &mut tokio::time::Interval,
    runtime_interval: &mut tokio::time::Interval,
) {
    let old = active.clone();
    *active = next.clone();
    report_context.update_runtime_config(&next);

    {
        let old_cache = shared_cache.load_full();
        let mut next_cache = old_cache.as_ref().clone();
        next_cache.update_runtime_config(
            next.max_delta,
            next.prune_delta_by_snapshot_size,
            next.timing(),
            store,
        );
        shared_cache.store(std::sync::Arc::new(next_cache));
    }

    if old.source_refresh_interval_seconds != next.source_refresh_interval_seconds {
        *refresh_interval = tokio::time::interval(std::time::Duration::from_secs(
            next.source_refresh_interval_seconds,
        ));
        refresh_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    }
    if old.runtime_report_interval_seconds != next.runtime_report_interval_seconds {
        *runtime_interval = tokio::time::interval_at(
            tokio::time::Instant::now()
                + std::time::Duration::from_secs(next.runtime_report_interval_seconds),
            std::time::Duration::from_secs(next.runtime_report_interval_seconds),
        );
        runtime_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    }

    info!(
        "RTR runtime config applied: max_delta={}, prune_delta_by_snapshot_size={}, source_refresh_interval_seconds={}, runtime_report_interval_seconds={}, report_history_limit={}, timezone={}, timing=({}, {}, {})",
        next.max_delta,
        next.prune_delta_by_snapshot_size,
        next.source_refresh_interval_seconds,
        next.runtime_report_interval_seconds,
        next.report_history_limit,
        next.timezone,
        next.timing.refresh,
        next.timing.retry,
        next.timing.expire
    );
}

fn log_cache_memory_stats(phase: &str, shared_cache: &SharedRtrCache, notifier: &RtrNotifier) {
    let cache = shared_cache.load_full();
    let stats = cache.memory_stats();
    let rss_mib = current_rss_mib();

    info!(
        "RTR memory observe: phase={}, listener_count={}, serials={:?}, snapshot_payload_counts={:?}, delta_lengths={:?}, delta_payload_counts={:?}, rss_mib={:?}",
        phase,
        notifier.listener_count(),
        stats.serials,
        stats.snapshot_payload_counts,
        stats.delta_lengths,
        stats.delta_payload_counts,
        rss_mib
    );
}