asched_core/routine/
client.rs

1//! Supported application boundary for routine daemon requests and startup.
2
3use super::ipc::{self, Action, Request, Response};
4use super::{FireOutcome, RoutineError, PROTOCOL_VERSION};
5use serde_json::Value;
6use std::io::Read;
7use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
8use std::os::unix::process::CommandExt;
9use std::path::{Path, PathBuf};
10use std::process::{Child, Command};
11use std::time::{Duration, Instant};
12
13pub const STARTUP_FD_ENV: &str = "ASCHED_STARTUP_FD";
14const DEFAULT_STARTUP_TIMEOUT: Duration = Duration::from_secs(3);
15const POLL_INTERVAL: Duration = Duration::from_millis(50);
16
17/// Client for a routine daemon rooted at one machine-local state directory.
18///
19/// [`request`](Self::request) never starts a daemon. [`start`](Self::start) ensures
20/// a daemon is running without accessing a project. [`request_with_start`](Self::request_with_start)
21/// accepts routine list and mutation requests, probes daemon availability with
22/// a status request, then starts the caller-provided command only when the daemon
23/// is unavailable. Status and shutdown requests are rejected by the auto-start
24/// boundary and must use [`request`](Self::request). The caller's request is sent
25/// exactly once. Availability probing and startup share the configured startup
26/// timeout. The command must run a routine daemon that
27/// writes `ready` or `error:<message>` to the descriptor supplied in the
28/// `ASCHED_STARTUP_FD` environment variable. The client adds that
29/// handshake and a detached process group; executable, arguments, standard
30/// streams, and other environment are controlled by the caller.
31// ^ [[Shared Daemon and Versioned IPC]]
32#[derive(Debug, Clone)]
33pub struct RoutineClient {
34    root: PathBuf,
35    startup_timeout: Duration,
36    request_timeout: Duration,
37}
38
39impl RoutineClient {
40    pub fn new(root: PathBuf) -> Self {
41        Self {
42            root,
43            startup_timeout: DEFAULT_STARTUP_TIMEOUT,
44            request_timeout: Duration::from_secs(30),
45        }
46    }
47
48    /// Override the bounded startup timeout, primarily for hosts and tests.
49    pub fn with_startup_timeout(mut self, timeout: Duration) -> Self {
50        self.startup_timeout = timeout;
51        self
52    }
53
54    /// Override the timeout for one request to an already-running daemon.
55    pub fn with_request_timeout(mut self, timeout: Duration) -> Self {
56        self.request_timeout = timeout;
57        self
58    }
59
60    pub fn root(&self) -> &Path {
61        &self.root
62    }
63
64    pub fn socket_path(&self) -> PathBuf {
65        self.root.join("daemon-v1.sock")
66    }
67
68    pub fn fire(
69        &self,
70        project: &Path,
71        kind: &str,
72        payload: Value,
73        event_id: &str,
74    ) -> Result<FireOutcome, RoutineError> {
75        let response = self.request(&Request::new(
76            project.to_path_buf(),
77            Action::Fire {
78                kind: kind.to_string(),
79                payload,
80                event_id: event_id.to_string(),
81            },
82        ))?;
83        match response {
84            Response::Fire { outcome } => Ok(outcome),
85            _ => Err(RoutineError::Corrupt(
86                "routine daemon returned a non-fire response".into(),
87            )),
88        }
89    }
90
91    /// Send to an already-running daemon without changing daemon lifecycle.
92    pub fn request(&self, request: &Request) -> Result<Response, RoutineError> {
93        ipc::send_with_timeout(&self.socket_path(), request, self.request_timeout)?.into_result()
94    }
95
96    /// Ensure a daemon is running without issuing a project-scoped request.
97    pub fn start(&self, mut command: Command) -> Result<(), RoutineError> {
98        let status = Request::new(PathBuf::new(), Action::Status);
99        let deadline = self.startup_deadline()?;
100        self.ensure_started(&status, &mut command, deadline)
101    }
102
103    /// Send a request once, starting the daemon with `command` if it is unavailable.
104    pub fn request_with_start(
105        &self,
106        request: &Request,
107        mut command: Command,
108    ) -> Result<Response, RoutineError> {
109        if matches!(&request.action, Action::Status | Action::Shutdown) {
110            return Err(RoutineError::Validation(
111                "status and shutdown requests cannot auto-start the routine daemon".into(),
112            ));
113        }
114
115        let deadline = self.startup_deadline()?;
116        // ^ Probe with an observation so a mutation is never retried after an ambiguous disconnect.
117        let status = Request::new(request.project.clone(), Action::Status);
118        self.ensure_started(&status, &mut command, deadline)?;
119        self.request(request)
120    }
121
122    fn startup_deadline(&self) -> Result<Instant, RoutineError> {
123        Instant::now()
124            .checked_add(self.startup_timeout)
125            .ok_or_else(|| {
126                RoutineError::Validation("routine daemon startup timeout is too large".into())
127            })
128    }
129
130    fn ensure_started(
131        &self,
132        status: &Request,
133        command: &mut Command,
134        deadline: Instant,
135    ) -> Result<(), RoutineError> {
136        let remaining = deadline.saturating_duration_since(Instant::now());
137        if remaining.is_zero() {
138            return Err(self.startup_timeout_error());
139        }
140        match ipc::send_with_timeout(&self.socket_path(), status, remaining)
141            .and_then(Response::into_result)
142        {
143            Ok(response) => {
144                validate_status_response(response)?;
145                return Ok(());
146            }
147            Err(RoutineError::Unavailable(_)) => {}
148            Err(error) => return Err(error),
149        }
150
151        self.spawn_and_await(status, command, deadline)
152    }
153
154    fn spawn_and_await(
155        &self,
156        status: &Request,
157        command: &mut Command,
158        deadline: Instant,
159    ) -> Result<(), RoutineError> {
160        if Instant::now() >= deadline {
161            return Err(self.startup_timeout_error());
162        }
163        let (read_end, write_end) = startup_pipe()?;
164        let read_fd = read_end.as_raw_fd();
165        let write_fd = write_end.as_raw_fd();
166        command.env(STARTUP_FD_ENV, write_fd.to_string());
167        unsafe {
168            command.pre_exec(move || {
169                libc::close(read_fd);
170                if libc::setsid() == -1 {
171                    return Err(std::io::Error::last_os_error());
172                }
173                Ok(())
174            });
175        }
176        let mut child = command.spawn()?;
177        drop(write_end);
178
179        let mut read_end = std::fs::File::from(read_end);
180        let result = self.await_startup(status, &mut child, &mut read_end, deadline);
181        if result.is_err() {
182            stop_startup_child(&mut child);
183        }
184        result
185    }
186
187    fn await_startup(
188        &self,
189        status: &Request,
190        child: &mut Child,
191        read_end: &mut std::fs::File,
192        deadline: Instant,
193    ) -> Result<(), RoutineError> {
194        while Instant::now() < deadline {
195            if fd_readable(read_end.as_raw_fd())? {
196                // ^ Concurrent spawns may inherit peer pipe writers, so handshake reads cannot wait for EOF.
197                let mut startup = [0_u8; 4096];
198                let bytes = read_end.read(&mut startup)?;
199                let startup = std::str::from_utf8(&startup[..bytes]).map_err(|_| {
200                    RoutineError::Corrupt(
201                        "routine daemon returned non-UTF-8 startup response".into(),
202                    )
203                })?;
204                if startup == "ready" {
205                    return self
206                        .poll_ready(status, deadline)?
207                        .then_some(())
208                        .ok_or_else(|| {
209                            RoutineError::Unavailable(
210                                "routine daemon reported ready but did not accept status requests"
211                                    .into(),
212                            )
213                        });
214                }
215                if let Some(error) = startup.strip_prefix("error:") {
216                    if self.poll_ready(status, deadline)? {
217                        // ^ A singleton-race loser has failed even though its peer is ready.
218                        stop_startup_child(child);
219                        return Ok(());
220                    }
221                    return Err(RoutineError::Unavailable(format!(
222                        "routine daemon failed to start: {error}"
223                    )));
224                }
225                if startup.is_empty() {
226                    if let Some(exit) = child.try_wait()? {
227                        return Err(RoutineError::Unavailable(format!(
228                            "routine daemon exited during startup: {exit}"
229                        )));
230                    }
231                }
232                return Err(RoutineError::Corrupt(
233                    "routine daemon returned an invalid startup response".into(),
234                ));
235            }
236            if let Some(exit) = child.try_wait()? {
237                return Err(RoutineError::Unavailable(format!(
238                    "routine daemon exited during startup: {exit}"
239                )));
240            }
241            std::thread::sleep(POLL_INTERVAL);
242        }
243        Err(self.startup_timeout_error())
244    }
245
246    fn poll_ready(&self, status: &Request, deadline: Instant) -> Result<bool, RoutineError> {
247        while Instant::now() < deadline {
248            let remaining = deadline.saturating_duration_since(Instant::now());
249            if remaining.is_zero() {
250                return Ok(false);
251            }
252            match ipc::send_with_timeout(&self.socket_path(), status, remaining)
253                .and_then(Response::into_result)
254            {
255                Ok(response) => {
256                    validate_status_response(response)?;
257                    return Ok(true);
258                }
259                Err(RoutineError::Unavailable(_)) => {}
260                Err(error) => return Err(error),
261            }
262            std::thread::sleep(POLL_INTERVAL);
263        }
264        Ok(false)
265    }
266
267    fn startup_timeout_error(&self) -> RoutineError {
268        RoutineError::Unavailable(format!(
269            "routine daemon did not become ready within {} ms",
270            self.startup_timeout.as_millis()
271        ))
272    }
273}
274
275fn validate_status_response(response: Response) -> Result<(), RoutineError> {
276    match response {
277        Response::Daemon { protocol, .. } if protocol == PROTOCOL_VERSION => Ok(()),
278        Response::Daemon { protocol, .. } => Err(RoutineError::ProtocolMismatch {
279            client: PROTOCOL_VERSION,
280            daemon: protocol,
281        }),
282        _ => Err(RoutineError::Corrupt(
283            "routine daemon returned a non-daemon response to a status request".into(),
284        )),
285    }
286}
287
288fn startup_pipe() -> Result<(OwnedFd, OwnedFd), RoutineError> {
289    let mut fds = [0; 2];
290    if unsafe { libc::pipe(fds.as_mut_ptr()) } != 0 {
291        return Err(std::io::Error::last_os_error().into());
292    }
293    unsafe { Ok((OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1]))) }
294}
295
296fn fd_readable(fd: i32) -> Result<bool, RoutineError> {
297    let mut descriptor = libc::pollfd {
298        fd,
299        events: libc::POLLIN | libc::POLLHUP,
300        revents: 0,
301    };
302    let result = unsafe { libc::poll(&mut descriptor, 1, 0) };
303    if result < 0 {
304        return Err(std::io::Error::last_os_error().into());
305    }
306    Ok(result > 0)
307}
308
309fn stop_startup_child(child: &mut Child) {
310    if matches!(child.try_wait(), Ok(Some(_))) {
311        return;
312    }
313    unsafe {
314        libc::kill(-(child.id() as i32), libc::SIGKILL);
315    }
316    let _ = child.wait();
317}
318
319#[cfg(test)]
320mod tests {
321    use super::*;
322    use std::io::Write;
323    use std::os::fd::FromRawFd;
324    use std::sync::atomic::{AtomicU64, Ordering};
325    use std::sync::{Arc, Barrier};
326
327    static NEXT: AtomicU64 = AtomicU64::new(0);
328
329    fn test_root(label: &str) -> PathBuf {
330        PathBuf::from(".tmp").join(format!(
331            "routine-client-{label}-{}-{}",
332            std::process::id(),
333            NEXT.fetch_add(1, Ordering::Relaxed)
334        ))
335    }
336
337    fn helper_command(root: &Path, mode: &str) -> Command {
338        let mut command = Command::new(std::env::current_exe().unwrap());
339        command
340            .arg("--exact")
341            .arg("routine::client::tests::daemon_helper")
342            .arg("--ignored")
343            .arg("--nocapture")
344            .env("ASCHED_TEST_ROOT", root)
345            .env("ASCHED_TEST_MODE", mode);
346        command
347    }
348
349    fn status(root: &Path) -> Request {
350        Request::new(root.to_path_buf(), Action::Status)
351    }
352
353    fn list(root: &Path) -> Request {
354        Request::new(root.to_path_buf(), Action::List)
355    }
356
357    #[test]
358    fn direct_request_does_not_start_an_absent_daemon() {
359        let root = test_root("direct");
360        let result = RoutineClient::new(root.clone()).request(&status(&root));
361        assert!(matches!(result, Err(RoutineError::Unavailable(_))));
362        assert!(!root.exists());
363    }
364
365    #[test]
366    fn given_accepting_socket_without_response_when_requested_then_configured_timeout_is_honored() {
367        use std::os::unix::net::UnixListener;
368
369        let root = test_root("request-timeout");
370        std::fs::create_dir_all(&root).unwrap();
371        let listener = UnixListener::bind(root.join("daemon-v1.sock")).unwrap();
372        let (accepted_tx, accepted_rx) = std::sync::mpsc::channel();
373        let (release_tx, release_rx) = std::sync::mpsc::channel();
374        let server = std::thread::spawn(move || {
375            let (_stream, _) = listener.accept().unwrap();
376            let _ = accepted_tx.send(());
377            let _ = release_rx.recv_timeout(Duration::from_secs(1));
378        });
379        let timeout = Duration::from_millis(75);
380        let started = Instant::now();
381
382        let result = RoutineClient::new(root.clone())
383            .with_request_timeout(timeout)
384            .request(&status(&root));
385        let elapsed = started.elapsed();
386        let accepted = accepted_rx.recv_timeout(Duration::from_secs(1)).is_ok();
387        let _ = release_tx.send(());
388        let joined = server.join().is_ok();
389        let _ = std::fs::remove_dir_all(root);
390
391        assert_eq!(
392            (
393                matches!(result, Err(RoutineError::Unavailable(_))),
394                accepted,
395                elapsed >= Duration::from_millis(50),
396                elapsed < Duration::from_millis(500),
397                joined,
398            ),
399            (true, true, true, true, true)
400        );
401    }
402
403    #[test]
404    fn lifecycle_requests_cannot_use_auto_start() {
405        for action in [Action::Status, Action::Shutdown] {
406            let root = test_root("lifecycle");
407            let request = Request::new(root.clone(), action);
408            let result = RoutineClient::new(root.clone())
409                .request_with_start(&request, helper_command(&root, "must_not_start"));
410
411            assert!(matches!(result, Err(RoutineError::Validation(_))));
412            assert!(!root.exists());
413        }
414    }
415
416    #[test]
417    fn auto_start_serves_list_and_direct_shutdown_stops_daemon() {
418        let root = test_root("start");
419        let client = RoutineClient::new(root.clone());
420        let response = client
421            .request_with_start(&list(&root), helper_command(&root, "daemon"))
422            .unwrap();
423        assert!(matches!(response, Response::Routines { .. }));
424        client
425            .request(&Request::new(root.clone(), Action::Shutdown))
426            .unwrap();
427        let _ = std::fs::remove_dir_all(root);
428    }
429
430    #[test]
431    fn concurrent_auto_starts_converge_on_one_daemon() {
432        let root = test_root("race");
433        let barrier = Arc::new(Barrier::new(3));
434        let mut callers = Vec::new();
435        for _ in 0..2 {
436            let root = root.clone();
437            let barrier = barrier.clone();
438            callers.push(std::thread::spawn(move || {
439                barrier.wait();
440                RoutineClient::new(root.clone())
441                    .request_with_start(&list(&root), helper_command(&root, "daemon"))
442            }));
443        }
444        barrier.wait();
445        for caller in callers {
446            assert!(matches!(
447                caller.join().unwrap(),
448                Ok(Response::Routines { .. })
449            ));
450        }
451        RoutineClient::new(root.clone())
452            .request(&Request::new(root.clone(), Action::Shutdown))
453            .unwrap();
454        let _ = std::fs::remove_dir_all(root);
455    }
456
457    #[test]
458    fn singleton_race_reaps_the_losing_startup_child() {
459        let root = test_root("race-reap");
460        let barrier = Arc::new(Barrier::new(3));
461        let mut callers = Vec::new();
462        for _ in 0..2 {
463            let root = root.clone();
464            let barrier = barrier.clone();
465            callers.push(std::thread::spawn(move || {
466                let client = RoutineClient::new(root.clone());
467                let mut command = helper_command(&root, "daemon_with_pid");
468                barrier.wait();
469                let deadline = client.startup_deadline().unwrap();
470                client.spawn_and_await(&status(&root), &mut command, deadline)
471            }));
472        }
473        barrier.wait();
474        for caller in callers {
475            caller.join().unwrap().unwrap();
476        }
477
478        let winner_pid = match RoutineClient::new(root.clone())
479            .request(&status(&root))
480            .unwrap()
481        {
482            Response::Daemon { pid, .. } => pid,
483            response => panic!("expected daemon response, got {response:?}"),
484        };
485        let helper_pids: Vec<u32> = std::fs::read_dir(&root)
486            .unwrap()
487            .filter_map(|entry| {
488                let name = entry.ok()?.file_name();
489                name.to_str()?
490                    .strip_prefix("helper-")?
491                    .strip_suffix(".pid")?
492                    .parse()
493                    .ok()
494            })
495            .collect();
496        assert_eq!(helper_pids.len(), 2);
497        let loser_pid = *helper_pids.iter().find(|&&pid| pid != winner_pid).unwrap();
498        assert_eq!(unsafe { libc::kill(loser_pid as i32, 0) }, -1);
499        assert_eq!(
500            std::io::Error::last_os_error().raw_os_error(),
501            Some(libc::ESRCH)
502        );
503
504        RoutineClient::new(root.clone())
505            .request(&Request::new(root.clone(), Action::Shutdown))
506            .unwrap();
507        let _ = std::fs::remove_dir_all(root);
508    }
509
510    #[test]
511    fn startup_timeout_kills_and_reaps_the_spawned_process() {
512        let root = test_root("timeout");
513        std::fs::create_dir_all(&root).unwrap();
514        let pid_path = root.join("helper.pid");
515        let result = RoutineClient::new(root.clone())
516            .with_startup_timeout(Duration::from_millis(300))
517            .request_with_start(&list(&root), helper_command(&root, "hang"));
518        assert!(matches!(result, Err(RoutineError::Unavailable(_))));
519        let pid: i32 = std::fs::read_to_string(&pid_path).unwrap().parse().unwrap();
520        assert_eq!(unsafe { libc::kill(pid, 0) }, -1);
521        assert_eq!(
522            std::io::Error::last_os_error().raw_os_error(),
523            Some(libc::ESRCH)
524        );
525        let _ = std::fs::remove_dir_all(root);
526    }
527
528    #[test]
529    fn initial_availability_probe_respects_startup_timeout() {
530        use std::io::{BufRead, BufReader};
531        use std::os::unix::net::UnixListener;
532
533        let root = test_root("probe-timeout");
534        std::fs::create_dir_all(&root).unwrap();
535        let listener = UnixListener::bind(root.join("daemon-v1.sock")).unwrap();
536        let server = std::thread::spawn(move || {
537            let (stream, _) = listener.accept().unwrap();
538            let mut request = String::new();
539            BufReader::new(stream).read_line(&mut request).unwrap();
540            std::thread::sleep(Duration::from_secs(1));
541        });
542        let started = Instant::now();
543        let result = RoutineClient::new(root.clone())
544            .with_startup_timeout(Duration::from_millis(300))
545            .request_with_start(&list(&root), helper_command(&root, "must_not_start"));
546
547        assert!(matches!(result, Err(RoutineError::Unavailable(_))));
548        assert!(started.elapsed() < Duration::from_secs(1));
549        let pid_path = root.join("helper.pid");
550        // ^ Socket read timeouts may return just before the deadline, allowing a promptly reaped spawn.
551        if pid_path.exists() {
552            assert_helper_reaped(&pid_path);
553        }
554        server.join().unwrap();
555        let _ = std::fs::remove_dir_all(root);
556    }
557
558    #[test]
559    fn explicit_start_does_not_issue_a_project_scoped_request() {
560        let root = test_root("explicit-start");
561        let client = RoutineClient::new(root.clone());
562
563        client.start(helper_command(&root, "daemon")).unwrap();
564        assert!(matches!(
565            client.request(&status(&root)).unwrap(),
566            Response::Daemon { .. }
567        ));
568        client
569            .request(&Request::new(root.clone(), Action::Shutdown))
570            .unwrap();
571        let _ = std::fs::remove_dir_all(root);
572    }
573
574    #[test]
575    fn unrepresentable_startup_timeout_is_rejected_without_spawning() {
576        let root = test_root("unrepresentable-timeout");
577        let result = RoutineClient::new(root.clone())
578            .with_startup_timeout(Duration::MAX)
579            .request_with_start(&list(&root), helper_command(&root, "must_not_start"));
580        assert!(matches!(result, Err(RoutineError::Validation(_))));
581        assert!(!root.join("helper.pid").exists());
582    }
583
584    #[test]
585    fn invalid_transport_response_does_not_trigger_a_start() {
586        use std::io::{BufRead, BufReader};
587        use std::os::unix::net::UnixListener;
588
589        let root = test_root("transport");
590        std::fs::create_dir_all(&root).unwrap();
591        let listener = UnixListener::bind(root.join("daemon-v1.sock")).unwrap();
592        let server = std::thread::spawn(move || {
593            let (mut stream, _) = listener.accept().unwrap();
594            let mut request = String::new();
595            BufReader::new(stream.try_clone().unwrap())
596                .read_line(&mut request)
597                .unwrap();
598            stream.write_all(b"not-json\n").unwrap();
599        });
600        let result = RoutineClient::new(root.clone())
601            .request_with_start(&list(&root), helper_command(&root, "must_not_start"));
602        assert!(matches!(result, Err(RoutineError::Corrupt(_))));
603        server.join().unwrap();
604        assert!(!root.join("helper.pid").exists());
605        let _ = std::fs::remove_dir_all(root);
606    }
607
608    #[test]
609    fn non_daemon_status_response_does_not_trigger_a_start() {
610        use std::io::{BufRead, BufReader};
611        use std::os::unix::net::UnixListener;
612
613        let root = test_root("wrong-status-response");
614        std::fs::create_dir_all(&root).unwrap();
615        let listener = UnixListener::bind(root.join("daemon-v1.sock")).unwrap();
616        let server = std::thread::spawn(move || {
617            let (mut stream, _) = listener.accept().unwrap();
618            let mut request = String::new();
619            BufReader::new(stream.try_clone().unwrap())
620                .read_line(&mut request)
621                .unwrap();
622            stream
623                .write_all(b"{\"result\":\"ok\",\"revision\":null}\n")
624                .unwrap();
625        });
626        let result = RoutineClient::new(root.clone())
627            .request_with_start(&list(&root), helper_command(&root, "must_not_start"));
628        assert!(matches!(result, Err(RoutineError::Corrupt(_))));
629        server.join().unwrap();
630        assert!(!root.join("helper.pid").exists());
631        let _ = std::fs::remove_dir_all(root);
632    }
633
634    #[test]
635    fn mismatched_status_protocol_does_not_trigger_a_start() {
636        use std::io::{BufRead, BufReader};
637        use std::os::unix::net::UnixListener;
638
639        let root = test_root("wrong-status-protocol");
640        std::fs::create_dir_all(&root).unwrap();
641        let listener = UnixListener::bind(root.join("daemon-v1.sock")).unwrap();
642        let server = std::thread::spawn(move || {
643            let (mut stream, _) = listener.accept().unwrap();
644            let mut request = String::new();
645            BufReader::new(stream.try_clone().unwrap())
646                .read_line(&mut request)
647                .unwrap();
648            writeln!(
649                stream,
650                "{{\"result\":\"daemon\",\"protocol\":{},\"pid\":1}}",
651                PROTOCOL_VERSION + 1
652            )
653            .unwrap();
654        });
655        let result = RoutineClient::new(root.clone())
656            .request_with_start(&list(&root), helper_command(&root, "must_not_start"));
657        assert!(matches!(
658            result,
659            Err(RoutineError::ProtocolMismatch {
660                client: PROTOCOL_VERSION,
661                daemon,
662            }) if daemon == PROTOCOL_VERSION + 1
663        ));
664        server.join().unwrap();
665        assert!(!root.join("helper.pid").exists());
666        let _ = std::fs::remove_dir_all(root);
667    }
668
669    #[test]
670    fn ambiguous_mutation_disconnect_is_not_retried_or_auto_started() {
671        use std::io::{BufRead, BufReader};
672        use std::os::unix::net::UnixListener;
673
674        let root = test_root("ambiguous-mutation");
675        std::fs::create_dir_all(&root).unwrap();
676        let listener = UnixListener::bind(root.join("daemon-v1.sock")).unwrap();
677        let server = std::thread::spawn(move || {
678            let (mut probe, _) = listener.accept().unwrap();
679            let mut probe_request = String::new();
680            BufReader::new(probe.try_clone().unwrap())
681                .read_line(&mut probe_request)
682                .unwrap();
683            let probe_request: Request = serde_json::from_str(&probe_request).unwrap();
684            assert!(matches!(probe_request.action, Action::Status));
685            writeln!(
686                probe,
687                "{{\"result\":\"daemon\",\"protocol\":{PROTOCOL_VERSION},\"pid\":1}}"
688            )
689            .unwrap();
690
691            let (mutation, _) = listener.accept().unwrap();
692            let mut mutation_request = String::new();
693            let mut mutation = BufReader::new(mutation);
694            mutation.read_line(&mut mutation_request).unwrap();
695            let mutation_request: Request = serde_json::from_str(&mutation_request).unwrap();
696            assert!(matches!(mutation_request.action, Action::Delete { .. }));
697            mutation
698                .get_ref()
699                .shutdown(std::net::Shutdown::Write)
700                .unwrap();
701        });
702        let request = Request::new(
703            root.clone(),
704            Action::Delete {
705                revision: 1,
706                name: "daily".into(),
707            },
708        );
709        let result = RoutineClient::new(root.clone())
710            .request_with_start(&request, helper_command(&root, "must_not_start"));
711        assert!(matches!(result, Err(RoutineError::Unavailable(_))));
712        server.join().unwrap();
713        assert!(!root.join("helper.pid").exists());
714        let _ = std::fs::remove_dir_all(root);
715    }
716
717    #[test]
718    fn startup_error_handshake_respects_timeout_and_reaps_child() {
719        let root = test_root("error-timeout");
720        std::fs::create_dir_all(&root).unwrap();
721        let pid_path = root.join("helper.pid");
722        let started = Instant::now();
723        let result = RoutineClient::new(root.clone())
724            .with_startup_timeout(Duration::from_millis(300))
725            .request_with_start(&list(&root), helper_command(&root, "error_hang"));
726        assert!(matches!(result, Err(RoutineError::Unavailable(_))));
727        assert!(started.elapsed() < Duration::from_secs(2));
728        let pid: i32 = std::fs::read_to_string(&pid_path).unwrap().parse().unwrap();
729        assert_eq!(unsafe { libc::kill(pid, 0) }, -1);
730        assert_eq!(
731            std::io::Error::last_os_error().raw_os_error(),
732            Some(libc::ESRCH)
733        );
734        let _ = std::fs::remove_dir_all(root);
735    }
736
737    #[test]
738    fn invalid_startup_notification_is_corrupt_and_reaps_child() {
739        let root = test_root("invalid-notification");
740        std::fs::create_dir_all(&root).unwrap();
741        let pid_path = root.join("helper.pid");
742        let result = RoutineClient::new(root.clone())
743            .request_with_start(&list(&root), helper_command(&root, "invalid_hang"));
744        assert!(matches!(result, Err(RoutineError::Corrupt(_))));
745        assert_helper_reaped(&pid_path);
746        let _ = std::fs::remove_dir_all(root);
747    }
748
749    #[test]
750    fn early_child_exit_is_unavailable_and_reaped() {
751        let root = test_root("early-exit");
752        std::fs::create_dir_all(&root).unwrap();
753        let pid_path = root.join("helper.pid");
754        let result = RoutineClient::new(root.clone())
755            .request_with_start(&list(&root), helper_command(&root, "exit"));
756        assert!(matches!(result, Err(RoutineError::Unavailable(_))));
757        assert_helper_reaped(&pid_path);
758        let _ = std::fs::remove_dir_all(root);
759    }
760
761    #[test]
762    fn ready_but_unreachable_is_unavailable_and_reaps_child() {
763        let root = test_root("ready-unreachable");
764        std::fs::create_dir_all(&root).unwrap();
765        let pid_path = root.join("helper.pid");
766        let result = RoutineClient::new(root.clone())
767            .with_startup_timeout(Duration::from_millis(300))
768            .request_with_start(&list(&root), helper_command(&root, "ready_hang"));
769        assert!(matches!(result, Err(RoutineError::Unavailable(_))));
770        assert_helper_reaped(&pid_path);
771        let _ = std::fs::remove_dir_all(root);
772    }
773
774    #[test]
775    fn ready_status_read_cannot_exceed_startup_timeout() {
776        let root = test_root("ready-status-hang");
777        std::fs::create_dir_all(&root).unwrap();
778        let pid_path = root.join("helper.pid");
779        let started = Instant::now();
780        let result = RoutineClient::new(root.clone())
781            .with_startup_timeout(Duration::from_millis(300))
782            .request_with_start(&list(&root), helper_command(&root, "ready_status_hang"));
783
784        assert!(matches!(
785            result,
786            Err(RoutineError::Unavailable(message))
787                if message.contains("reported ready but did not accept status requests")
788        ));
789        assert!(started.elapsed() < Duration::from_secs(2));
790        assert_helper_reaped(&pid_path);
791        let _ = std::fs::remove_dir_all(root);
792    }
793
794    #[test]
795    fn ready_with_non_daemon_status_is_corrupt_and_reaps_child() {
796        let root = test_root("ready-wrong-status");
797        std::fs::create_dir_all(&root).unwrap();
798        let pid_path = root.join("helper.pid");
799        let result = RoutineClient::new(root.clone())
800            .request_with_start(&list(&root), helper_command(&root, "ready_wrong_status"));
801        assert!(matches!(result, Err(RoutineError::Corrupt(_))));
802        assert_helper_reaped(&pid_path);
803        let _ = std::fs::remove_dir_all(root);
804    }
805
806    fn assert_helper_reaped(pid_path: &Path) {
807        let pid: i32 = std::fs::read_to_string(pid_path).unwrap().parse().unwrap();
808        assert_eq!(unsafe { libc::kill(pid, 0) }, -1);
809        assert_eq!(
810            std::io::Error::last_os_error().raw_os_error(),
811            Some(libc::ESRCH)
812        );
813    }
814
815    #[test]
816    #[ignore = "spawned by RoutineClient lifecycle tests"]
817    fn daemon_helper() {
818        let root = PathBuf::from(std::env::var_os("ASCHED_TEST_ROOT").unwrap());
819        let mode = std::env::var("ASCHED_TEST_MODE").unwrap();
820        if mode == "hang" || mode == "must_not_start" {
821            std::fs::write(root.join("helper.pid"), std::process::id().to_string()).unwrap();
822            std::thread::sleep(Duration::from_secs(30));
823            return;
824        }
825        let fd = std::env::var(STARTUP_FD_ENV)
826            .unwrap()
827            .parse::<i32>()
828            .unwrap();
829        let mut startup = unsafe { std::fs::File::from_raw_fd(fd) };
830        if mode == "ready_wrong_status" {
831            use std::io::{BufRead, BufReader};
832            use std::os::unix::net::UnixListener;
833
834            std::fs::create_dir_all(&root).unwrap();
835            std::fs::write(root.join("helper.pid"), std::process::id().to_string()).unwrap();
836            let listener = UnixListener::bind(root.join("daemon-v1.sock")).unwrap();
837            startup.write_all(b"ready").unwrap();
838            let (mut stream, _) = listener.accept().unwrap();
839            let mut request = String::new();
840            BufReader::new(stream.try_clone().unwrap())
841                .read_line(&mut request)
842                .unwrap();
843            stream
844                .write_all(b"{\"result\":\"ok\",\"revision\":null}\n")
845                .unwrap();
846            std::thread::sleep(Duration::from_secs(30));
847            return;
848        }
849        if mode == "ready_status_hang" {
850            use std::os::unix::net::UnixListener;
851
852            std::fs::create_dir_all(&root).unwrap();
853            std::fs::write(root.join("helper.pid"), std::process::id().to_string()).unwrap();
854            let listener = UnixListener::bind(root.join("daemon-v1.sock")).unwrap();
855            startup.write_all(b"ready").unwrap();
856            let _connection = listener.accept().unwrap();
857            std::thread::sleep(Duration::from_secs(30));
858            return;
859        }
860        if matches!(mode.as_str(), "invalid_hang" | "exit" | "ready_hang") {
861            std::fs::write(root.join("helper.pid"), std::process::id().to_string()).unwrap();
862            let notification = match mode.as_str() {
863                "invalid_hang" => Some("invalid"),
864                "ready_hang" => Some("ready"),
865                "exit" => None,
866                _ => unreachable!(),
867            };
868            if let Some(notification) = notification {
869                startup.write_all(notification.as_bytes()).unwrap();
870                std::thread::sleep(Duration::from_secs(30));
871            }
872            return;
873        }
874        if mode == "error_hang" {
875            std::fs::write(root.join("helper.pid"), std::process::id().to_string()).unwrap();
876            startup.write_all(b"error:startup failed").unwrap();
877            std::thread::sleep(Duration::from_secs(30));
878            return;
879        }
880        if mode == "daemon_with_pid" {
881            // ^ Startup helpers run before daemon root initialization.
882            std::fs::create_dir_all(&root).unwrap();
883            std::fs::write(
884                root.join(format!("helper-{}.pid", std::process::id())),
885                std::process::id().to_string(),
886            )
887            .unwrap();
888        }
889        let _ = super::super::daemon::serve_with_startup(root, move |result| {
890            let message = match result {
891                Ok(()) => "ready".to_string(),
892                Err(error) => format!("error:{error}"),
893            };
894            startup.write_all(message.as_bytes()).unwrap();
895        });
896    }
897}
asched_core/routine/client.rs

asched_core/routine/
client.rs
