1/*
  2 * SPDX-FileCopyrightText: 2026 Copyright (c) Contributors to the Eclipse Foundation
  3 *
  4 * See the NOTICE file(s) distributed with this work for additional
  5 * information regarding copyright ownership.
  6 *
  7 * This program and the accompanying materials are made available under the
  8 * terms of the Apache License Version 2.0 which is available at
  9 * https://www.apache.org/licenses/LICENSE-2.0
 10 *
 11 * SPDX-License-Identifier: Apache-2.0
 12 */
 13
 14use std::time::Duration;
 15
 16use cda_health::{HealthState, Status as HealthStatus};
 17use cda_interfaces::spawn_named;
 18use tokio::task::JoinHandle;
 19
 20/// Creates a background task that integrates with the systemd watchdog.
 21///
[docs] 22/// [[ dimpl~system-sd-notify-watchdog-integration, Systemd Watchdog Integration ]]
 23///
 24/// If the process was booted via systemd and the watchdog is enabled,
 25/// this function spawns a task that periodically checks the health state
 26/// and sends the appropriate `sd_notify` notification:
 27///
 28/// - `Ready` when transitioning from `Starting` to `Up`
 29/// - `Watchdog` while healthy (`Up` -> `Up`)
 30/// - `WatchdogTrigger` when health degrades (`Up` -> `Failed`)
 31///
 32/// The notification interval is the systemd-configured watchdog timeout
 33/// minus 5 seconds (to avoid timing issues).
 34///
 35/// If systemd is not detected or the watchdog is not enabled, returns `None`
 36/// and the CDA runs without `sd_notify` behavior.
 37pub fn create_sd_notify_task<F>(
 38    health_state: Option<HealthState>,
 39    shutdown_signal: F,
 40) -> Option<JoinHandle<()>>
 41where
 42    F: Future<Output = ()> + Clone + Send + 'static,
 43{
 44    // map std_notify::booted Err(_) to false and treat as not systemd booted
 45    if !sd_notify::booted().unwrap_or(false) {
 46        tracing::info!("Systemd not detected, skipping sd_notify initialization");
 47        return None;
 48    }
 49    let Some(interval) = determine_interval() else {
 50        tracing::info!("Systemd watchdog not enabled, skipping sd_notify initialization");
 51        return None;
 52    };
 53    // init sd notify task
 54    let watchdog_future = async move {
 55        let mut interval_timer = tokio::time::interval(interval);
 56        let mut state = cda_health::Status::Starting;
 57        loop {
 58            interval_timer.tick().await;
 59            state = trigger_watchdog(health_state.as_ref(), state).await;
 60        }
 61    };
 62    let sd_notify_future = spawn_named!("sd_notify", async move {
 63        tokio::select! {
 64            _ = watchdog_future => {},
 65            () = shutdown_signal => {},
 66        }
 67    });
 68    tracing::info!(
 69        "Systemd detected and watchdog enabled, initialized sd_notify task with interval of {:?} \
 70         seconds",
 71        interval.as_secs()
 72    );
 73    Some(sd_notify_future)
 74}
 75
 76fn determine_interval() -> Option<Duration> {
 77    sd_notify::watchdog_enabled().map(|duration| {
 78        // ensure we are sending a bit more often than the watchdog expects,
 79        // to avoid any timing issues
 80        if let Some(interval) = duration.checked_sub(Duration::from_secs(5)) {
 81            interval.min(Duration::from_secs(1))
 82        } else {
 83            duration
 84        }
 85    })
 86}
 87
 88async fn trigger_watchdog(
 89    health_state: Option<&HealthState>,
 90    prev_state: cda_health::Status,
 91) -> cda_health::Status {
 92    let new_status = fold_health_state(health_state).await;
 93    let notify = match (prev_state, new_status) {
 94        (cda_health::Status::Starting, cda_health::Status::Up) => sd_notify::NotifyState::Ready,
 95        (cda_health::Status::Up, cda_health::Status::Failed) => {
 96            sd_notify::NotifyState::WatchdogTrigger
 97        }
 98        (cda_health::Status::Up, cda_health::Status::Up) => sd_notify::NotifyState::Watchdog,
 99        // this would be Failure -> Starting or Up -> Starting. so makes no sense
100        _ => {
101            tracing::warn!(
102                "Unexpected health status transition from {prev_state:?} to {new_status:?} when \
103                 triggering sd_notify watchdog"
104            );
105            sd_notify::NotifyState::Watchdog
106        }
107    };
108    tracing::debug!("Triggering sd_notify watchdog with status {notify:?}");
109    if let Err(e) = sd_notify::notify(&[notify]) {
110        tracing::warn!(error = %e, "Failed to send sd_notify watchdog notification");
111    }
112    new_status
113}
114
115fn fold_status(acc: HealthStatus, status: HealthStatus) -> HealthStatus {
116    match (acc, status) {
117        (
118            HealthStatus::Up | HealthStatus::Starting | HealthStatus::Pending,
119            HealthStatus::Failed,
120        )
121        | (HealthStatus::Failed, _) => HealthStatus::Failed,
122        (HealthStatus::Starting | HealthStatus::Pending, HealthStatus::Pending) => {
123            HealthStatus::Starting
124        }
125        (HealthStatus::Starting | HealthStatus::Pending, HealthStatus::Up) => HealthStatus::Up,
126        (_, status) => status,
127    }
128}
129
130async fn fold_health_state(health_state: Option<&HealthState>) -> HealthStatus {
131    let Some(health_state) = health_state else {
132        return cda_health::Status::Up;
133    };
134    health_state
135        .query_all_providers()
136        .await
137        .into_values()
138        .fold(cda_health::Status::Starting, fold_status)
139}
140
[docs]141/// [[ test~system-sd-notify-watchdog-integration, Systemd Watchdog Health Aggregation Tests, test ]]
142#[cfg(test)]
143mod tests {
144    use cda_health::Status;
145
146    use super::fold_status;
147
148    #[test]
149    fn fold_up_then_failed_yields_failed() {
150        assert_eq!(fold_status(Status::Up, Status::Failed), Status::Failed);
151    }
152
153    #[test]
154    fn fold_starting_then_failed_yields_failed() {
155        assert_eq!(
156            fold_status(Status::Starting, Status::Failed),
157            Status::Failed
158        );
159    }
160
161    #[test]
162    fn fold_pending_then_failed_yields_failed() {
163        assert_eq!(fold_status(Status::Pending, Status::Failed), Status::Failed);
164    }
165
166    #[test]
167    fn fold_failed_then_up_yields_failed() {
168        assert_eq!(fold_status(Status::Failed, Status::Up), Status::Failed);
169    }
170
171    #[test]
172    fn fold_failed_then_starting_yields_failed() {
173        assert_eq!(
174            fold_status(Status::Failed, Status::Starting),
175            Status::Failed
176        );
177    }
178
179    #[test]
180    fn fold_failed_then_pending_yields_failed() {
181        assert_eq!(fold_status(Status::Failed, Status::Pending), Status::Failed);
182    }
183
184    #[test]
185    fn fold_failed_then_failed_yields_failed() {
186        assert_eq!(fold_status(Status::Failed, Status::Failed), Status::Failed);
187    }
188
189    #[test]
190    fn fold_starting_then_pending_yields_starting() {
191        assert_eq!(
192            fold_status(Status::Starting, Status::Pending),
193            Status::Starting
194        );
195    }
196
197    #[test]
198    fn fold_pending_then_pending_yields_starting() {
199        assert_eq!(
200            fold_status(Status::Pending, Status::Pending),
201            Status::Starting
202        );
203    }
204
205    #[test]
206    fn fold_starting_then_up_yields_up() {
207        assert_eq!(fold_status(Status::Starting, Status::Up), Status::Up);
208    }
209
210    #[test]
211    fn fold_pending_then_up_yields_up() {
212        assert_eq!(fold_status(Status::Pending, Status::Up), Status::Up);
213    }
214
215    #[test]
216    fn fold_up_then_up_yields_up() {
217        assert_eq!(fold_status(Status::Up, Status::Up), Status::Up);
218    }
219
220    #[test]
221    fn fold_up_then_starting_yields_starting() {
222        assert_eq!(fold_status(Status::Up, Status::Starting), Status::Starting);
223    }
224
225    #[test]
226    fn fold_up_then_pending_yields_pending() {
227        assert_eq!(fold_status(Status::Up, Status::Pending), Status::Pending);
228    }
229
230    #[test]
231    fn fold_sequence_all_up_from_starting() {
232        // initial acc = Starting (as fold_health_state does), all providers Up
233        let result = [Status::Up, Status::Up, Status::Up]
234            .into_iter()
235            .fold(Status::Starting, fold_status);
236        assert_eq!(result, Status::Up);
237    }
238
239    #[test]
240    fn fold_sequence_any_failed_dominates() {
241        let result = [Status::Up, Status::Failed, Status::Up]
242            .into_iter()
243            .fold(Status::Starting, fold_status);
244        assert_eq!(result, Status::Failed);
245    }
246
247    #[test]
248    fn fold_sequence_failed_at_start_stays_failed() {
249        let result = [Status::Failed, Status::Up, Status::Up]
250            .into_iter()
251            .fold(Status::Starting, fold_status);
252        assert_eq!(result, Status::Failed);
253    }
254
255    #[test]
256    fn fold_sequence_all_starting_stays_starting() {
257        let result = [Status::Starting, Status::Starting]
258            .into_iter()
259            .fold(Status::Starting, fold_status);
260        assert_eq!(result, Status::Starting);
261    }
262
263    #[test]
264    fn fold_sequence_mix_starting_pending_yields_starting() {
265        let result = [Status::Pending, Status::Starting, Status::Pending]
266            .into_iter()
267            .fold(Status::Starting, fold_status);
268        assert_eq!(result, Status::Starting);
269    }
270
271    #[test]
272    fn fold_sequence_empty_providers_yields_starting() {
273        // no providers -> fold over empty iterator -> initial accumulator = Starting
274        let result = std::iter::empty::<Status>().fold(Status::Starting, fold_status);
275        assert_eq!(result, Status::Starting);
276    }
277
278    #[tokio::test]
279    async fn fold_health_state_none_yields_up() {
280        let result = super::fold_health_state(None).await;
281        assert_eq!(result, Status::Up);
282    }
283}