qsu/

rt.rs

//! Server application wrapper runtime.
//!
//! # Overview
//! The _qsu_'s runtime lives in a layer between the actual server application
//! and the "service subsystems" that servier applications can integrate
//! against:
//!
//! <pre>
//! +---------------------+
//! |  Server Application |  (HTTP server, Matrix server, etc)
//! +---------------------+
//! |       qsu rt        |
//! +---------------------+
//! |  Platform service   |  (Windows Service subsystem, systemd, etc)
//! |      subsystem      |
//! +---------------------+
//! </pre>
//!
//! The primary goal of the _qsu_ runtime is to provide a consistent interface
//! that will be mapped to whatever service subsystem it is actually running
//! on.  This including no special subsystem at all; such as when running the
//! server application as a regular foreground process (which is common during
//! development).
//!
//! # How server applications are implemented and used
//! _qsu_ needs to know what kind of runtime the server application expects.
//! Server applications pick the runtime type by implementing a trait, of which
//! there are currently three recognized types:
//! - [`ServiceHandler`] is used for "non-async" server applications.
//! - [`TokioServiceHandler`] is used for server applications that run under
//!   the tokio executor.
//! - [`RocketServiceHandler`] is for server applications that are built on top
//!   of the Rocket HTTP framework.
//!
//! Each of these implement three methods:
//! - `init()` is for initializing the service.
//! - `run()` is for running the actual server application.
//! - `shutdown()` is for shutting down the server application.
//!
//! The specifics of these trait methods may look quite different.
//!
//! Once a service trait has been implemented, the application creates a
//! [`RunCtx`] object and calls its [`run()`](RunCtx::run()) method, passing in
//! an service implementation object.
//!
//! Note: Only one service wrapper must be initialized per process.
//!
//! ## Service Handler semantics
//! When a handler is run through the [`RunCtx`] it will first call the
//! handler's `init()` method.  If it returns `Ok()`, its `run()` method will
//! be run.
//!
//! The handler's `shutdown()` will be called regardless of whether `init()` or
//! `run()` was successful (the only precondition for `shutdown()` to be called
//! is that `init()` was called).
//!
//! # Argument parser
//! _qsu_ offers an [argument parser](crate::argp::ArgParser), which can
//! abstract away much of the runtime management and service registration.

mod nosvc;
mod rttype;
mod signals;

#[cfg(all(target_os = "linux", feature = "systemd"))]
#[cfg_attr(docsrs, doc(cfg(feature = "systemd")))]
mod systemd;

#[cfg(windows)]
pub mod winsvc;

use std::sync::{
  atomic::{AtomicU32, Ordering},
  Arc, OnceLock
};

#[cfg(any(feature = "tokio", feature = "rocket"))]
use async_trait::async_trait;

#[cfg(feature = "tokio")]
use tokio::runtime;

use tokio::sync::broadcast;

#[cfg(all(target_os = "linux", feature = "systemd"))]
use sd_notify::NotifyState;


use crate::{err::CbErr, lumberjack::LumberJack};


#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub enum RunAs {
  Foreground,
  SvcSubsys
}

/// Keep track of whether the service application is running as foreground
/// process or running under a service subsystem.
static RUNAS: OnceLock<RunAs> = OnceLock::new();

pub fn runas() -> Option<RunAs> {
  RUNAS.get().copied()
}


/// The run time environment.
///
/// Can be used by the application callbacks to determine whether it is running
/// as a service.
#[derive(Debug, Clone)]
pub enum RunEnv {
  /// Running as a foreground process.
  Foreground,

  /// Running as a service.
  ///
  /// If the service subsystem has named services, this will contain the
  /// service name.
  Service(Option<String>)
}

/// Used to pass an optional message to the service subsystem whenever a
/// startup or shutdown checkpoint as been reached.
pub enum StateMsg {
  Ref(&'static str),
  Owned(String)
}

impl From<&'static str> for StateMsg {
  fn from(msg: &'static str) -> Self {
    Self::Ref(msg)
  }
}

