reductstore 1.19.8

// Copyright 2021-2026 ReductSoftware UG
// Licensed under the Apache License, Version 2.0

use crate::api::http::AxumAppBuilder;
#[cfg(feature = "zenoh-api")]
use crate::api::zenoh;
use crate::cfg::{CfgParser, ExtCfgBounds, ExtCfgParser, InstanceRole};
use crate::core::env::StdEnvGetter;
use crate::core::sync::set_rwlock_timeout;
use crate::storage::engine::StorageEngine;
use axum_server::tls_rustls::RustlsConfig;
use axum_server::Handle;
use log::{error, info, warn};
use reduct_base::logger::Logger;
use std::net::{IpAddr, SocketAddr};
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;

static SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
#[cfg(test)]
static RW_LOCK_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);

#[cfg(not(test))]
static RW_LOCK_SHUTDOWN_TIMEOUT: Duration = Duration::from_hours(1);

pub async fn launch_server<Parser, ExtCfg: ExtCfgBounds + 'static>(ext_cfg_pareser: Parser)
where
    Parser: ExtCfgParser<StdEnvGetter, Cfg = ExtCfg>,
{
    let version: &str = env!("CARGO_PKG_VERSION");

    Logger::init("INFO");
    info!(
        "ReductStore Core {} [{} at {}]",
        version,
        env!("COMMIT"),
        env!("BUILD_TIME")
    );

    let parser =
        CfgParser::from_env_with_ext(StdEnvGetter::default(), &ext_cfg_pareser, version).await;
    let handle = Handle::new();
    let lock_file = Arc::new(parser.build_lock_file().unwrap());

    // Run initialization in a separate thread to avoid blocking HTTP server startup
    // if waiting for the lock file.
    let config_lock = Arc::clone(&lock_file);
    let signal_handle = handle.clone();
    let cfg = parser.cfg.clone();
    let engine_config = cfg.engine_config.clone();
    let instance_role = cfg.role.clone();
    let (tx, rx) = mpsc::channel(1);
    tokio::spawn(async move {
        while config_lock.is_waiting().await.unwrap_or(false) {
            tokio::time::sleep(Duration::from_millis(100)).await;
        }

        if config_lock.is_failed().await.unwrap_or(true) {
            panic!("Another ReductStore instance is holding the lock. Exiting.");
        }

        let components = parser.build().await.unwrap();

        if !engine_config.compaction_interval.is_zero() {
            tokio::spawn(periodical_compact_storage(
                components.storage.clone(),
                engine_config.compaction_interval,
            ));
        }

        if instance_role == InstanceRole::Replica
            && !engine_config.replica_update_interval.is_zero()
        {
            tokio::spawn(periodical_replica_reload(
                components.storage.clone(),
                engine_config.replica_update_interval,
            ));
        }

        tokio::spawn(shutdown_ctrl_c(signal_handle.clone()));
        #[cfg(unix)]
        tokio::spawn(shutdown_signal(signal_handle.clone()));
        #[cfg(test)]
        tokio::spawn(tests::shutdown_server(signal_handle.clone()));

        tx.send(components).await.unwrap();
    });

    info!("Public URL: {}", cfg.public_url);

    let addr = SocketAddr::new(
        IpAddr::from_str(&cfg.host).expect("Invalid host address"),
        cfg.port,
    );

    let (app, state_keeper) = AxumAppBuilder::new()
        .with_cfg(cfg.clone())
        .with_component_receiver(rx)
        .with_lock_file(lock_file.clone())
        .build();

    // Spawn Zenoh API runtime if enabled
    #[cfg(feature = "zenoh-api")]
    let zenoh_runtime = zenoh::spawn_runtime(cfg.zenoh_api.clone(), state_keeper.clone());

    #[cfg(not(test))]
    {
        // Ensure that the process exits with a non-zero exit code on panic.
        let default_panic = std::panic::take_hook();
        std::panic::set_hook(Box::new(move |info| {
            default_panic(info);
            std::process::exit(1);
        }));
    }

    macro_rules! apply_http_settings {
        ($server:expr) => {{
            let mut server = $server.handle(handle);
            server
                .http_builder()
                .http1()
                .max_headers(cfg.io_conf.batch_max_records + 15);
            server
                .http_builder()
                .http1()
                .max_buf_size(cfg.io_conf.batch_max_metadata_size);
            server
        }};
    }

    if cfg.cert_path.is_none() {
        apply_http_settings!(axum_server::bind(addr))
            .serve(app.into_make_service_with_connect_info::<SocketAddr>())
            .await
            .unwrap_or_else(|e| error!("Server error: {}", e));
    } else {
        rustls::crypto::aws_lc_rs::default_provider()
            .install_default()
            .expect("Failed to install rustls crypto provider");
        let config = RustlsConfig::from_pem_file(
            cfg.cert_path.expect("Cert path must be set"),
            cfg.cert_key_path.expect("Cert key path must be set"),
        )
        .await
        .expect("Failed to load TLS certificate");
        apply_http_settings!(axum_server::bind_rustls(addr, config))
            .serve(app.into_make_service_with_connect_info::<SocketAddr>())
            .await
            .unwrap_or_else(|e| error!("Server error: {}", e));
    };

    // shutdown procedure
    #[cfg(feature = "zenoh-api")]
    if let Some(handle) = zenoh_runtime {
        handle.shutdown().await;
    }

    // remote synchronization can lock resources for a long time,
    // so we set rwlock timeout to 1 hour to avoid panics during shutdown
    set_rwlock_timeout(RW_LOCK_SHUTDOWN_TIMEOUT);
    state_keeper
        .get_anonymous()
        .await
        .expect("Failed to access storage engine")
        .storage
        .sync_fs()
        .await
        .expect("Failed to shutdown storage");
    drop(lock_file);
    info!("Server has been shut down.");
}

