tag2upload_service_manager/

global.rs

//! Global state
//!
//! # Lock hierarchy
//!
//! In order of acquisition:
//!
//!  1. sqlite database lock
//!  2. `state`, `running`, `db_trigger`,
//!     `host_retry_state` - only one at a time

use crate::prelude::*;

use crate::o2m_support::OracleSeq;

use std::sync::Mutex;

const USER_AGENT: &str =
    "tag2upload-service-manager https://salsa.debian.org/dgit-team";

#[derive(Debug)]
pub struct Globals {
    pub cli_options: CliOptions,
    pub config: Config,
    pub computed_config: ComputedConfig,

    /// Shared global state *excluding the db*
    ///
    /// ### Lock hierarchy
    ///
    /// See the [docs for the `global` module](crate::global)
    /// for the rules for when you may mutate or borrow this.
    pub state: watch::Sender<State>,

    pub worker_tracker: Arc<WorkerTracker>,

    /// Trigger watch for database updates
    ///
    ///  * When a db transaction modifies the DB such that a
    ///    job may become processable that wasn't previously,
    ///    this watch will be signaled (even though the
    ///    contained `DbAssocState` may not have changed).
    ///
    ///    `db_support.rs` functions taking a [`TransactionNature`]
    ///    ensure this.
    ///
    ///  * Additionally `DbAssocState` contains some information
    ///    about pausing mirrored from the database.
    ///
    /// ### Lock hierarchy
    ///
    /// See the [docs for the `global` module](crate::global)
    /// for the rules for when you may mutate or borrow this.
    pub db_trigger: watch::Sender<DbAssocState>,

    /// Set iff we have run the db initialisation (including schema)
    ///
    /// This is done from the Rocket liftoff fairing callback, while the
    /// rest of the program is (supposedly) waiting.  We want to avoid
    /// running any normal db transactions before then.  This is a booby
    /// trap which would cause us to crash (or fail tests) instead.
    pub db_initialised: OnceLock<Result<(), Arc<AE>>>,

    pub http_client: reqwest::Client,
    pub dns_resolver: dns::Resolver,

    /// Signal for when we have started up
    ///
    /// ### Lock hierarchy
    ///
    /// See the [docs for the `global` module](crate::global)
    /// for the rules for when you may mutate or borrow this.
    pub running: watch::Receiver<Option<Running>>,

    /// Ephemeral per-host retry/success records
    ///
    /// ### Lock hierarchy
    ///
    /// See the [docs for the `global` module](crate::global)
    /// for the rules for when you may lock this.
    pub host_retry_state: Mutex<retry::HostStates>,

    pub scratch_dir: String,
    pub temp_dir_retain: Option<tempfile::TempDir>,
    pub tera: Tera,
    pub version_info: crate::version::Info,
    pub test_suppl: test::GlobalSupplement,
    pub last_worker_restart: RwLock<SystemTime>,

    /// Connection sequence number for the next connection to come in.
    ///
    /// This is a global so that we can determine which worker
    /// connections should be restarted across, and independently of,
    /// all active listeners (even though for now we have only one
    /// listener).
    pub next_seq: Mutex<OracleSeq>,

    /// Lowest connection sequence number of a connection to an Oracle
    /// worker that doesn't need restarting.  Connections to workers
    /// with sequence numbers strictly less than this should issue a
    /// 'restart-worker' instruction and disconnect.
    pub oldest_norestart_seq: watch::Sender<OracleSeq>,
}

#[derive(Debug)]
pub struct Unchecked<T> { pub unchecked: T }

#[derive(Clone, Debug)]
pub struct Running {
    pub port: u16,
}

pub struct Started {
    pub rocket: RocketIgnite,
}

/// State that is maintained in tandem with db transactions
///
/// But not stored permenantly.
#[derive(Default, Debug)]
pub struct DbAssocState {
    pub paused: Option<IsPaused>,

    pub release_jobs: HashMap<JobId, WorkerFidelity>,
}

#[derive(Debug, Clone, Copy)] pub struct IsPaused;
#[derive(Debug, Clone, Copy)] pub struct IsThrottled;

#[cfg(not(test))]
mod imp_globals {
    use super::*;
    static GLOBALS: OnceLock<Arc<Globals>> = OnceLock::new();

    /// Obtain the global variables - **only for use after startup**
    ///
    /// # Panics
    ///
    /// Panics if called when globals have not yet been set,
    /// during [`startup`].
    /// 
    /// This is only allowed after [`startup`]
    /// (in `mod globals`, after `set_globals`).
    pub fn globals() -> Arc<Globals> {
        GLOBALS.get()
            .expect("Using globals() before set")
            .clone()
    }