impl From<String> for StateMsg {
  fn from(msg: String) -> Self {
    Self::Owned(msg)
  }
}

impl AsRef<str> for StateMsg {
  fn as_ref(&self) -> &str {
    match self {
      Self::Ref(s) => s,
      Self::Owned(s) => s
    }
  }
}


/// Report the current startup/shutdown state to the platform service
/// subsystem.
pub(crate) trait StateReporter {
  fn starting(&self, checkpoint: u32, msg: Option<StateMsg>);

  fn started(&self);

  fn stopping(&self, checkpoint: u32, msg: Option<StateMsg>);

  fn stopped(&self);
}


/// Context passed to `init()` service application callback.
///
/// This context can be used to query whether the service application is
/// running as foreground process or running within a service subsystem.
///
/// It can also be used to report progress status back to the service
/// subsystems, for platforms that support it.
pub struct InitCtx {
  re: RunEnv,
  sr: Arc<dyn StateReporter + Send + Sync>,
  cnt: Arc<AtomicU32>
}

impl InitCtx {
  /// Return context used to identify whether service application is running as
  /// a foreground process or a system service.
  #[must_use]
  pub fn runenv(&self) -> RunEnv {
    self.re.clone()
  }

  /// Report startup state to the system service manager.
  ///
  /// For foreground processes and services that do not support startup state
  /// notifications this method has no effect.
  pub fn report(&self, status: Option<StateMsg>) {
    let checkpoint = self.cnt.fetch_add(1, Ordering::SeqCst);
    if let Some(ref msg) = status {
      tracing::trace!(
        "Reached init checkpoint {checkpoint}; {}",
        msg.as_ref()
      );
    } else {
      tracing::trace!("Reached init checkpoint {checkpoint}");
    }
    self.sr.starting(checkpoint, status);
  }
}


/// Context passed to `term()` service application callback.
///
/// This context can be used to query whether the service application is
/// running as foreground process or running within a service subsystem.
///
/// It can also be used to report progress status back to the service
/// subsystems, for platforms that support it.
pub struct TermCtx {
  re: RunEnv,
  sr: Arc<dyn StateReporter + Send + Sync>,
  cnt: Arc<AtomicU32>
}

impl TermCtx {
  /// Return context used to identify whether service application is running as
  /// a foreground process or a system service.
  #[must_use]
  pub fn runenv(&self) -> RunEnv {
    self.re.clone()
  }

  /// Report shutdown state to the system service manager.
  ///
  /// For foreground processes and services that do not support shutdown state
  /// notifications this method has no effect.
  pub fn report(&self, status: Option<StateMsg>) {
    let checkpoint = self.cnt.fetch_add(1, Ordering::SeqCst);
    if let Some(ref msg) = status {
      tracing::trace!(
        "Reached term checkpoint {checkpoint}; {}",
        msg.as_ref()
      );
    } else {
      tracing::trace!("Reached term checkpoint {checkpoint}");
    }
    self.sr.stopping(checkpoint, status);
  }
}


/// "Synchronous" (non-`async`) server application.
///
/// Implement this for an object that wraps a server application that does not
/// use an async runtime.
pub trait ServiceHandler {
  type AppErr;

  /// Implement to handle service application initialization.
  ///
  /// # Errors
  /// Application-defined error returned from callback will be wrapped in
  /// [`CbErr::App`] and returned to application.
  fn init(&mut self, ictx: InitCtx) -> Result<(), Self::AppErr>;

  /// Implement to run service application.
  ///
  /// # Errors
  /// Application-defined error returned from callback will be wrapped in
  /// [`CbErr::App`] and returned to application.
  fn run(&mut self, re: &RunEnv) -> Result<(), Self::AppErr>;

  /// Implement to handle service application termination.
  ///
  /// # Errors
  /// Application-defined error returned from callback will be wrapped in
  /// [`CbErr::App`] and returned to application.
  fn shutdown(&mut self, tctx: TermCtx) -> Result<(), Self::AppErr>;
}


