systemd-transient 0.2.0

//! Tracks a unit's exit.
//!
//! The unit is started with `AddRef=true`, so systemd keeps it loaded until
//! the handle releases it. Its final state therefore stays readable, and the
//! watch can start after the start job: it subscribes to property changes,
//! then reads the current state, so no transition falls between the two.

use futures_util::StreamExt;
use tokio::{sync::watch, task::JoinHandle};
use tracing::warn;
use zbus::{Connection, Variant, fdo::PropertiesProxy, proxy::CacheProperties};

use crate::{UnitExit, UnitStatus, dbus::unit::UnitProxy, error::SystemdTransientError};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum WatchState {
  Running,
  Exited(UnitExit),
  /// The signal stream ended before the unit exited.
  Lost,
}

#[derive(Debug)]
pub(crate) struct ExitWatch {
  rx: watch::Receiver<WatchState>,
  task: JoinHandle<()>,
}

impl ExitWatch {
  /// Must run after the start job finished, while the handle holds its
  /// reference on the unit.
  pub(crate) async fn subscribe(
    conn: &Connection,
    unit: &UnitProxy<'_>,
  ) -> Result<Self, SystemdTransientError> {
    let props = PropertiesProxy::builder(conn)
      .destination(unit.inner().destination().to_owned())
      .path(unit.inner().path().to_owned())
      .cache_properties(CacheProperties::No)
      .build()
      .await?;

    // Subscribe first, read second: a change after the read is in the stream.
    let mut changed = props.receive_properties_changed().await?;
    let initial = unit.active_state().await?;

    let (tx, rx) = watch::channel(match exit_of(&initial) {
      Some(exit) => WatchState::Exited(exit),
      None => WatchState::Running,
    });

    let interface = unit.inner().interface().to_owned();
    let task = tokio::spawn(async move {
      if *tx.borrow() != WatchState::Running {
        return;
      }

      let result = loop {
        let Some(signal) = changed.next().await else {
          break WatchState::Lost;
        };
        let args = match signal.args() {
          Ok(args) => args,
          Err(err) => {
            warn!("unparsable PropertiesChanged signal: {}", err);
            continue;
          }
        };
        if *args.interface_name() != interface {
          continue;
        }

        // systemd sends every emits-change property of the interface in each
        // signal, so coalesced transitions still show the end state.
        if let Some(Variant::Str(s)) = args.changed_properties().get("ActiveState")
          && let Some(exit) = exit_of(s.as_str())
        {
          break WatchState::Exited(exit);
        }
      };

      let _ = tx.send(result);
    });

    Ok(Self { rx, task })
  }

  pub(crate) fn current(&self) -> WatchState {
    *self.rx.borrow()
  }

  pub(crate) async fn wait(&self) -> Result<UnitExit, SystemdTransientError> {
    let mut rx = self.rx.clone();
    let state = *rx
      .wait_for(|s| *s != WatchState::Running)
      .await
      .map_err(|_| SystemdTransientError::UnitStreamEnded)?;

    match state {
      WatchState::Exited(exit) => Ok(exit),
      _ => Err(SystemdTransientError::UnitStreamEnded),
    }
  }
}

impl Drop for ExitWatch {
  fn drop(&mut self) {
    self.task.abort();
  }
}

impl From<WatchState> for UnitStatus {
  fn from(state: WatchState) -> Self {
    match state {
      WatchState::Running => Self::Running,
      WatchState::Exited(exit) => Self::Exited(exit),
      WatchState::Lost => Self::Lost,
    }
  }
}

pub(crate) fn exit_of(state: &str) -> Option<UnitExit> {
  match state {
    "inactive" => Some(UnitExit::Inactive),
    "failed" => Some(UnitExit::Failed),
    _ => None,
  }
}