refactor: systemd timeout handling and change events

Signed-off-by: Jan Zachmann <[email protected]>

Author: JanZachmann

Cargo.lock

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 name = "omnect-device-service"
 readme = "README.md"
 repository = "https://github.com/omnect/omnect-device-service.git"
-version = "0.40.0"
+version = "0.40.1"
 
 [dependencies]
 actix-server = { version = "2.3", default-features = false }

Cargo.toml

@@ -2,12 +2,12 @@ pub mod networkd;
 pub mod unit;
 pub mod watchdog;
 
-use anyhow::{bail, Result};
-use log::{debug, info};
+use anyhow::{Context, Result};
+use log::info;
 use sd_notify::NotifyState;
-use std::{sync::Once, thread, time, time::Duration};
+use std::sync::Once;
 use systemd_zbus::ManagerProxy;
-use tokio::time::{timeout_at, Instant};
+use tokio_stream::StreamExt;
 
 pub fn sd_notify_ready() {
     static SD_NOTIFY_ONCE: Once = Once::new();
@@ -20,11 +20,11 @@ pub fn sd_notify_ready() {
 #[cfg(not(feature = "mock"))]
 pub async fn reboot() -> Result<()> {
     use anyhow::Context;
-    use log::error;
+    use log::{debug, error};
     use std::process::Command;
 
     info!("systemd::reboot");
-    
+
     //journalctl seems not to have a dbus api
     match Command::new("sudo").args(["journalctl", "--sync"]).status() {
         Ok(status) if !status.success() => error!("reboot: failed to execute 'journalctl --sync'"),
@@ -47,48 +47,27 @@ pub async fn reboot() -> Result<()> {
     Ok(())
 }
 
-pub async fn wait_for_system_running(timeout: Duration) -> Result<()> {
-    let begin = Instant::now();
-    let deadline = begin + timeout;
-    let system = timeout_at(deadline, zbus::Connection::system()).await??;
-    let manager = timeout_at(
-        deadline,
-        // here we use manager which explicitly doesn't cache the system state
-        ManagerProxy::builder(&system)
-            .uncached_properties(&["SystemState"])
-            .build(),
-    )
-    .await??;
-
-    loop {
-        // timeout_at doesn't return error/timeout if the future returns immediately, so we have to check.
-        // e.g. manager.system_state() returns immediately if using a caching ManagerProxy
-        if begin.elapsed() >= timeout {
-            bail!("wait_for_system_running timeout occurred");
-        }
-
-        let system_state = timeout_at(deadline, manager.system_state()).await??;
+pub async fn wait_for_system_running() -> Result<()> {
+    let connection = zbus::Connection::system()
+        .await
+        .context("wait_for_system_running: failed to create connection")?;
+    // here we use manager which explicitly doesn't cache the system state
+    let manager = ManagerProxy::builder(&connection)
+        .uncached_properties(&["SystemState"])
+        .build()
+        .await
+        .context("wait_for_system_running: failed to create manager")?;
 
-        match system_state.as_str() {
-            "running" => {
-                debug!("wait_for_system_running: system_state == running");
-                return Ok(());
-            }
-            "initializing" | "starting" => {
-                /*
-                 * ToDo https://github.com/omnect/omnect-device-service/pull/39#discussion_r1142147564
-                 * This is tricky because you have to receive system state signals, which blocks if you
-                 * are already in state "running". if you receive the system state on condition after
-                 * you got state "starting" by polling, there would by a race which can result in a
-                 * deadlock of receiving the state signal again.
-                 * So a solution would be to poll, start signal receiving, poll again and stop
-                 * possibly stop receiving.
-                 */
-                thread::sleep(time::Duration::from_millis(100));
-            }
-            _ => bail!("system in error state: \"{system_state}\""),
-        }
+    if manager.system_state().await? != "running" {
+        manager
+            .receive_system_state_changed()
+            .await
+            .filter(|p| p.name() == "running")
+            .next()
+            .await;
     }
+
+    Ok(())
 }
 
 #[cfg(feature = "mock")]

src/systemd/mod.rs

@@ -1,16 +1,6 @@
 use anyhow::Result;
-use std::time::Duration;
 pub use systemd_zbus::Mode;
 
-#[cfg(not(any(feature = "mock", test)))]
-use {
-    anyhow::{ensure, Context},
-    futures_util::{join, StreamExt},
-    log::debug,
-    systemd_zbus::ManagerProxy,
-    tokio::time::{timeout_at, Instant},
-};
-
 #[derive(Copy, Clone, Debug)]
 pub enum UnitAction {
     Reload,
@@ -20,17 +10,24 @@ pub enum UnitAction {
 }
 
 #[cfg(not(feature = "mock"))]
-pub async fn unit_action(
-    unit: &str,
-    unit_action: UnitAction,
-    timeout: Duration,
-    mode: Mode,
-) -> Result<()> {
-    let deadline = Instant::now() + timeout;
-    let system = timeout_at(deadline, zbus::Connection::system()).await??;
-    let manager = timeout_at(deadline, ManagerProxy::new(&system)).await??;
-
-    let mut job_removed_stream = timeout_at(deadline, manager.receive_job_removed()).await??;
+pub async fn unit_action(unit: &str, unit_action: UnitAction, mode: Mode) -> Result<()> {
+    use anyhow::Context;
+    use log::debug;
+    use systemd_zbus::ManagerProxy;
+    use tokio_stream::StreamExt;
+
+    debug!("unit_action: {unit} {unit_action:?} {mode:?}");
+
+    let connection = zbus::Connection::system()
+        .await
+        .context("failed to create connection")?;
+    let manager: ManagerProxy<'_> = ManagerProxy::new(&connection)
+        .await
+        .context("failed to create manager")?;
+    let job_removed_stream = manager
+        .receive_job_removed()
+        .await
+        .context("failed to create job_removed_stream")?;
 
     let action = |&unit_action| async move {
         match unit_action {
@@ -41,55 +38,25 @@ pub async fn unit_action(
         }
     };
 
-    let (job_removed, job) = join!(
-        timeout_at(deadline, job_removed_stream.next()),
-        timeout_at(deadline, action(&unit_action))
-    );
-
-    let job = job
-        .context(format!("systemd unit {unit_action:?} \"{unit}\" failed"))??
+    let job = action(&unit_action)
+        .await
+        .context("unit action failed")?
         .into_inner();
 
-    let job_removed = job_removed
-        .context("systemd job_removed_stream")?
-        .context("failed to get next item in job removed stream")?;
-
-    let job_removed_args = job_removed.args().context("get removed args")?;
+    job_removed_stream
+        .filter(|job_removed| {
+            if let Ok(args) = job_removed.args() {
+                return args.job() == &job && args.result == "done";
+            };
+            false
+        })
+        .next()
+        .await;
 
-    debug!("job removed: {job_removed_args:?}");
-    if job_removed_args.job() == &job {
-        ensure!(
-            job_removed_args.result == "done",
-            "systemd (1) unit {unit_action:?} \"{unit}\" isn't done: {}",
-            job_removed_args.result
-        );
-        return Ok(());
-    }
-
-    loop {
-        let job_removed = timeout_at(deadline, job_removed_stream.next())
-            .await?
-            .context("no job_removed signal received")?;
-
-        let job_removed_args = job_removed.args()?;
-        debug!("job removed: {job_removed_args:?}");
-        if job_removed_args.job() == &job {
-            ensure!(
-                job_removed_args.result == "done",
-                "systemd (2) unit {unit_action:?} \"{unit}\" isn't done: {}",
-                job_removed_args.result
-            );
-            return Ok(());
-        }
-    }
+    Ok(())
 }
 
 #[cfg(feature = "mock")]
-pub async fn unit_action(
-    _unit: &str,
-    _unit_action: UnitAction,
-    _timeout: Duration,
-    _mode: Mode,
-) -> Result<()> {
+pub async fn unit_action(_unit: &str, _unit_action: UnitAction, _mode: Mode) -> Result<()> {
     Ok(())
 }

src/systemd/unit.rs

@@ -28,7 +28,6 @@ use update_validation::UpdateValidation;
 
 static LOAD_UPDATE_WDT_INTERVAL_SECS: u64 = 120;
 static RUN_UPDATE_WDT_INTERVAL_SECS: u64 = 600;
-static UNIT_ACTION_TIMEOUT_SECS: u64 = 30;
 
 struct LoadUpdateGuard {
     wdt: Option<Duration>,
@@ -77,15 +76,12 @@ impl RunUpdateGuard {
         systemd::unit::unit_action(
             IOT_HUB_DEVICE_UPDATE_SERVICE_TIMER,
             UnitAction::Stop,
-            Duration::from_secs(UNIT_ACTION_TIMEOUT_SECS),
             systemd_zbus::Mode::Replace,
         )
         .await?;
-
         systemd::unit::unit_action(
             IOT_HUB_DEVICE_UPDATE_SERVICE,
             UnitAction::Stop,
-            Duration::from_secs(UNIT_ACTION_TIMEOUT_SECS),
             systemd_zbus::Mode::Replace,
         )
         .await?;
@@ -117,7 +113,6 @@ impl Drop for RunUpdateGuard {
                 if let Err(e) = systemd::unit::unit_action(
                     IOT_HUB_DEVICE_UPDATE_SERVICE,
                     UnitAction::Start,
-                    Duration::from_secs(UNIT_ACTION_TIMEOUT_SECS),
                     systemd_zbus::Mode::Fail,
                 )
                 .await
@@ -128,7 +123,6 @@ impl Drop for RunUpdateGuard {
                 if let Err(e) = systemd::unit::unit_action(
                     IOT_HUB_DEVICE_UPDATE_SERVICE_TIMER,
                     UnitAction::Start,
-                    Duration::from_secs(UNIT_ACTION_TIMEOUT_SECS),
                     systemd_zbus::Mode::Fail,
                 )
                 .await

src/twin/firmware_update/mod.rs

@@ -165,11 +165,9 @@ impl UpdateValidation {
 
     async fn validate(&mut self) -> Result<()> {
         debug!("started");
-        let now = Duration::from(nix::time::clock_gettime(
-            nix::time::ClockId::CLOCK_MONOTONIC,
-        )?);
-        let timeout = self.validation_timeout - (now - self.start_monotonic_time);
-        systemd::wait_for_system_running(timeout).await?;
+        systemd::wait_for_system_running()
+            .await
+            .context("validate: wait_for_system_running timed out")?;
 
         /* ToDo: if it returns with an error, we may want to handle the state
          * "degraded" and possibly ignore certain failed services via configuration
@@ -180,19 +178,13 @@ impl UpdateValidation {
         debug!("starting {IOT_HUB_DEVICE_UPDATE_SERVICE}");
         fs::remove_file(UPDATE_VALIDATION_FILE).context("remove UPDATE_VALIDATION_FILE")?;
 
-        let now = Duration::from(nix::time::clock_gettime(
-            nix::time::ClockId::CLOCK_MONOTONIC,
-        )?);
-        let timeout = self.validation_timeout - (now - self.start_monotonic_time);
-
         // in case of local update we don't take care of starting deviceupdate-agent.service,
         // since it might fail because of missing iothub connection.
         // instead we let deviceupdate-agent.timer doing the job periodically
         if !self.local_update {
             systemd::unit::unit_action(
                 IOT_HUB_DEVICE_UPDATE_SERVICE,
                 UnitAction::Start,
-                timeout,
                 systemd_zbus::Mode::Fail,
             )
             .await?;

src/twin/firmware_update/update_validation.rs

@@ -23,7 +23,6 @@ lazy_static! {
 }
 
 static NETWORK_SERVICE: &str = "systemd-networkd.service";
-static NETWORK_SERVICE_RELOAD_TIMEOUT_IN_SECS: u64 = 15;
 
 #[derive(PartialEq, Serialize)]
 pub struct Address {
@@ -93,7 +92,6 @@ impl Feature for Network {
                 unit::unit_action(
                     NETWORK_SERVICE,
                     unit::UnitAction::Reload,
-                    Duration::from_secs(NETWORK_SERVICE_RELOAD_TIMEOUT_IN_SECS),
                     systemd_zbus::Mode::Fail,
                 )
                 .await?