synit/synit-pid1/src/main.rs

use clap::Parser;

use nix::sys::reboot::{reboot, RebootMode};
use nix::sys::signal::{kill, Signal, SigHandler};
use nix::sys::wait;
use nix::unistd;

use std::convert::TryInto;
use std::io::Write;
use std::sync::Arc;

use syndicate::actor::*;
use syndicate::relay;
use syndicate::sturdy;
use syndicate::value::NestedValue;
use syndicate::value::Value;

use tokio::process;
use tokio::select;
use tokio::signal::unix::{signal, SignalKind};

#[derive(Parser, Clone, Debug)]
#[command(version)]
pub struct CommandLine {
    #[arg(long, default_value="/usr/bin/syndicate-server")]
    server_path: String,

    #[arg(long, default_value="/sbin/synit-log")]
    log: String,
}

pub struct Pid1Listener {
    // config: Arc<CommandLine>,
}

impl Pid1Listener {
    fn new(_config: Arc<CommandLine>) -> Self {
        Self {
            // config,
        }
    }
}

impl Entity<AnyValue> for Pid1Listener {
}

fn ignore_signal(kind: SignalKind) -> ActorResult {
    let sig: Signal = kind.as_raw_value().try_into()?;
    unsafe { nix::sys::signal::signal(sig, SigHandler::SigIgn) }?;
    Ok(())
}

async fn handle_sigchld_and_waitpid() -> Result<RebootMode, ActorError> {
    // For now, we use Alpine's userland reboot, poweroff and halt tools.
    // These send SIGTERM, SIGUSR2 and SIGUSR1, respectively.

    let mut sigchlds = signal(SignalKind::child())?;
    let mut sigints = signal(SignalKind::interrupt())?;
    let mut sigterms = signal(SignalKind::terminate())?;
    let mut sigusr2s = signal(SignalKind::user_defined2())?;
    let mut sigusr1s = signal(SignalKind::user_defined1())?;

    tracing::info!("Awaiting signals...");
    let next_step = loop {
        select! {
            _ = sigchlds.recv() => {
                loop {
                    match wait::waitpid(None, Some(wait::WaitPidFlag::WNOHANG)) {
                        Ok(wait::WaitStatus::StillAlive) => {
                            tracing::debug!("No child processes to reap at this time");
                            break;
                        }
                        Ok(status) =>
                            tracing::debug!("Child process reaped: {:?}", status),
                        Err(nix::errno::Errno::ECHILD) => {
                            tracing::debug!("waitpid(2) yielded ECHILD");
                            break;
                        }
                        Err(e) =>
                            Err(e)?,
                    }
                }
            }
            _ = sigints.recv() => {
                tracing::debug!("Received SIGINT");
                break RebootMode::RB_AUTOBOOT;
            }
            _ = sigterms.recv() => {
                tracing::debug!("Received SIGTERM");
                break RebootMode::RB_AUTOBOOT;
            }
            _ = sigusr2s.recv() => {
                tracing::debug!("Received SIGUSR2");
                break RebootMode::RB_POWER_OFF;
            }
            _ = sigusr1s.recv() => {
                tracing::debug!("Received SIGUSR1");
                break RebootMode::RB_HALT_SYSTEM;
            }
        }
    };

    ignore_signal(SignalKind::interrupt())?;
    ignore_signal(SignalKind::terminate())?;
    ignore_signal(SignalKind::user_defined2())?;
    ignore_signal(SignalKind::user_defined1())?;

    tracing::info!("Terminating, {:?}", next_step);

    std::io::stdout().flush()?;
    std::io::stderr().flush()?;
    nix::unistd::sync();
    std::thread::sleep(std::time::Duration::from_millis(100));

    nix::unistd::sync();
    let _ = kill(unistd::Pid::from_raw(-1), Some(Signal::SIGINT));
    // ^ ignore result! We're about to reboot anyway

    nix::unistd::sync();
    Ok(next_step)
}

#[tokio::main]
async fn main() -> ActorResult {
    syndicate::convenient_logging()?;

    tracing::info!("Startup with PID {}", unistd::getpid());

    match unistd::setsid() {
        Ok(_pid) => tracing::info!("setsid(2): new session is {}", _pid),
        Err(e) => tracing::info!("setsid(2) failed: {:?}", &e),
    }

    match unistd::setpgid(unistd::Pid::from_raw(0), unistd::Pid::from_raw(0)) {
        Ok(()) => tracing::info!("setpgid(2) succeeded"),
        Err(e) => tracing::info!("setpgid(2) failed: {:?}", &e),
    }

    let config = Arc::new(CommandLine::parse());

    Actor::top(None, move |t| {
        let server = process::Command::new(&config.server_path)
            .arg("--inferior")
            .arg("--config")
            .arg("/etc/syndicate/boot")
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            .spawn()
            .expect("Could not spawn main server");

        let server_stderr: std::process::Stdio = server.stderr
            .expect("fetching server stderr")
            .try_into()
            .expect("converting server stderr to Stdio");

        if config.log.len() > 0 {
            process::Command::new(&config.log)
                .stdin(server_stderr)
                .spawn()
                .expect("Could not spawn log program");
        }

        let listener = t.create(Pid1Listener::new(config));
        let from_server = server.stdout.expect("Missing dataspace server stdout");
        let to_server = server.stdin.expect("Missing dataspace server stdin");

        let ds = &relay::TunnelRelay::run(t,
                                          relay::Input::Bytes(Box::pin(from_server)),
                                          relay::Output::Bytes(Box::pin(to_server)),
                                          None,
                                          Some(sturdy::Oid(0.into())),
                                          false)
            .expect("Missing reference to dataspace")
            .underlying;
        t.assert(ds, Value::simple_record1("init", AnyValue::domain(Cap::new(&listener))).wrap());
        Ok(())
    });

    reboot(handle_sigchld_and_waitpid().await?)?;

    Ok(())
}