async fn shutdown_ctrl_c(server_handle: Handle<SocketAddr>) {
    tokio::signal::ctrl_c().await.unwrap();
    info!("Received Ctrl-C, shutting down server...");
    server_handle.graceful_shutdown(Some(SHUTDOWN_TIMEOUT));
}

#[cfg(unix)]
async fn shutdown_signal(server_handle: Handle<SocketAddr>) {
    tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
        .unwrap()
        .recv()
        .await;
    info!("Received termination signal, shutting down server...");
    server_handle.graceful_shutdown(Some(SHUTDOWN_TIMEOUT));
}

#[cfg(test)]
mod test_observer {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{LazyLock, Mutex};

    pub static COMPACTION_OBSERVER: LazyLock<Mutex<Option<Arc<AtomicUsize>>>> =
        LazyLock::new(|| Mutex::new(None));
    pub static REPLICA_RELOAD_OBSERVER: LazyLock<Mutex<Option<Arc<AtomicUsize>>>> =
        LazyLock::new(|| Mutex::new(None));

    pub fn set_compaction_observer(observer: Option<Arc<AtomicUsize>>) {
        *COMPACTION_OBSERVER.lock().unwrap() = observer;
    }

    pub fn set_replica_reload_observer(observer: Option<Arc<AtomicUsize>>) {
        *REPLICA_RELOAD_OBSERVER.lock().unwrap() = observer;
    }

    pub fn observe_compaction_tick() {
        if let Some(observer) = COMPACTION_OBSERVER.lock().unwrap().as_ref() {
            observer.fetch_add(1, Ordering::Relaxed);
        }
    }

    pub fn observe_replica_reload_tick() {
        if let Some(observer) = REPLICA_RELOAD_OBSERVER.lock().unwrap().as_ref() {
            observer.fetch_add(1, Ordering::Relaxed);
        }
    }
}

async fn periodical_compact_storage(storage: Arc<StorageEngine>, sync_interval: Duration) {
    run_periodic_task(sync_interval, "compaction", || {
        let storage = storage.clone();
        async move {
            #[cfg(test)]
            test_observer::observe_compaction_tick();

            if let Err(e) = storage.compact().await {
                log::error!("Failed to sync storage: {}", e);
            }
        }
    })
    .await;
}