    /// Set the global variables - call precisely once
    pub(crate) fn set_globals(g: Arc<Globals>) {
        GLOBALS.set(g)
            .expect("set_globals called more than once?");
    }
}
#[cfg(test)]
mod imp_globals {
    // In testing, we use a single-threaded tokio executor
    // provided by `#[tokio::test]`.
    //
    // We don't spawn threads, and tokio relies on the async executor.
    // So we can use thread locals (which isolates different tests).

    use super::*;
    thread_local! {
        static GLOBALS: RefCell<Option<Arc<Globals>>> =
            const { RefCell::new(None) };
    }
    pub fn globals() -> Arc<Globals> {
        GLOBALS.with_borrow(|gl| gl.clone())
            .expect("Using globals() before set")
    }
    pub(crate) fn set_globals(g: Arc<Globals>) {
        if let Some(was) = GLOBALS.replace(Some(g)) {
            panic!("set_globals called more than once, previously {was:?}");
        }
    }
}
pub use imp_globals::*;

/// Token indicating that we are shutting down
///
/// This is not an error.
/// If an error causes shutdown, `State.shutdown_reason` is `Err`,
/// but other functions may return a [`ShuttingDown`];
#[derive(Debug)]
pub struct ShuttingDown;

#[derive(Debug)]
pub struct State {
    pub shutdown_reason: Option<Result<ShuttingDown, InternalError>>,

    pub test_suppl: test::StateSupplement,
}

impl State {
    pub fn new(test_suppl: test::StateSupplement) -> Self {
        State {
            shutdown_reason: None,
            test_suppl,
        }
    }

    /// Checks for shutdown (but doesn't do anything with errors)
    pub fn check_shutdown(&self) -> Result<(), ShuttingDown> {
        match &self.shutdown_reason {
            Some(Ok(ShuttingDown)) => Err(ShuttingDown),
            Some(Err(_)) => Err(ShuttingDown),
            None => Ok(()),
        }
    }
}

impl Globals {
    /// Waits for shutdown (possibly indefinitely)
    ///
    /// Like `check_shutdown` this doesn't do anything with errors
    pub async fn await_shutdown(&self) -> ShuttingDown {
        self.state.subscribe()
            .wait_for_then(|state| {
                state.shutdown_reason
                    .as_ref()
                    .map(|_: &Result<ShuttingDown, IE>| ())
            })
            .await
            .unwrap_or_else(|e| {
                error!("shutdown handler task recv failed {e}");
            });

       ShuttingDown
    }
}

/// If we are running tests, submit a test webhook.
/// Otherwise do nothing.
pub async fn test_hook<S: Display>(
    #[allow(unused_variables)]
    point: impl FnOnce() -> S + Send + Sync,
) {
    #[cfg(test)]
    test::hook_point(point().to_string()).await;
}

#[derive(Debug)]
pub enum HttpFetchedRaw {
    Success(reqwest::Response),
    SuccessFile { file: String, data: String },
    NotFound(HttpNotFound),
}

impl Globals {
    /// Start this task when we have entered the Running state
    pub fn spawn_task_running(
        self: &Arc<Self>,
        what: impl Display,
        fut: impl Future<Output = TaskResult> + Send + 'static,
    ) {
        let gl = self.clone();
        self.spawn_task_immediate(what, async move {
            let Running { .. } = gl.await_running().await?;
            fut.await
        })
    }

    /// Start this task immediately
    pub fn spawn_task_immediate(
        self: &Arc<Self>,
        what: impl Display,
        fut: impl Future<Output = TaskResult> + Send + 'static,
    ) {
        trace!("spawning task {what}");
        tokio::spawn({
            let what = what.to_string();
            async move {
                let _: TaskResult = AssertUnwindSafe(fut)
                    .catch_unwind()
                    .await
                    .unwrap_or_else(|_: Box<dyn Any + Send>| {
                        Err(internal!("task {what} panicked!").into())
                    });
            }
        });
    }

    /// Checks for shutdown (but doesn't do anything with errors)
    pub fn check_shutdown(&self) -> Result<(), ShuttingDown> {
        self.state.borrow().check_shutdown()
    }

    pub async fn await_running(&self) -> Result<Running, ShuttingDown> {
        match self.running.clone().wait_for_then(|p| p.clone()).await {
            Ok(y) => Ok(y),
            Err(e) => {
                debug!("shutting down, no port: {e}");
                Err(ShuttingDown)
            }
        }
    }