/// `async` server application built on the tokio runtime.
///
/// Implement this for an object that wraps a server application that uses
/// tokio as an async runtime.
#[cfg(feature = "tokio")]
#[cfg_attr(docsrs, doc(cfg(feature = "tokio")))]
#[async_trait]
pub trait TokioServiceHandler {
  type AppErr;

  async fn init(&mut self, ictx: InitCtx) -> Result<(), Self::AppErr>;

  async fn run(&mut self, re: &RunEnv) -> Result<(), Self::AppErr>;

  async fn shutdown(&mut self, tctx: TermCtx) -> Result<(), Self::AppErr>;
}


/// Rocket server application handler.
///
/// While Rocket is built on top of tokio, it \[Rocket\] wants to initialize
/// tokio itself.
///
/// There are two major ways to write Rocket services using qsu; either the
/// application can let qsu be aware of the server applications' `Rocket`
/// instances.  It does this by creating the `Rocket` instances in
/// `RocketServiceHandler::init()` and returns them.  _qsu_ will ignite these
/// rockets and pass them to `RocketServiceHandler::run()`.  The application is
/// responsible for launching the rockets at this point.
///
/// The other way to do it is to completely manage the `Rocket` instances in
/// application code (by not returning rocket instances from `init()`).
///
/// Allowing _qsu_ to manage the `Rocket` instances will cause _qsu_ to request
/// graceful shutdown of all `Rocket` instances once a `SvcEvt::Shutdown` is
/// sent by the runtime.
///
/// It is recommended that `ctrlc` shutdown and termination signals are
/// disabled in each `Rocket` instance's configuration, and allow the _qsu_
/// runtime to be responsible for initiating the `Rocket` shutdown.
#[cfg(feature = "rocket")]
#[cfg_attr(docsrs, doc(cfg(feature = "rocket")))]
#[async_trait]
pub trait RocketServiceHandler {
  type AppErr;

  /// Rocket service initialization.
  ///
  /// The returned `Rocket`s will be ignited and their shutdown handlers will
  /// be triggered on shutdown.
  async fn init(
    &mut self,
    ictx: InitCtx
  ) -> Result<Vec<rocket::Rocket<rocket::Build>>, Self::AppErr>;

  /// Server application main entry point.
  ///
  /// If the `init()` trait method returned `Rocket<Build>` instances to the
  /// qsu runtime it will ignote these and pass them to `run()`.  It is the
  /// responsibility of this method to launch the rockets and await them.
  async fn run(
    &mut self,
    rockets: Vec<rocket::Rocket<rocket::Ignite>>,
    re: &RunEnv
  ) -> Result<(), Self::AppErr>;

  async fn shutdown(&mut self, tctx: TermCtx) -> Result<(), Self::AppErr>;
}


/// The means through which the service termination happened.
#[derive(Copy, Clone, Debug)]
pub enum Demise {
  /// On unixy platforms, this indicates that the termination was initiated
  /// via a `SIGINT` signal.  When running as a foreground process on Windows
  /// this indicates that Ctrl+C was issued.
  Interrupted,

  /// On unixy platforms, this indicates that the termination was initiated
  /// via the `SIGTERM` signal.
  ///
  /// On Windows, running as a service, this indicates that the service
  /// subsystem requested service to be shut down. Running as a foreground
  /// process, this means that Ctrl+Break was issued or that the console
  /// window was closed.
  Terminated,

  /// Reached the end of the service application without any external requests
  /// to terminate.
  ReachedEnd
}

#[derive(Copy, Clone, Debug)]
pub enum UserSig {
  /// SIGUSR1
  Sig1,

  /// SIGUSR2
  Sig2
}


/// Event notifications that originate from the service subsystem that is
/// controlling the server application.
#[derive(Copy, Clone, Debug)]
pub enum SvcEvt {
  /// User events.
  ///
  /// These will be generated on unixy platform if the process receives
  /// SIGUSR1 or SIGUSR2.
  User(UserSig),

  /// Service subsystem has requested that the server application should pause
  /// its operations.
  ///
  /// Only the Windows service subsystem will emit these events.
  Pause,