async fn periodical_replica_reload(storage: Arc<StorageEngine>, sync_interval: Duration) {
    run_periodic_task(sync_interval, "replica reload", || {
        let storage = storage.clone();
        async move {
            #[cfg(test)]
            test_observer::observe_replica_reload_tick();

            if let Err(e) = storage.reload_replica().await {
                log::error!("Failed to reload replica state: {}", e);
            }
        }
    })
    .await;
}

async fn run_periodic_task<F, Fut>(interval: Duration, task_name: &'static str, mut task: F)
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = ()>,
{
    let mut next_tick = tokio::time::Instant::now() + interval;

    loop {
        tokio::time::sleep_until(next_tick).await;
        let started_at = std::time::Instant::now();

        task().await;

        let execution_time = started_at.elapsed();
        if execution_time > interval {
            warn!(
                "Periodic {} took {:?}, exceeding configured interval {:?}",
                task_name, execution_time, interval
            );
        }

        next_tick += interval;
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cfg::storage_engine::StorageEngineConfig;
    use crate::cfg::Cfg;
    use crate::cfg::CoreExtCfgParser;
    use log::warn;
    use reduct_base::msg::bucket_api::BucketSettings;
    use rstest::rstest;
    use serial_test::serial;
    use std::collections::HashMap;
    use std::env;

    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{Arc, LazyLock};
    use std::thread::{spawn, JoinHandle};
    use tempfile::tempdir;
    use tokio::sync::Mutex;
    use tokio::time::sleep;

    static STOP_SERVER: LazyLock<Mutex<bool>> = LazyLock::new(|| Mutex::new(false));
    pub(super) async fn shutdown_server(handle: Handle<SocketAddr>) {
        while !*STOP_SERVER.lock().await {
            sleep(Duration::from_millis(10)).await;
        }
        warn!("Shutting down server");
        handle.shutdown();
    }

    #[rstest]
    #[tokio::test(flavor = "multi_thread")]
    #[serial]
    async fn test_launch_http() {
        let task = set_env_and_run(HashMap::new()).await;

        reqwest::get("http://127.0.0.1:8383/api/v1/info")
            .await
            .expect("Failed to get info")
            .error_for_status()
            .expect("Failed to get info");

        // send shutdown signal
        *STOP_SERVER.lock().await = true;
        task.join().unwrap();
    }

    #[rstest]
    #[tokio::test(flavor = "multi_thread")]
    #[serial]
    async fn test_launch_https() {
        let cert_path = resolve_misc_file("certificate.crt");
        let cert_key_path = resolve_misc_file("privateKey.key");
        let mut cfg = HashMap::new();
        cfg.insert(
            "RS_CERT_PATH".to_string(),
            cert_path.to_string_lossy().to_string(),
        );
        cfg.insert(
            "RS_CERT_KEY_PATH".to_string(),
            cert_key_path.to_string_lossy().to_string(),
        );

        let task = set_env_and_run(cfg).await;
        let client = reqwest::Client::builder()
            .danger_accept_invalid_certs(true)
            .build()
            .unwrap();

        client
            .get("https://127.0.0.1:8383/api/v1/info")
            .send()
            .await
            .expect("Failed to get info")
            .error_for_status()
            .expect("Failed to get info");

        // send shutdown signal
        *STOP_SERVER.lock().await = true;
        task.join().unwrap();
    }

    #[rstest]
    #[tokio::test(flavor = "multi_thread")]
    #[serial]
    async fn test_compaction_task_runs_when_interval_non_zero() {
        let compactions = Arc::new(AtomicUsize::new(0));
        test_observer::set_compaction_observer(Some(compactions.clone()));

        let data_path = tempdir().unwrap().keep();
        env::set_var("RS_DATA_PATH", data_path.to_str().unwrap());
        let parser = CfgParser::from_env(StdEnvGetter::default(), "0.0.0").await;
        let storage = parser.build().await.unwrap().storage;

        let handler = tokio::spawn(periodical_compact_storage(
            storage,
            Duration::from_millis(50),
        ));

        sleep(Duration::from_millis(120)).await;
        handler.abort();
        let _ = handler.await;

        test_observer::set_compaction_observer(None);

        assert!(
            compactions.load(Ordering::Relaxed) > 0,
            "periodical_compact_storage should run when interval is non-zero"
        );
    }

    #[rstest]
    #[tokio::test(flavor = "multi_thread")]
    #[serial]
    async fn test_replica_reload_task_runs_when_interval_non_zero() {
        let reloads = Arc::new(AtomicUsize::new(0));
        test_observer::set_replica_reload_observer(Some(reloads.clone()));

        let data_path = tempdir().unwrap().keep();
        let cfg = Cfg {
            data_path: data_path.clone(),
            role: InstanceRole::Primary,
            engine_config: StorageEngineConfig {
                replica_update_interval: Duration::from_millis(50),
                ..StorageEngineConfig::default()
            },
            ..Cfg::default()
        };
        let primary_storage = StorageEngine::builder()
            .with_cfg(cfg.clone())
            .with_data_path(cfg.data_path.clone())
            .build()
            .await;
        primary_storage
            .create_bucket("bucket-1", BucketSettings::default())
            .await
            .unwrap();

        let mut replica_cfg = cfg.clone();
        replica_cfg.role = InstanceRole::Replica;
        let replica_storage = Arc::new(
            StorageEngine::builder()
                .with_cfg(replica_cfg.clone())
                .with_data_path(replica_cfg.data_path.clone())
                .build()
                .await,
        );
        primary_storage
            .create_bucket("bucket-2", BucketSettings::default())
            .await
            .unwrap();

        let handler = tokio::spawn(periodical_replica_reload(
            replica_storage.clone(),
            Duration::from_millis(50),
        ));

        sleep(Duration::from_millis(120)).await;
        handler.abort();
        let _ = handler.await;

        test_observer::set_replica_reload_observer(None);

        assert!(
            reloads.load(Ordering::Relaxed) > 0,
            "periodical_replica_reload should run when interval is non-zero"
        );
        let bucket_names = replica_storage
            .bucket_list_snapshot()
            .await
            .unwrap()
            .into_iter()
            .map(|bucket| bucket.name().to_string())
            .collect::<Vec<_>>();
        assert!(bucket_names.contains(&"bucket-1".to_string()));
        assert!(bucket_names.contains(&"bucket-2".to_string()));
    }

    async fn set_env_and_run(cfg: HashMap<String, String>) -> JoinHandle<()> {
        let data_path = tempdir().unwrap().keep();

        env::set_var("RS_DATA_PATH", data_path.to_str().unwrap());
        env::set_var("RS_CERT_PATH", "");
        env::set_var("RS_CERT_KEY_PATH", "");
        env::set_var("RS_INSTANCE_ROLE", "STANDALONE");
        env::set_var("RS_ENGINE_REPLICA_UPDATE_INTERVAL", "60");

        for (key, value) in cfg {
            env::set_var(key, value);
        }

        let task = spawn(|| {
            tokio::runtime::Runtime::new().unwrap().block_on(async {
                *STOP_SERVER.lock().await = false;
                launch_server(CoreExtCfgParser).await;
            });
        });

        sleep(Duration::from_secs(1)).await;
        task
    }

    fn resolve_misc_file(file_name: &str) -> std::path::PathBuf {
        let candidates = [format!("misc/{file_name}"), format!("../misc/{file_name}")];

        for candidate in candidates {
            let path = std::path::PathBuf::from(candidate);
            if path.exists() {
                return std::fs::canonicalize(path)
                    .expect("Failed to resolve path in misc directory");
            }
        }

        panic!("Failed to find misc/{file_name}");
    }
}