    pub async fn http_fetch_raw(
        &self,
        url: Url,
    ) -> Result<HttpFetchedRaw, AE> {
        use HttpFetchedRaw as H;

        let via_file = |file: String| {
            if {
                file
                    .strip_prefix('/').expect("allow not absolute?")
                    .split('/')
                    .any(|ent| ent == "" || ent == "..")
            } {
                return Err(anyhow!("file url {url:?} path traversal"));
            }
            let data = match fs::read_to_string(&file) {
                Err(e) if e.kind() == io::ErrorKind::NotFound => {
                    return Ok(H::NotFound(HttpNotFound))
                },
                Err(e) if e.kind() == io::ErrorKind::IsADirectory => {
                    let data = "DIRECTORY LISTING".into();
                    return Ok(H::SuccessFile { file, data })
                },
                other => {
                    other.with_context(|| format!("{file:?}"))
                        .context("failed to read file")?
                },
            };
            Ok(H::SuccessFile { file, data })
        };

        // Handle certain file:/// URLs.
        // We only want to serve URLs that are expected by our test
        // harness.  Otherwise other processes on the machine could
        // suborn our test runs into accessing arbitrary files!
        let file_allows = [
            self.config.testing.fake_https_dir.clone()
                .into_iter().collect_vec(),
            #[cfg(test)]
            self.test_suppl.url_map_target_files()
        ];
        let file_allows = file_allows.iter().flatten();

        if file_allows.clone().next().is_some() {
            let url = url.to_string();

            if let Some(path) = url.strip_prefix("file://") {
                if !file_allows.clone().any(|allow| path.starts_with(allow)) {
                    return Err(anyhow!("file url {url:?} not allowed ({:?})",
                                       file_allows.clone().collect_vec()));
                }
                return via_file(path.to_owned());
            }
        }

        if let Some(fake) = &self.config.testing.fake_https_dir {
            let url = url.to_string();
            let url = url.strip_prefix("https://")
                .ok_or_else(|| anyhow!(
                    "failed to strip https:// prefix from url {url:?}"
                ))?;
            let file = format!("{fake}/{url}");
            return via_file(file);
        }

        let response = self.http_client.get(url)
            .send().await.context("send")?;
        if response.status() == reqwest::StatusCode::NOT_FOUND {
            return Ok(H::NotFound(HttpNotFound))
        }
        let response = response
            .error_for_status().context("status")?;

        Ok(H::Success(response))
    }

    pub async fn http_fetch_json_maybe<T: serde::de::DeserializeOwned>(
        &self,
        url: Url,
    ) -> Result<Result<T, HttpNotFound>, AE> {
        use HttpFetchedRaw as H;

        let r = match self.http_fetch_raw(url.clone()).await? {
            H::Success(response) => {
                response.json().await.context("response")?
            },
            H::SuccessFile { file, data } => {
                serde_json::from_str(&data)
                    .with_context(|| format!("{file:?}"))
                    .context("failed to deser file")?
            },
            H::NotFound(e) => {
                return Ok(Err(e))
            },
        };

        Ok(Ok(r))
    }

    pub async fn http_fetch_json<T: serde::de::DeserializeOwned>(
        &self,
        url: Url,
    ) -> Result<T, AE> {
        Ok(self.http_fetch_json_maybe(url).await??)
    }
}

macro_rules! test_hook_url { { $url:ident } => {
    #[cfg(test)]
    let $url = crate::test::UrlMappable::map(&$url);
} }

