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}