  /// Service subsystem has requested that the server application should
  /// resume its operations.
  ///
  /// Only the Windows service subsystem will emit these events.
  Resume,

  /// Service subsystem has requested that the services configuration should
  /// be reread.
  ///
  /// On Unixy platforms this is triggered by SIGHUP, and is unsupported on
  /// Windows.
  ReloadConf,

  /// The service application is terminating.  The `Demise` value indicates
  /// the reason for the shutdown.
  Shutdown(Demise)
}


/// The server application runtime type.
// large_enum_variant isn't relevant here because only one instance of this is
// ever created for a process.
#[allow(clippy::large_enum_variant, clippy::module_name_repetitions)]
pub enum SrvAppRt<ApEr> {
  /// A plain non-async (sometimes referred to as "blocking") server
  /// application.
  Sync {
    svcevt_handler: Box<dyn FnMut(SvcEvt) + Send>,
    rt_handler: Box<dyn ServiceHandler<AppErr = ApEr> + Send>
  },

  /// Initializa a tokio runtime, and call each application handler from an
  /// async context.
  #[cfg(feature = "tokio")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tokio")))]
  Tokio {
    rtbldr: Option<runtime::Builder>,
    svcevt_handler: Box<dyn FnMut(SvcEvt) + Send>,
    rt_handler: Box<dyn TokioServiceHandler<AppErr = ApEr> + Send>
  },

  /// Allow Rocket to initialize the tokio runtime, and call each application
  /// handler from an async context.
  #[cfg(feature = "rocket")]
  #[cfg_attr(docsrs, doc(cfg(feature = "rocket")))]
  Rocket {
    svcevt_handler: Box<dyn FnMut(SvcEvt) + Send>,
    rt_handler: Box<dyn RocketServiceHandler<AppErr = ApEr> + Send>
  }
}


/// Service runner context.
pub struct RunCtx {
  service: bool,
  svcname: String,
  log_init: bool,
  test_mode: bool
}

impl RunCtx {
  /// Run as a systemd service.
  #[cfg(all(target_os = "linux", feature = "systemd"))]
  fn systemd<ApEr>(self, st: SrvAppRt<ApEr>) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + std::fmt::Debug
  {
    LumberJack::default().set_init(self.log_init).init()?;

    tracing::debug!("Running service '{}'", self.svcname);

    let sr = Arc::new(systemd::ServiceReporter {});

    let re = RunEnv::Service(Some(self.svcname.clone()));

    match st {
      SrvAppRt::Sync {
        svcevt_handler,
        rt_handler
      } => rttype::sync_main(rttype::SyncMainParams {
        re,
        svcevt_handler,
        rt_handler,
        sr,
        svcevt_ch: None,
        test_mode: self.test_mode
      }),
      SrvAppRt::Tokio {
        rtbldr,
        svcevt_handler,
        rt_handler
      } => rttype::tokio_main(
        rtbldr,
        rttype::TokioMainParams {
          re,
          svcevt_handler,
          rt_handler,
          sr,
          svcevt_ch: None
        }
      ),
      #[cfg(feature = "rocket")]
      SrvAppRt::Rocket {
        svcevt_handler,
        rt_handler
      } => rttype::rocket_main(rttype::RocketMainParams {
        re,
        svcevt_handler,
        rt_handler,
        sr,
        svcevt_ch: None
      })
    }
  }

  /// Run as a Windows service.
  #[cfg(windows)]
  fn winsvc<ApEr>(self, st: SrvAppRt<ApEr>) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + 'static + std::fmt::Debug
  {
    winsvc::run(&self.svcname, st)?;

    Ok(())
  }