pub fn shutdown_start_tasks(
    globals: &Arc<Globals>,
) -> Result<(), StartupError> {
    use tokio::signal::unix::{signal, SignalKind as SK};

    #[cfg(test)]
    match globals.t_shutdown_handlers() {
        Ok(()) => {},
        Err(()) => return Ok(()),
    };

    let mut terminate = signal(SK::terminate())
        .into_internal("failed to set up SIGTERM handler")?;

    globals.spawn_task_immediate("shutdown SIGTEREM watch", {
        let globals = globals.clone();
        async move {

            let () = terminate.recv().await
                .ok_or_else(|| internal!("no more SIGTERM reception?!"))?;

            globals.state.send_modify(|state| {
                match state.shutdown_reason {
                    None => {
                        info!("received SIGTERM, shutting down...");
                        state.shutdown_reason = Some(Ok(ShuttingDown));
                    }
                    Some(Err(_)) => {
                        info!("SIGTERM, but already crashing!");
                    },
                    Some(Ok(ShuttingDown)) => {
                        info!("SIGTERM, but already shutting down");
                    }
                }
            });

            Ok(TaskWorkComplete {})
        }
    });

    globals.spawn_task_immediate("shutdown handler", {
        let globals = globals.clone();
        async move {

            let _: ShuttingDown = globals.await_shutdown().await;

            let mut subscription = globals.db_trigger.subscribe();

            match loop {
                let job = find_job_deferring_shutdown()?;

                let Some::<JobRow>(job) = job
                else { break Ok(()); };

                info!(jid=%job.jid, "shutdown awaits completion of build");

                match subscription.changed().await {
                    Ok(()) => {}, // go round again
                    Err(e) => break Err(e),
                }
            } {
                Ok(()) => info!("clean shutdown complete."),
                Err(e) => error!("shutdown terminating early, watch {}", e),
            };

            unsafe {
                libc::kill(0, libc::SIGHUP);
                error!("SIGHUP didn't kill us!!");
                std::process::abort();
            }
        }
    });

    Ok(())
}

pub fn find_job_deferring_shutdown() -> Result<Option<JobRow>, IE> {
    db_transaction(TN::Readonly, |dbt| {
        dbt.bsql_query_01(&bsql!("
                        SELECT * FROM jobs
                         WHERE processing != ''
                           AND status = " (JobStatus::Building) "
                      ORDER BY last_update ASC
                    "))
    })
}

fn handle_errors_during_rocket_liftoff(
    context: impl Display,
    f: impl FnOnce() -> Result<(), AE>,
) -> Result<(), ShuttingDown> {
    f()
        .with_context(|| context.to_string())
        .map_err(|ae| {
            IE::new_without_backtrace(ae).note_only();
            ShuttingDown
        })
}

fn write_port_report_file(gl: &Arc<Globals>, port: u16)
                          -> Result<(), ShuttingDown>
{
    if let Some(file) = gl.config.files.port_report_file.clone() {
        trace!(?file, "writing port");

        handle_errors_during_rocket_liftoff(format_args!("{file:?}"), || {
            let f = fs::File::create(&file).context("open")?;
            let mut f = io::BufWriter::new(f);
            writeln!(f, "{port}").context("write")?;
            f.flush().context("write (flush)")?;
            Ok(())
        })?;
    }
    Ok(())
}

pub fn resolve_config(
    cli_options: CliOptions,
    base_config: Figment,
) -> Result<Unchecked<WholeConfig>, StartupError> {
    use StartupError as SE;

    let rocket_base_config = {
        use rocket::config::*;
        rocket::Config {
            shutdown: Shutdown {
                ctrlc: false,
                signals: HashSet::new(),
                ..Default::default()
            },
            ..rocket::Config::release_default()
        }
    };

    let config = {
        let mut config = base_config;
        for file in &cli_options.config {
            let fig = figment::providers::Toml::file_exact(&file);
            config = config.merge(fig);
        }
        config
    }
        // we want to merge the default logging schedule with the config
        .join(figment::providers::Serialized::default(
            "log.schedule",
            logging::default_schedule(),
        ))
        // rocket::Config's Deserialize impl doesn't include any of
        // the usual defaults, so we must inject them here.
        .join(figment::providers::Serialized::default(
            "rocket",
            rocket_base_config,
        ))
        .join(figment::providers::Serialized::default(
            "rocket",
            // configuration for rocket addons that expect to find their
            // information in the configuration passed to rocket::custom,
            // but which *aren't* part of rocket::Config.
            json! {{
                // Currently there are none of these.
            }},
        ));

    let config = {
        let mut c = config;

        for s in &cli_options.config_toml {
            c = c.merge(figment::providers::Toml::string(s));
        }

        c
    };

    let rocket_config = config
        .focus("rocket");

    let config: Config = config
        .extract()
        .map_err(SE::ParseConfig)?;

    let computed_config = ComputedConfig::try_from(&config)?;

    let unchecked = 
        WholeConfig { cli_options, config, computed_config, rocket_config };
    Ok(Unchecked { unchecked })
}

impl Unchecked<WholeConfig> {
    pub fn check(self) -> Result<WholeConfig, StartupError> {
        self.unchecked.config.check()?;
        Ok(self.unchecked)
    }
}

pub async fn startup(
    config: WholeConfig,
    test_global_suppl: test::GlobalSupplement,
    test_state_suppl: test::StateSupplement,
    rocket_hook: impl FnOnce(RocketBuild) -> RocketBuild,
) -> Result<Started, StartupError> {
    use StartupError as SE;

    let WholeConfig { cli_options, config, computed_config, rocket_config }
        = config;

    logging::setup(&config)?;

    debug!(?config, ?cli_options, ?computed_config, "starting");

    let scratch_dir;
    let temp_dir_retain;
    match &config.files.scratch_dir {
        Some(s) => {
            scratch_dir = s.clone();
            temp_dir_retain = None;
        }
        None => {
            let td = tempfile::TempDir::new()
                .context("create temp dir")
                .map_err(SE::TempDir)?;
            scratch_dir = td.path().to_str()
                .ok_or_else(|| anyhow!("not utf-8"))
                .map_err(SE::TempDir)?
                .to_owned();
            temp_dir_retain = Some(td);
        }
    };

    remove_dir_all::remove_dir_contents(&scratch_dir)
        .context("clean out old contents of scratch directory")
        .map_err(SE::TempDir)?;

    let version_info = crate::version::calculate_describe(&config.files);

    let http_client = reqwest::Client::builder()
            .user_agent(USER_AGENT)
        .timeout(*config.timeouts.http_request)
        .build()?;

    let dns_resolver = dns::Resolver::builder_tokio()?.build();

    // We could use rocket_dyn_templates, but it insists on reloading
    // the /dev/null we have to supply as template_dir, whenever anyone
    // on the whole system touches it.
    let tera = ui_render::tera_templates(&config)?;

    let (running_tx, running) = watch::channel(None);

    let last_worker_restart = Globals::now_systemtime_inner(
        &config,
        &test_global_suppl,
    ).into();

    let globals = Arc::new(Globals {
        cli_options,
        config,
        computed_config,
        db_trigger: watch::Sender::new(DbAssocState::default()),
        db_initialised: OnceLock::new(),
        state: watch::Sender::new(State::new(test_state_suppl)),
        worker_tracker: Default::default(),
        http_client,
        dns_resolver,
        scratch_dir,
        running,
        tera,
        last_worker_restart,
        host_retry_state: Default::default(),
        version_info,
        temp_dir_retain,
        test_suppl: test_global_suppl,
        next_seq: Mutex::new(1),
        oldest_norestart_seq: watch::Sender::new(1),
    });
    set_globals(globals.clone());

    let listener = o2m_listener::Listener::new(&globals)?;

    shutdown_start_tasks(&globals)?;

    expire::start_task(&globals);

    globals.spawn_task_running("unpause", {
        let globals = globals.clone();
        db_workflow::unpause_task(globals)
    });

    let rocket = rocket::custom(&rocket_config);
    let rocket = rocket_hook(rocket);
    let rocket = api_routes::mount_api(rocket);
    let rocket = ui_routes::mount_ui_catchers(rocket);

    let rocket = rocket.attach({
        let globals = globals.clone();

        rocket::fairing::AdHoc::on_liftoff(
            "spawn workers",
            |rocket: &rocket::Rocket<_>| Box::pin(
                async move {
                    handle_errors_during_rocket_liftoff("initialise db", || {
                        db_support::initialise(&globals)?;
                        Ok(())
                    })?;

                    if globals.state.borrow().shutdown_reason.is_some() {
                        // Typically, if this happens, an internal error
                        trace!(
                         "shutdown triggered during startup, not continuing"
                        );
                        return Err(ShuttingDown);
                    }
                    fetcher::start_tasks(&globals);
                    listener.start_task();

                    let port = rocket.config().port;
                    // We do this here, directly, rather than in a task
                    // hanging off the channel, because our tests check
                    // that this file has been written and use the channel
                    // as a signal to see that it has been done.
                    write_port_report_file(&globals, port)?;

                    let running = Running {
                        port
                    };
                    running_tx.send(Some(running.clone()))
                        .expect("no-one wanted our port");

                    info!(?running, "running");

                    Ok::<_, ShuttingDown>(())
                }.map(
                    // Err(()) means we cancelled the startup early.
                    // We've reported it, but we can't return it to Rocket.
                    |r| r.unwrap_or_else(|ShuttingDown| {
                        eprintln!("exiting due to error during startup!");
                        error!("exiting due to error during startup!");

                        #[cfg(not(test))] 
                        std::process::exit(16);
                        #[allow(unreachable_code)] // what a joke
                        { unreachable!(); }
                    })
                )
            )
        )
    });

    let rocket = rocket.ignite().await?;

    Ok(Started {
        rocket,
    })
}