  /// Run as a foreground server
  fn foreground<ApEr>(self, st: SrvAppRt<ApEr>) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + std::fmt::Debug
  {
    LumberJack::default().set_init(self.log_init).init()?;

    tracing::debug!("Running service '{}'", self.svcname);

    let sr = Arc::new(nosvc::ServiceReporter {});

    match st {
      SrvAppRt::Sync {
        svcevt_handler,
        rt_handler
      } => rttype::sync_main(rttype::SyncMainParams {
        re: RunEnv::Foreground,
        svcevt_handler,
        rt_handler,
        sr,
        svcevt_ch: None,
        test_mode: self.test_mode
      }),

      #[cfg(feature = "tokio")]
      SrvAppRt::Tokio {
        rtbldr,
        svcevt_handler,
        rt_handler
      } => rttype::tokio_main(
        rtbldr,
        rttype::TokioMainParams {
          re: RunEnv::Foreground,
          svcevt_handler,
          rt_handler,
          sr,
          svcevt_ch: None
        }
      ),

      #[cfg(feature = "rocket")]
      SrvAppRt::Rocket {
        svcevt_handler,
        rt_handler
      } => rttype::rocket_main(rttype::RocketMainParams {
        re: RunEnv::Foreground,
        svcevt_handler,
        rt_handler,
        sr,
        svcevt_ch: None
      })
    }
  }
}

impl RunCtx {
  /// Create a new service running context.
  #[must_use]
  pub fn new(name: &str) -> Self {
    Self {
      service: false,
      svcname: name.into(),
      log_init: true,
      test_mode: false
    }
  }

  /// Enable test mode.
  ///
  /// This method is intended for tests only.
  ///
  /// qsu performs a few global initialization that will fail if run repeatedly
  /// within the same process.  This causes some problem when running tests,
  /// because rust may run tests in threads within the same process.
  #[doc(hidden)]
  #[must_use]
  pub const fn test_mode(mut self) -> Self {
    self.log_init = false;
    self.test_mode = true;
    self
  }

  /// Disable logging/tracing initialization.
  ///
  /// This is useful in tests because tests may run in different threads within
  /// the same process, causing the log/tracing initialization to panic.
  #[doc(hidden)]
  #[must_use]
  pub const fn log_init(mut self, flag: bool) -> Self {
    self.log_init = flag;
    self
  }

  /// Reference version of [`RunCtx::log_init()`].
  #[doc(hidden)]
  pub fn log_init_ref(&mut self, flag: bool) -> &mut Self {
    self.log_init = flag;
    self
  }

  /// Mark this run context to run under the operating system's subservice, if
  /// one is available on this platform.
  #[must_use]
  pub const fn service(mut self) -> Self {
    self.service = true;
    self
  }

  /// Mark this run context to run under the operating system's subservice, if
  /// one is available on this platform.
  pub fn service_ref(&mut self) -> &mut Self {
    self.service = true;
    self
  }

  #[must_use]
  pub const fn is_service(&self) -> bool {
    self.service
  }

  /// Launch the application.
  ///
  /// If this `RunCtx` has been marked as a _service_ then it will perform the
  /// appropriate service subsystem integration before running the actual
  /// server application code.
  ///
  /// This function must only be called from the main thread of the process,
  /// and must be called before any other threads are started.
  ///
  /// # Errors
  /// [`CbErr::App`] is returned, containing application-specific error, if n
  /// application callback returned an error. [`CbErr::Lib`] indicates that an
  /// error occurred in the qsu runtime.
  pub fn run<ApEr>(self, st: SrvAppRt<ApEr>) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + 'static + std::fmt::Debug
  {
    if self.service {
      let _ = RUNAS.set(RunAs::SvcSubsys);

      #[cfg(all(target_os = "linux", feature = "systemd"))]
      self.systemd(st)?;

      #[cfg(windows)]
      self.winsvc(st)?;

      // ToDo: We should check for other platforms here (like macOS/launchd)
    } else {
      let _ = RUNAS.set(RunAs::Foreground);

      // Do not run against any specific service subsystem.  Despite its name
      // this isn't necessarily running as a foreground process; some service
      // subsystems do not make a distinction.  Perhaps a better mental model
      // is that certain service subsystems expects to run regular "foreground"
      // processes.
      self.foreground(st)?;
    }

    Ok(())
  }

  /// Convenience method around [`Self::run()`] using [`SrvAppRt::Sync`].
  #[allow(clippy::missing_errors_doc)]
  pub fn run_sync<ApEr>(
    self,
    svcevt_handler: Box<dyn FnMut(SvcEvt) + Send>,
    rt_handler: Box<dyn ServiceHandler<AppErr = ApEr> + Send>
  ) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + 'static + std::fmt::Debug
  {
    self.run(SrvAppRt::Sync {
      svcevt_handler,
      rt_handler
    })
  }

  /// Convenience method around [`Self::run()`] using [`SrvAppRt::Tokio`].
  #[cfg(feature = "tokio")]
  #[cfg_attr(docsrs, doc(cfg(feature = "tokio")))]
  #[allow(clippy::missing_errors_doc)]
  pub fn run_tokio<ApEr>(
    self,
    rtbldr: Option<runtime::Builder>,
    svcevt_handler: Box<dyn FnMut(SvcEvt) + Send>,
    rt_handler: Box<dyn TokioServiceHandler<AppErr = ApEr> + Send>
  ) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + 'static + std::fmt::Debug
  {
    self.run(SrvAppRt::Tokio {
      rtbldr,
      svcevt_handler,
      rt_handler
    })
  }

  /// Convenience method around [`Self::run()`] using [`SrvAppRt::Rocket`].
  #[cfg(feature = "rocket")]
  #[cfg_attr(docsrs, doc(cfg(feature = "rocket")))]
  #[allow(clippy::missing_errors_doc)]
  pub fn run_rocket<ApEr>(
    self,
    svcevt_handler: Box<dyn FnMut(SvcEvt) + Send>,
    rt_handler: Box<dyn RocketServiceHandler<AppErr = ApEr> + Send>
  ) -> Result<(), CbErr<ApEr>>
  where
    ApEr: Send + 'static + std::fmt::Debug
  {
    self.run(SrvAppRt::Rocket {
      svcevt_handler,
      rt_handler
    })
  }
}

/// Internal thread used to run service event handler.
#[tracing::instrument(name = "svcevtthread", skip_all)]
fn svcevt_thread(
  mut rx: broadcast::Receiver<SvcEvt>,
  mut evt_handler: Box<dyn FnMut(SvcEvt) + Send>
) {
  while let Ok(msg) = rx.blocking_recv() {
    tracing::debug!("Received {:?}", msg);

    #[cfg(all(target_os = "linux", feature = "systemd"))]
    if matches!(msg, SvcEvt::ReloadConf) {
      //
      // Reload has been requested -- report RELOADING=1 and MONOTONIC_USEC to
      // systemd before calling the application callback.
      //
      let ts =
        nix::time::clock_gettime(nix::time::ClockId::CLOCK_MONOTONIC).unwrap();
      let s = format!(
        "RELOADING=1\nMONOTONIC_USEC={}{:06}",
        ts.tv_sec(),
        ts.tv_nsec() / 1000
      );
      tracing::trace!("Sending notification to systemd: {}", s);

      let custom = NotifyState::Custom(&s);
      if let Err(e) = sd_notify::notify(false, &[custom]) {
        log::error!(
          "Unable to send RELOADING=1 notification to systemd; {}",
          e
        );
      }
    }

    //
    // Call the application callback
    //
    evt_handler(msg);

    #[cfg(all(target_os = "linux", feature = "systemd"))]
    if matches!(msg, SvcEvt::ReloadConf) {
      //
      // This is a reload; report READY=1 to systemd after the application
      // callback has been called.
      //
      tracing::trace!("Sending notification to systemd: READY=1");
      if let Err(e) = sd_notify::notify(false, &[NotifyState::Ready]) {
        log::error!("Unable to send READY=1 notification to systemd; {}", e);
      }
    }

    // If the event message was either shutdown or terminate, break out of loop
    // so the thread will terminate
    if let SvcEvt::Shutdown(_) = msg {
      tracing::debug!("Terminating thread");
      // break out of loop when the service shutdown has been rquested
      break;
    }
  }
}

// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :