1use std::io::{BufRead, BufReader, Write};
39use std::path::{Path, PathBuf};
40use std::process::{Child, Command, ExitStatus, Stdio};
41use std::sync::{mpsc, Arc, Condvar, Mutex};
42use std::time::{Duration, Instant};
43
44use crate::delegation::{
45 self, CompletionAddress, DelegationOutcome, DelegationStatus, Reporter, Spawner, SpawnerHandle,
46 SPAWNER_ENV,
47};
48use crate::errors::RuntimeError;
49use crate::formation::{FormationId, FORMATION_ID_ENV};
50use crate::host_warnings::HOST_WARNINGS_REPORTED_ENV;
51use crate::lifeline::SPAWNER_LIFELINE_ENV;
52use crate::mac_token;
53use crate::spawn_credential::SpawnApproval;
54use crate::workdir_fs::WorkdirRoot;
55
56pub const MUR_BINARY_ENV: &str = "MURMUR_MUR_BINARY";
62
63pub const READINESS_TOKENS_FLAG: &str = "--readiness-tokens";
68
69pub const WORKDIR_IDENTITY_FLAG: &str = "--workdir-identity";
73
74#[must_use]
76pub fn format_workdir_identity(identity: (u64, u64)) -> String {
77 format!("{}:{}", identity.0, identity.1)
78}
79
80pub fn parse_workdir_identity(value: &str) -> Result<(u64, u64), String> {
82 let (dev, ino) = value
83 .split_once(':')
84 .ok_or_else(|| format!("{WORKDIR_IDENTITY_FLAG} takes <dev>:<ino>, not {value:?}"))?;
85 let number = |part: &str| {
86 part.parse::<u64>()
87 .map_err(|_| format!("{WORKDIR_IDENTITY_FLAG} takes <dev>:<ino>, not {value:?}"))
88 };
89 Ok((number(dev)?, number(ino)?))
90}
91
92const CHILD_READY_TIMEOUT: Duration = Duration::from_secs(180);
98
99#[cfg(unix)]
102const CHILD_DIR_MODE: u32 = 0o700;
103
104const CHILD_STDERR_TAIL_LINES: usize = 20;
110
111const CHILD_STDERR_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);
118
119const CHILD_WATCH_INTERVAL: Duration = Duration::from_millis(100);
125
126pub const CHILD_END_GRACE: Duration = Duration::from_secs(5);
136
137const CHILD_END_POLL: Duration = Duration::from_millis(20);
140
141#[derive(Debug)]
147pub struct ChildLaunchRequest {
148 pub parent_records_root: WorkdirRoot,
151 pub capsule_name: String,
152 pub capsule_version: String,
153 pub grant: SpawnApproval,
155 pub child_env_allow: Vec<String>,
159 pub roost_url: String,
161 pub spawner: Option<Spawner>,
170 pub completion_deadline: Option<Duration>,
178 pub formation_id: Option<FormationId>,
182 pub deadline: Option<crate::session_status::SessionDeadline>,
187}
188
189enum Ending {
191 Exited(Option<ExitStatus>),
194 Deliberate,
197}
198
199struct ChildProcess {
206 child: Option<Child>,
208 deliberate: bool,
212 status: Option<ExitStatus>,
214 terminated_at: Option<Instant>,
218 #[cfg(unix)]
223 lifeline: Option<crate::lifeline::ChildLifeline>,
224}
225
226impl ChildProcess {
227 fn close_lifeline(&mut self) {
229 #[cfg(unix)]
230 if let Some(mut lifeline) = self.lifeline.take() {
231 lifeline.close();
232 }
233 }
234
235 fn poll(&mut self) -> Option<Ending> {
240 if self.deliberate {
241 return (!self.is_being_ended()).then_some(Ending::Deliberate);
244 }
245 let Some(child) = self.child.as_mut() else {
246 return Some(Ending::Exited(self.status));
249 };
250 let ending = match child.try_wait() {
251 Ok(Some(status)) => {
252 self.status = Some(status);
253 Ending::Exited(Some(status))
254 }
255 Ok(None) => return None,
256 Err(_) => Ending::Exited(self.status),
259 };
260 self.close_lifeline();
261 Some(ending)
262 }
263
264 #[allow(unsafe_code)]
271 fn terminate(&mut self) {
272 if self.status.is_none() {
273 self.deliberate = true;
274 }
275 if self.terminated_at.is_some() {
276 return;
277 }
278 let Some(child) = self.child.as_mut() else {
279 return;
280 };
281 if let Ok(Some(status)) = child.try_wait() {
282 self.status = Some(status);
283 return;
284 }
285 #[cfg(unix)]
286 if let Ok(pid) = libc::pid_t::try_from(child.id()) {
287 unsafe {
291 libc::kill(pid, libc::SIGTERM);
292 }
293 }
294 #[cfg(not(unix))]
295 let _ = child.kill();
296 self.terminated_at = Some(Instant::now());
297 }
298
299 fn is_being_ended(&self) -> bool {
302 self.deliberate && self.status.is_none() && self.terminated_at.is_some()
303 }
304}
305
306fn end_process(process: &Mutex<ChildProcess>) -> Result<(), std::io::Error> {
316 let (mut child, signalled) = {
317 let mut process = lock(process);
318 process.terminate();
319 let Some(child) = process.child.take() else {
320 if process.status.is_some() || process.terminated_at.is_none() {
323 process.close_lifeline();
324 }
325 return Ok(());
326 };
327 (child, process.terminated_at)
328 };
329 let waited = match signalled {
330 None => child.wait(),
331 Some(signalled) => loop {
332 match child.try_wait() {
333 Ok(Some(status)) => break Ok(status),
334 Ok(None) if signalled.elapsed() < CHILD_END_GRACE => {
335 std::thread::sleep(CHILD_END_POLL);
336 }
337 _ => {
341 let _ = child.kill();
342 break child.wait();
343 }
344 }
345 },
346 };
347 let mut process = lock(process);
348 process.close_lifeline();
349 match waited {
350 Ok(status) => {
351 process.status = Some(status);
352 Ok(())
353 }
354 Err(error) => {
356 process.terminated_at = None;
357 Err(error)
358 }
359 }
360}
361
362pub struct LaunchedChild {
364 pub workdir: PathBuf,
368 pub(crate) workdir_root: WorkdirRoot,
371 pub session_id: String,
373 pub records_dir: PathBuf,
377 pub(crate) records_root: WorkdirRoot,
381 pub capsule_url: String,
384 pub argv: Vec<String>,
389 pub env: Vec<(String, String)>,
392 pub delegation_id: Option<String>,
397 pub door_token: Option<crate::door_auth::DoorToken>,
401 pub formation_id: Option<FormationId>,
404 pub(crate) task_id: Option<String>,
408 process: Arc<Mutex<ChildProcess>>,
409 stderr_tail: Arc<StderrTail>,
411 started: Instant,
413 watcher: Option<std::thread::JoinHandle<()>>,
415}
416
417impl std::fmt::Debug for LaunchedChild {
418 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
419 f.debug_struct("LaunchedChild")
420 .field("workdir", &self.workdir)
421 .field("session_id", &self.session_id)
422 .field("records_dir", &self.records_dir)
423 .field("capsule_url", &self.capsule_url)
424 .field("delegation_id", &self.delegation_id)
425 .field("formation_id", &self.formation_id)
426 .field(
427 "door_token",
428 &self.door_token.as_ref().map(|_| "<redacted>"),
429 )
430 .field("pid", &self.pid())
431 .finish()
432 }
433}
434
435impl LaunchedChild {
436 pub fn pid(&self) -> u32 {
438 lock(&self.process)
439 .child
440 .as_ref()
441 .map(Child::id)
442 .unwrap_or(0)
443 }
444
445 pub fn stderr_tail(&self) -> Vec<String> {
450 self.stderr_tail.lines()
451 }
452
453 pub fn begin_ending(&self) {
459 lock(&self.process).terminate();
460 }
461
462 pub fn shutdown(&mut self) -> Result<(), RuntimeError> {
468 end_process(&self.process).map_err(|error| {
469 RuntimeError::Runtime(format!("failed to reap child capsule: {error}"))
470 })
471 }
472
473 pub fn is_being_ended(&self) -> bool {
479 lock(&self.process).is_being_ended()
480 }
481
482 pub fn watcher_finished(&self) -> bool {
489 self.watcher
490 .as_ref()
491 .is_none_or(std::thread::JoinHandle::is_finished)
492 }
493
494 pub fn has_exited(&self) -> bool {
500 lock(&self.process).poll().is_some()
501 }
502}
503
504impl Drop for LaunchedChild {
505 fn drop(&mut self) {
509 let _ = end_process(&self.process);
510 }
511}
512
513fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
516 mutex
517 .lock()
518 .unwrap_or_else(|poisoned| poisoned.into_inner())
519}
520
521pub fn child_workdir_for(parent_records_dir: &Path, capsule_name: &str) -> PathBuf {
529 parent_records_dir.join(child_workdir_rel(capsule_name))
530}
531
532pub(crate) const CHILDREN_DIR: &str = "children";
535
536fn child_workdir_rel(capsule_name: &str) -> PathBuf {
538 let suffix = mac_token::random_hex(8).unwrap_or_else(|_| {
539 format!(
542 "{:016x}",
543 std::time::SystemTime::now()
544 .duration_since(std::time::UNIX_EPOCH)
545 .map(|elapsed| elapsed.as_nanos() as u64)
546 .unwrap_or(0)
547 )
548 });
549 Path::new(CHILDREN_DIR).join(format!(
550 "{}-{suffix}",
551 murmur_artifact::tool_name_of(capsule_name)
552 ))
553}
554
555pub fn launch_child_capsule(request: ChildLaunchRequest) -> Result<LaunchedChild, RuntimeError> {
560 #[cfg(unix)]
563 let mut lifeline = crate::lifeline::ChildLifeline::new().map_err(|error| {
564 RuntimeError::Runtime(format!(
565 "the child capsule's spawner lifeline could not be created: {error}"
566 ))
567 })?;
568 let workdir_rel = child_workdir_rel(&request.capsule_name);
569 let workdir_root = create_child_dir(&request.parent_records_root, &workdir_rel)?;
570 let identity = workdir_root.identity()?;
571 let workdir = workdir_root.path().to_path_buf();
572
573 let binary = mur_binary()?;
574 let mut argv: Vec<String> = vec![
575 binary.display().to_string(),
576 "run".to_string(),
577 "--capsule".to_string(),
578 request.capsule_name.clone(),
579 "--capsule-version".to_string(),
580 request.capsule_version.clone(),
581 "--workdir".to_string(),
582 workdir.display().to_string(),
583 WORKDIR_IDENTITY_FLAG.to_string(),
584 format_workdir_identity(identity),
585 "--json".to_string(),
586 READINESS_TOKENS_FLAG.to_string(),
587 "--no-env-file".to_string(),
588 "--spawn-grant-stdin".to_string(),
589 ];
590 if let Some(deadline) = request.deadline {
591 argv.push(crate::session_status::INHERITED_DEADLINE_FLAG.to_string());
592 argv.push(deadline.at_unix_millis().to_string());
593 }
594 let handle = request
597 .spawner
598 .as_ref()
599 .map(|spawner| SpawnerHandle::for_delegation(spawner, delegation::new_delegation_id()));
600 #[cfg_attr(not(unix), allow(unused_mut))]
601 let mut env = child_environment(&request, handle.as_ref());
602 #[cfg(unix)]
605 if let Some((name, value)) = lifeline.env_pair() {
606 let at = env
607 .iter()
608 .position(|(key, _)| key == SPAWNER_ENV || key == FORMATION_ID_ENV)
609 .unwrap_or(env.len());
610 env.insert(at, (name.to_string(), value));
611 }
612 let started = Instant::now();
613
614 let StartedProcess {
618 child,
619 stderr_tail,
620 stdin_written: write_result,
621 } = start_process(&ProcessLaunch {
622 binary: &binary,
623 args: &argv[1..],
624 cwd: Some(&workdir),
625 env: ProcessEnv::Cleared(&env),
626 stdin_line: Some(request.grant.expose()),
627 stderr_prefix: None,
628 inherit_output: false,
629 own_process_group: false,
630 inherit_fd: None,
631 #[cfg(unix)]
632 lifeline: Some(Lifeline::Child(&lifeline)),
633 })
634 .map_err(|error| {
635 RuntimeError::Runtime(format!(
636 "failed to start the child capsule process '{}': {error}",
637 binary.display()
638 ))
639 })?;
640 #[cfg(unix)]
641 lifeline.spawned();
642 let unclaimed = UnclaimedChild {
643 process: Arc::new(Mutex::new(ChildProcess {
644 child: Some(child),
645 deliberate: false,
646 status: None,
647 terminated_at: None,
648 #[cfg(unix)]
649 lifeline: Some(lifeline),
650 })),
651 claimed: false,
652 };
653 if let Err(error) = write_result {
654 return Err(RuntimeError::Runtime(format!(
655 "failed to hand the child its launch grant: {error}"
656 )));
657 }
658
659 let readiness = read_readiness(&unclaimed.process, &workdir, &stderr_tail, &workdir_root)?;
660 let mut launched = LaunchedChild {
661 workdir,
662 workdir_root,
663 session_id: readiness.session_id,
664 records_dir: readiness.records_dir,
665 records_root: readiness.records_root,
666 capsule_url: readiness.capsule_url,
667 argv,
668 env,
669 delegation_id: handle.as_ref().map(|handle| handle.delegation_id.clone()),
670 door_token: readiness.door_token,
671 formation_id: readiness.formation_id,
672 task_id: None,
673 process: unclaimed.claim(),
674 stderr_tail,
675 started,
676 watcher: None,
677 };
678
679 if let Some(handle) = handle {
683 if let Some(address) = handle.report_to.clone() {
684 launched.watcher = watch_for_completion(
685 &launched,
686 handle,
687 address,
688 &request.capsule_name,
689 &request.capsule_version,
690 request.completion_deadline,
691 );
692 }
693 }
694 Ok(launched)
695}
696
697struct UnclaimedChild {
701 process: Arc<Mutex<ChildProcess>>,
702 claimed: bool,
703}
704
705impl UnclaimedChild {
706 fn claim(mut self) -> Arc<Mutex<ChildProcess>> {
708 self.claimed = true;
709 Arc::clone(&self.process)
710 }
711}
712
713impl Drop for UnclaimedChild {
714 fn drop(&mut self) {
715 if !self.claimed {
716 let _ = end_process(&self.process);
717 }
718 }
719}
720
721struct Readiness {
723 session_id: String,
724 records_dir: PathBuf,
725 records_root: WorkdirRoot,
727 door_token: Option<crate::door_auth::DoorToken>,
728 formation_id: Option<FormationId>,
729 capsule_url: String,
731}
732
733fn read_readiness(
736 process: &Mutex<ChildProcess>,
737 workdir: &Path,
738 stderr_tail: &StderrTail,
739 latch_of: &WorkdirRoot,
740) -> Result<Readiness, RuntimeError> {
741 let line = first_json_line(process, workdir, stderr_tail)?;
742 let report: serde_json::Value = serde_json::from_str(&line).map_err(|error| {
743 RuntimeError::Runtime(format!(
744 "the child capsule's first --json line did not parse: {error}"
745 ))
746 })?;
747 let session_id = report
748 .get("session_id")
749 .and_then(serde_json::Value::as_str)
750 .ok_or_else(|| {
751 RuntimeError::Runtime(
752 "the child capsule's --json line carried no session_id".to_string(),
753 )
754 })?;
755 let url = report
756 .get("url")
757 .and_then(serde_json::Value::as_str)
758 .unwrap_or_default();
759
760 let records_dir = readiness_records_dir(&report, session_id)?;
761 let records_root =
762 WorkdirRoot::open_sharing_latch_absolute(&records_dir, latch_of).map_err(|error| {
763 RuntimeError::Runtime(format!(
764 "the child capsule's records directory could not be opened: {error}"
765 ))
766 })?;
767 Ok(Readiness {
768 session_id: session_id.to_string(),
769 records_dir,
770 records_root,
771 door_token: readiness_door_token(&report),
772 formation_id: report
773 .get("formation_id")
774 .and_then(serde_json::Value::as_str)
775 .and_then(|id| FormationId::parse(id).ok()),
776 capsule_url: if url.is_empty() {
779 String::new()
780 } else {
781 format!("http://{url}")
782 },
783 })
784}
785
786pub(crate) fn readiness_records_dir(
791 report: &serde_json::Value,
792 session_id: &str,
793) -> Result<PathBuf, RuntimeError> {
794 let refused = |why: String| {
795 RuntimeError::Runtime(format!(
796 "the child capsule's --json line {why}, so its outcome has nowhere to be read from"
797 ))
798 };
799 let reported = report
800 .get("records_dir")
801 .and_then(serde_json::Value::as_str)
802 .map(PathBuf::from)
803 .ok_or_else(|| refused("carried no records_dir".to_string()))?;
804 let records_root = crate::records_dir::records_dir_root()
805 .map_err(|error| refused(format!("named a records directory and this host's {error}")))?;
806 let well_formed = reported.is_absolute()
807 && reported.starts_with(&records_root)
808 && reported.file_name().and_then(|name| name.to_str()) == Some(session_id)
809 && !reported
810 .components()
811 .any(|part| matches!(part, std::path::Component::ParentDir));
812 if !well_formed {
813 return Err(refused(format!(
814 "named {} as its records directory, which is not {}/<workdir root>/{session_id}",
815 reported.display(),
816 records_root.display()
817 )));
818 }
819 Ok(reported)
820}
821
822pub(crate) fn readiness_door_token(
826 report: &serde_json::Value,
827) -> Option<crate::door_auth::DoorToken> {
828 report
829 .pointer("/tokens/operator")
830 .and_then(serde_json::Value::as_str)
831 .filter(|token| !token.is_empty())
832 .map(|token| crate::door_auth::DoorToken::new(token.to_string()))
833}
834
835fn watch_for_completion(
864 launched: &LaunchedChild,
865 handle: SpawnerHandle,
866 address: CompletionAddress,
867 capsule_name: &str,
868 capsule_version: &str,
869 deadline: Option<Duration>,
870) -> Option<std::thread::JoinHandle<()>> {
871 let process = Arc::clone(&launched.process);
872 let stderr_tail = Arc::clone(&launched.stderr_tail);
873 let workdir = launched.workdir.clone();
874 let records_root = launched.records_root.clone();
875 let session_id = launched.session_id.clone();
876 let capsule_name = capsule_name.to_string();
877 let capsule_version = capsule_version.to_string();
878 let started = launched.started;
879
880 let watching_from = Instant::now();
881 let spawned = std::thread::Builder::new().spawn(move || {
882 let expires_at = deadline.map(|bound| watching_from + bound);
883 let mut expired = false;
886 let ending = loop {
887 if expires_at.is_some_and(|at| Instant::now() >= at) {
888 let _ = end_process(&process);
889 expired = true;
890 break Ending::Deliberate;
891 }
892 if let Some(ending) = lock(&process).poll() {
893 break ending;
894 }
895 std::thread::sleep(CHILD_WATCH_INTERVAL);
896 };
897 let duration_ms = started.elapsed().as_millis().try_into().unwrap_or(u64::MAX);
898
899 let recorded = match ending {
903 Ending::Exited(_) => delegation::read_completion(&records_root),
904 Ending::Deliberate => None,
905 };
906 if let Some(recorded) = recorded {
907 if recorded.delivered {
908 return;
909 }
910 let retried =
911 delegation::report_completion(&handle, &address, recorded, &records_root);
912 if !retried.delivered {
913 crate::runtime_err!(
915 "[capsule-runtime] delegation {}: the child's completion is undelivered after one retry",
916 handle.delegation_id
917 );
918 }
919 return;
920 }
921
922 let outcome = |status: DelegationStatus, detail: String| DelegationOutcome {
923 delegation_id: handle.delegation_id.clone(),
924 capsule_name: capsule_name.clone(),
925 capsule_version: capsule_version.clone(),
926 session_id: session_id.clone(),
927 status,
928 result_path: None,
931 workdir: workdir.display().to_string(),
932 duration_ms,
933 detail: Some(detail),
934 reported_by: Reporter::Launcher,
935 delivered: false,
936 delivery_error: None,
937 };
938
939 match ending {
940 Ending::Deliberate if expired => {
944 let bound = deadline.map(|bound| bound.as_secs()).unwrap_or_default();
945 delegation::report_completion(
946 &handle,
947 &address,
948 outcome(
949 DelegationStatus::Terminated,
950 format!(
951 "the capsule was still running after {bound}s and was ended at the \
952 delegation deadline"
953 ),
954 ),
955 &records_root,
956 );
957 }
958 Ending::Deliberate => {
959 delegation::record_terminated(
960 &records_root,
961 outcome(
962 DelegationStatus::Terminated,
963 delegation::PARENT_ENDED_DETAIL.to_string(),
964 ),
965 );
966 }
967 Ending::Exited(status) => {
968 let status_text = match status {
969 Some(status) => status.to_string(),
970 None => "unknown exit status".to_string(),
971 };
972 let tail = stderr_tail.lines_at_end().join("\n");
973 let detail = if tail.is_empty() {
974 format!(
975 "the child process ended without recording a completion ({status_text})"
976 )
977 } else {
978 format!(
979 "the child process ended without recording a completion ({status_text}); \
980 its last stderr lines were:\n{tail}"
981 )
982 };
983 delegation::report_completion(
984 &handle,
985 &address,
986 outcome(DelegationStatus::Crashed, detail),
987 &records_root,
988 );
989 }
990 }
991 });
992 match spawned {
993 Ok(watcher) => Some(watcher),
994 Err(error) => {
995 crate::runtime_err!(
996 "[capsule-runtime] delegation {}: the child's completion watcher could not be \
997 started: {error}",
998 launched.delegation_id.as_deref().unwrap_or_default()
999 );
1000 None
1001 }
1002 }
1003}
1004
1005pub(crate) fn child_environment(
1030 request: &ChildLaunchRequest,
1031 handle: Option<&SpawnerHandle>,
1032) -> Vec<(String, String)> {
1033 let mut env: Vec<(String, String)> = Vec::new();
1034 for key in &request.child_env_allow {
1035 if let Ok(value) = std::env::var(key) {
1036 env.push((key.clone(), value));
1037 }
1038 }
1039 env.retain(|(key, _)| {
1040 !matches!(key.as_str(), "PATH" | "HOME" | "MURMUR_ROOST_URL")
1041 && !crate::shell::ACCOUNT_ENV.contains(&key.as_str())
1042 && key != SPAWNER_ENV
1043 && key != FORMATION_ID_ENV
1044 && key != SPAWNER_LIFELINE_ENV
1045 && key != HOST_WARNINGS_REPORTED_ENV
1046 && !crate::formation::is_member_grant_env(key)
1047 });
1048
1049 if let Ok(path) = std::env::var("PATH") {
1050 env.push(("PATH".to_string(), path));
1051 }
1052 if let Ok(home) = std::env::var("HOME") {
1053 env.push(("HOME".to_string(), home));
1054 }
1055 env.extend(
1056 crate::shell::HostAccountEnv::from_host()
1057 .set()
1058 .iter()
1059 .cloned(),
1060 );
1061 env.push(("MURMUR_ROOST_URL".to_string(), request.roost_url.clone()));
1062 env.push(crate::host_warnings::reported_env_pair());
1063 if let Some(handle) = handle {
1067 env.push((SPAWNER_ENV.to_string(), handle.to_env_value()));
1068 }
1069 if let Some(formation_id) = &request.formation_id {
1071 let (name, value) = formation_id.env_pair();
1072 env.push((name.to_string(), value));
1073 }
1074 env
1075}
1076
1077pub(crate) fn mur_binary() -> Result<PathBuf, RuntimeError> {
1079 if let Some(path) = std::env::var_os(MUR_BINARY_ENV) {
1080 if !path.is_empty() {
1081 return Ok(PathBuf::from(path));
1082 }
1083 }
1084 std::env::current_exe().map_err(|error| {
1085 RuntimeError::Runtime(format!(
1086 "cannot locate the mur binary to launch a child capsule with: {error}"
1087 ))
1088 })
1089}
1090
1091fn create_child_dir(records_root: &WorkdirRoot, rel: &Path) -> Result<WorkdirRoot, RuntimeError> {
1096 records_root.create_dir_all(CHILDREN_DIR, CHILD_DIR_MODE)?;
1097 records_root.create_dir_all(rel, CHILD_DIR_MODE)?;
1098 records_root
1099 .open_beneath(&rel.to_string_lossy())?
1100 .ok_or_else(|| {
1101 RuntimeError::Runtime(format!(
1102 "the child capsule's directory {} vanished after its creation",
1103 records_root.display_path(rel)
1104 ))
1105 })
1106}
1107
1108pub(crate) struct StderrTail {
1112 state: Mutex<StderrState>,
1113 drained: Condvar,
1115}
1116
1117#[derive(Default)]
1118struct StderrState {
1119 lines: Vec<String>,
1120 at_eof: bool,
1122}
1123
1124impl StderrTail {
1125 pub(crate) fn lines(&self) -> Vec<String> {
1128 lock(&self.state).lines.clone()
1129 }
1130
1131 pub(crate) fn lines_at_end(&self) -> Vec<String> {
1139 let (state, _) = self
1140 .drained
1141 .wait_timeout_while(lock(&self.state), CHILD_STDERR_DRAIN_TIMEOUT, |state| {
1142 !state.at_eof
1143 })
1144 .unwrap_or_else(|poisoned| poisoned.into_inner());
1145 state.lines.clone()
1146 }
1147
1148 fn finish(&self) {
1150 lock(&self.state).at_eof = true;
1151 self.drained.notify_all();
1152 }
1153}
1154
1155fn drain_stderr(child: &mut Child, prefix: Option<&str>) -> Arc<StderrTail> {
1161 let tail = Arc::new(StderrTail {
1162 state: Mutex::new(StderrState::default()),
1163 drained: Condvar::new(),
1164 });
1165 let Some(stderr) = child.stderr.take() else {
1166 tail.finish();
1169 return tail;
1170 };
1171 let collector = Arc::clone(&tail);
1172 let prefix = prefix.unwrap_or_default().to_string();
1173 std::thread::spawn(move || {
1174 for line in BufReader::new(stderr).lines().map_while(Result::ok) {
1175 crate::runtime_err!("{prefix}{line}");
1176 let mut state = lock(&collector.state);
1177 if state.lines.len() == CHILD_STDERR_TAIL_LINES {
1178 state.lines.remove(0);
1179 }
1180 state.lines.push(line);
1181 }
1182 collector.finish();
1183 });
1184 tail
1185}
1186
1187fn first_json_line(
1192 process: &Mutex<ChildProcess>,
1193 workdir: &Path,
1194 stderr_tail: &StderrTail,
1195) -> Result<String, RuntimeError> {
1196 let stdout = lock(process)
1197 .child
1198 .as_mut()
1199 .and_then(|child| child.stdout.take())
1200 .ok_or_else(|| {
1201 RuntimeError::Runtime("the child process exposed no standard output".to_string())
1202 })?;
1203
1204 let rx = relay_readiness_line(stdout, AfterReadiness::Stdout);
1205
1206 match rx.recv_timeout(CHILD_READY_TIMEOUT) {
1207 Ok(Some(line)) => Ok(line),
1208 Ok(None) => {
1209 let status = lock(process)
1210 .child
1211 .as_mut()
1212 .and_then(|child| child.wait().ok())
1213 .map(|status| status.to_string())
1214 .unwrap_or_else(|| "unknown".to_string());
1215 let reason = stderr_tail.lines_at_end().join("\n");
1219 Err(RuntimeError::Runtime(format!(
1220 "the child capsule '{}' exited without reporting a launch line ({status}): {reason}",
1221 workdir.display()
1222 )))
1223 }
1224 Err(_) => Err(RuntimeError::Runtime(format!(
1225 "the child capsule '{}' did not report a launch line within {}s",
1226 workdir.display(),
1227 CHILD_READY_TIMEOUT.as_secs()
1228 ))),
1229 }
1230}
1231
1232pub(crate) enum ProcessEnv<'a> {
1234 Cleared(&'a [(String, String)]),
1237 Inherited {
1240 set: &'a [(String, String)],
1241 remove: &'a [&'a str],
1242 },
1243}
1244
1245pub(crate) enum AfterReadiness {
1247 Stdout,
1249 PrefixedStderr(String),
1251}
1252
1253pub(crate) struct ProcessLaunch<'a> {
1255 pub(crate) binary: &'a Path,
1256 pub(crate) args: &'a [String],
1258 pub(crate) cwd: Option<&'a Path>,
1260 pub(crate) env: ProcessEnv<'a>,
1261 pub(crate) stdin_line: Option<&'a str>,
1264 pub(crate) stderr_prefix: Option<&'a str>,
1266 pub(crate) inherit_output: bool,
1269 pub(crate) own_process_group: bool,
1273 pub(crate) inherit_fd: Option<i32>,
1277 #[cfg(unix)]
1280 pub(crate) lifeline: Option<Lifeline<'a>>,
1281}
1282
1283#[cfg(unix)]
1285#[derive(Clone, Copy)]
1286pub(crate) enum Lifeline<'a> {
1287 Member(&'a crate::lifeline::MemberLifeline),
1288 Child(&'a crate::lifeline::ChildLifeline),
1289}
1290
1291#[cfg(unix)]
1292impl Lifeline<'_> {
1293 fn hand_to(self, command: &mut Command) {
1294 match self {
1295 Lifeline::Member(lifeline) => lifeline.hand_to(command),
1296 Lifeline::Child(lifeline) => lifeline.hand_to(command),
1297 }
1298 }
1299}
1300
1301pub(crate) struct StartedProcess {
1303 pub(crate) child: Child,
1305 pub(crate) stderr_tail: Arc<StderrTail>,
1306 pub(crate) stdin_written: Result<(), String>,
1308}
1309
1310pub(crate) fn start_process(launch: &ProcessLaunch<'_>) -> std::io::Result<StartedProcess> {
1313 let mut command = Command::new(launch.binary);
1314 command
1315 .args(launch.args)
1316 .stdin(if launch.stdin_line.is_some() {
1317 Stdio::piped()
1318 } else {
1319 Stdio::null()
1320 })
1321 .stdout(if launch.inherit_output {
1322 Stdio::inherit()
1323 } else {
1324 Stdio::piped()
1325 })
1326 .stderr(if launch.inherit_output {
1327 Stdio::inherit()
1328 } else {
1329 Stdio::piped()
1330 });
1331 if let Some(cwd) = launch.cwd {
1332 command.current_dir(cwd);
1333 }
1334 match launch.env {
1335 ProcessEnv::Cleared(pairs) => {
1336 command.env_clear();
1337 for (key, value) in pairs {
1338 command.env(key, value);
1339 }
1340 }
1341 ProcessEnv::Inherited { set, remove } => {
1342 for name in remove {
1343 command.env_remove(name);
1344 }
1345 for (key, value) in set {
1346 command.env(key, value);
1347 }
1348 }
1349 }
1350 #[cfg(unix)]
1351 if launch.own_process_group {
1352 use std::os::unix::process::CommandExt;
1353 command.process_group(0);
1354 }
1355 #[cfg(unix)]
1356 if let Some(fd) = launch.inherit_fd {
1357 keep_across_exec(&mut command, fd);
1358 }
1359 #[cfg(unix)]
1364 if let Some(lifeline) = launch.lifeline {
1365 lifeline.hand_to(&mut command);
1366 }
1367
1368 let mut child = command.spawn()?;
1369 let stdin_written = match launch.stdin_line {
1370 None => Ok(()),
1371 Some(line) => child
1372 .stdin
1373 .take()
1374 .ok_or_else(|| "the process exposed no standard input".to_string())
1375 .and_then(|mut stdin| writeln!(stdin, "{line}").map_err(|error| error.to_string())),
1376 };
1377 let stderr_tail = drain_stderr(&mut child, launch.stderr_prefix);
1378 Ok(StartedProcess {
1379 child,
1380 stderr_tail,
1381 stdin_written,
1382 })
1383}
1384
1385#[cfg(unix)]
1388#[allow(unsafe_code)]
1389fn keep_across_exec(command: &mut Command, fd: i32) {
1390 use std::os::unix::process::CommandExt;
1391 unsafe {
1395 command.pre_exec(move || {
1396 let flags = libc::fcntl(fd, libc::F_GETFD);
1397 if flags == -1 || libc::fcntl(fd, libc::F_SETFD, flags & !libc::FD_CLOEXEC) == -1 {
1398 return Err(std::io::Error::last_os_error());
1399 }
1400 Ok(())
1401 });
1402 }
1403}
1404
1405pub(crate) fn relay_readiness_line(
1416 stdout: std::process::ChildStdout,
1417 after: AfterReadiness,
1418) -> mpsc::Receiver<Option<String>> {
1419 let (tx, rx) = mpsc::channel::<Option<String>>();
1420 std::thread::spawn(move || {
1421 let mut reader = BufReader::new(stdout);
1422 let mut line = String::new();
1423 let first = match reader.read_line(&mut line) {
1424 Ok(0) | Err(_) => None,
1425 Ok(_) => Some(line.trim_end().to_string()),
1426 };
1427 let _ = tx.send(first);
1428 let mut rest = String::new();
1429 while let Ok(read) = reader.read_line(&mut rest) {
1430 if read == 0 {
1431 break;
1432 }
1433 match &after {
1434 AfterReadiness::Stdout => {
1435 crate::diagnostic::raw_to_stdout(&rest);
1436 }
1437 AfterReadiness::PrefixedStderr(prefix) => {
1438 crate::runtime_err!("{prefix}{}", rest.trim_end_matches(['\r', '\n']));
1439 }
1440 }
1441 rest.clear();
1442 }
1443 });
1444 rx
1445}
1446
1447#[cfg(test)]
1448mod tests {
1449 use super::*;
1450
1451 #[test]
1452 fn a_child_directory_is_unique_per_delegation() {
1453 let parent = PathBuf::from("/tmp/parent");
1454 let first = child_workdir_for(&parent, "worker");
1455 let second = child_workdir_for(&parent, "worker");
1456
1457 assert_ne!(first, second);
1458 assert_eq!(first.parent(), Some(parent.join(CHILDREN_DIR).as_path()));
1459 assert!(first
1460 .file_name()
1461 .unwrap()
1462 .to_string_lossy()
1463 .starts_with("worker-"));
1464 assert_eq!(
1466 first.file_name().unwrap().to_string_lossy().len(),
1467 "worker-".len() + 16
1468 );
1469 }
1470
1471 #[test]
1474 fn a_child_directory_is_named_by_the_bare_part_of_a_full_name() {
1475 let parent = PathBuf::from("/tmp/parent");
1476 let scoped = child_workdir_for(&parent, "local/worker");
1477
1478 assert_eq!(scoped.parent(), Some(parent.join(CHILDREN_DIR).as_path()));
1479 let name = scoped.file_name().unwrap().to_string_lossy().into_owned();
1480 assert!(name.starts_with("worker-"), "{name}");
1481 assert_eq!(name.len(), "worker-".len() + 16);
1482 }
1483
1484 fn request(env_allow: &[&str], spawner: Option<Spawner>) -> ChildLaunchRequest {
1485 ChildLaunchRequest {
1486 parent_records_root: WorkdirRoot::open(&std::env::temp_dir()).unwrap(),
1487 capsule_name: "worker".to_string(),
1488 capsule_version: "0.1.0".to_string(),
1489 grant: SpawnApproval::new("msa1.token".to_string()),
1490 child_env_allow: env_allow.iter().map(|name| name.to_string()).collect(),
1491 roost_url: "http://127.0.0.1:7700".to_string(),
1492 spawner,
1493 completion_deadline: None,
1494 formation_id: None,
1495 deadline: None,
1496 }
1497 }
1498
1499 fn decoy() -> String {
1506 SpawnerHandle {
1507 session_id: "ses_decoy".to_string(),
1508 context_id: "ctx_decoy".to_string(),
1509 delegation_id: "dlg_decoy".to_string(),
1510 report_to: None,
1511 }
1512 .to_env_value()
1513 }
1514
1515 fn spawner() -> Spawner {
1516 Spawner {
1517 session_id: "ses_parent".to_string(),
1518 context_id: "ctx_parent".to_string(),
1519 report_to: Some(CompletionAddress {
1520 url: "http://127.0.0.1:7000".to_string(),
1521 }),
1522 }
1523 }
1524
1525 #[test]
1526 fn a_childs_environment_is_built_from_its_own_declaration() {
1527 std::env::set_var("MURMUR_CHILD_LAUNCH_TEST_A", "a");
1528 std::env::set_var("MURMUR_CHILD_LAUNCH_TEST_B", "b");
1529
1530 let env = child_environment(&request(&["MURMUR_CHILD_LAUNCH_TEST_A"], None), None);
1531 let names: Vec<&str> = env.iter().map(|(key, _)| key.as_str()).collect();
1532
1533 assert!(names.contains(&"MURMUR_CHILD_LAUNCH_TEST_A"));
1534 assert!(!names.contains(&"MURMUR_CHILD_LAUNCH_TEST_B"));
1535 assert!(names.contains(&"MURMUR_ROOST_URL"));
1536 assert!(!env.iter().any(|(_, value)| value.contains("msa1.token")));
1537 }
1538
1539 #[test]
1540 fn a_runtime_owned_name_cannot_be_displaced_by_a_declaration() {
1541 std::env::set_var("MURMUR_ROOST_URL", "http://127.0.0.1:1");
1542
1543 let env = child_environment(&request(&["MURMUR_ROOST_URL"], None), None);
1544 let urls: Vec<&String> = env
1545 .iter()
1546 .filter(|(key, _)| key == "MURMUR_ROOST_URL")
1547 .map(|(_, value)| value)
1548 .collect();
1549
1550 assert_eq!(urls, vec!["http://127.0.0.1:7700"]);
1551 }
1552
1553 #[test]
1556 fn the_spawner_handle_is_injected_last_and_cannot_be_displaced() {
1557 std::env::set_var(SPAWNER_ENV, decoy());
1558 let handle = SpawnerHandle::for_delegation(&spawner(), "dlg_0001".to_string());
1559
1560 let env = child_environment(&request(&[SPAWNER_ENV], Some(spawner())), Some(&handle));
1561 let injected: Vec<&String> = env
1562 .iter()
1563 .filter(|(key, _)| key == SPAWNER_ENV)
1564 .map(|(_, value)| value)
1565 .collect();
1566
1567 assert_eq!(injected.len(), 1, "{env:?}");
1568 assert_eq!(
1569 SpawnerHandle::parse(injected[0]).expect("the injected value is a handle"),
1570 handle
1571 );
1572 assert_eq!(
1573 env.last().map(|(key, _)| key.as_str()),
1574 Some(SPAWNER_ENV),
1575 "the handle is applied in the runtime-owned tail: {env:?}"
1576 );
1577 std::env::remove_var(SPAWNER_ENV);
1578 }
1579
1580 #[test]
1582 fn a_launch_without_a_spawner_injects_no_handle() {
1583 std::env::set_var(SPAWNER_ENV, decoy());
1584
1585 let env = child_environment(&request(&[SPAWNER_ENV], None), None);
1586
1587 assert!(!env.iter().any(|(key, _)| key == SPAWNER_ENV), "{env:?}");
1588 std::env::remove_var(SPAWNER_ENV);
1589 }
1590
1591 #[test]
1594 fn the_formation_id_is_injected_last_and_cannot_be_displaced() {
1595 let _guard = crate::formation::FORMATION_ENV_LOCK
1596 .lock()
1597 .unwrap_or_else(|poisoned| poisoned.into_inner());
1598 let decoy = FormationId::mint();
1599 std::env::set_var(FORMATION_ID_ENV, decoy.as_str());
1600 let injected = |env: &[(String, String)]| -> Vec<String> {
1601 env.iter()
1602 .filter(|(key, _)| key == FORMATION_ID_ENV)
1603 .map(|(_, value)| value.clone())
1604 .collect()
1605 };
1606
1607 let outside = child_environment(&request(&[FORMATION_ID_ENV], None), None);
1608 assert!(injected(&outside).is_empty(), "{outside:?}");
1609
1610 let other = FormationId::mint();
1611 let handle = SpawnerHandle::for_delegation(&spawner(), "dlg_0001".to_string());
1612 for (spawner, handle) in [(Some(spawner()), Some(&handle)), (None, None)] {
1613 let mut member = request(&[FORMATION_ID_ENV], spawner);
1614 member.formation_id = Some(other.clone());
1615 let env = child_environment(&member, handle);
1616 assert_eq!(injected(&env), vec![other.as_str().to_string()], "{env:?}");
1617 assert_eq!(
1618 env.last(),
1619 Some(&(FORMATION_ID_ENV.to_string(), other.as_str().to_string())),
1620 "the formation id is the last runtime-owned name: {env:?}"
1621 );
1622 }
1623 std::env::remove_var(FORMATION_ID_ENV);
1624 }
1625
1626 #[test]
1629 fn every_child_is_told_host_warnings_were_reported() {
1630 std::env::set_var(HOST_WARNINGS_REPORTED_ENV, "0");
1631 let handle = SpawnerHandle::for_delegation(&spawner(), "dlg_0001".to_string());
1632 for (spawner, handle) in [(Some(spawner()), Some(&handle)), (None, None)] {
1633 for allow in [&[][..], &[HOST_WARNINGS_REPORTED_ENV][..]] {
1634 let env = child_environment(&request(allow, spawner.clone()), handle);
1635 let values: Vec<&str> = env
1636 .iter()
1637 .filter(|(key, _)| key == HOST_WARNINGS_REPORTED_ENV)
1638 .map(|(_, value)| value.as_str())
1639 .collect();
1640 assert_eq!(values, vec!["1"], "{env:?}");
1641 }
1642 }
1643 std::env::remove_var(HOST_WARNINGS_REPORTED_ENV);
1644 }
1645
1646 fn values_of<'a>(env: &'a [(String, String)], name: &str) -> Vec<&'a str> {
1647 env.iter()
1648 .filter(|(key, _)| key == name)
1649 .map(|(_, value)| value.as_str())
1650 .collect()
1651 }
1652
1653 #[test]
1656 fn a_child_carries_the_parents_account_names() {
1657 let _guard = crate::shell::ACCOUNT_ENV_LOCK
1658 .lock()
1659 .unwrap_or_else(|poisoned| poisoned.into_inner());
1660 std::env::set_var("USER", "murmur-parent-user");
1661 std::env::set_var("LOGNAME", "murmur-parent-login");
1662 for allow in [&[][..], &["USER"][..], &["USER", "LOGNAME"][..]] {
1663 let env = child_environment(&request(allow, None), None);
1664 assert_eq!(
1665 values_of(&env, "USER"),
1666 vec!["murmur-parent-user"],
1667 "{env:?}"
1668 );
1669 assert_eq!(
1670 values_of(&env, "LOGNAME"),
1671 vec!["murmur-parent-login"],
1672 "{env:?}"
1673 );
1674 }
1675 }
1676
1677 #[test]
1679 fn a_child_of_a_parent_without_account_names_has_none() {
1680 let _guard = crate::shell::ACCOUNT_ENV_LOCK
1681 .lock()
1682 .unwrap_or_else(|poisoned| poisoned.into_inner());
1683 let saved: Vec<_> = crate::shell::ACCOUNT_ENV
1684 .iter()
1685 .map(|name| (*name, std::env::var_os(name)))
1686 .collect();
1687 for name in crate::shell::ACCOUNT_ENV {
1688 std::env::remove_var(name);
1689 }
1690 let env = child_environment(&request(&["USER", "LOGNAME"], None), None);
1691 for (name, value) in saved {
1692 if let Some(value) = value {
1693 std::env::set_var(name, value);
1694 }
1695 }
1696 assert!(values_of(&env, "USER").is_empty(), "{env:?}");
1697 assert!(values_of(&env, "LOGNAME").is_empty(), "{env:?}");
1698 }
1699
1700 #[test]
1704 fn a_tail_read_after_the_end_waits_for_the_drain_to_catch_up() {
1705 let tail = Arc::new(StderrTail {
1706 state: Mutex::new(StderrState::default()),
1707 drained: Condvar::new(),
1708 });
1709
1710 let writer = Arc::clone(&tail);
1711 std::thread::spawn(move || {
1712 std::thread::sleep(Duration::from_millis(200));
1713 lock(&writer.state)
1714 .lines
1715 .push("error[E-CAP-003]: not achievable on this host".to_string());
1716 writer.finish();
1717 });
1718
1719 assert!(tail.lines().is_empty());
1720 assert_eq!(
1721 tail.lines_at_end(),
1722 vec!["error[E-CAP-003]: not achievable on this host".to_string()]
1723 );
1724 }
1725
1726 #[test]
1729 fn a_tail_with_no_pipe_behind_it_is_at_its_end_already() {
1730 let tail = StderrTail {
1731 state: Mutex::new(StderrState::default()),
1732 drained: Condvar::new(),
1733 };
1734 tail.finish();
1735
1736 let started = Instant::now();
1737 assert!(tail.lines_at_end().is_empty());
1738 assert!(
1739 started.elapsed() < CHILD_STDERR_DRAIN_TIMEOUT,
1740 "{started:?}"
1741 );
1742 }
1743
1744 #[test]
1745 fn child_launch_reads_the_operator_token_from_the_readiness_line() {
1746 let report = serde_json::json!({
1747 "url": "localhost:1",
1748 "session_id": "ses_x",
1749 "tokens": {"operator": "mdt1.a.b", "watcher": "mdt1.c.d"},
1750 });
1751 assert_eq!(
1752 readiness_door_token(&report).map(|token| token.expose().to_string()),
1753 Some("mdt1.a.b".to_string())
1754 );
1755 let public = serde_json::json!({"url": "localhost:1", "session_id": "ses_x"});
1756 assert_eq!(readiness_door_token(&public), None);
1757 }
1758
1759 #[test]
1760 fn child_launch_debug_prints_no_door_token() {
1761 let child = LaunchedChild {
1762 workdir: PathBuf::from("/tmp/child"),
1763 workdir_root: WorkdirRoot::open(&std::env::temp_dir()).unwrap(),
1764 session_id: "ses_child".to_string(),
1765 records_dir: PathBuf::new(),
1766 records_root: WorkdirRoot::open(&std::env::temp_dir()).unwrap(),
1767 capsule_url: "http://localhost:1".to_string(),
1768 argv: Vec::new(),
1769 env: Vec::new(),
1770 delegation_id: None,
1771 formation_id: None,
1772 task_id: None,
1773 door_token: Some(crate::door_auth::DoorToken::new(
1774 "mdt1.secretpayload.secretmac".to_string(),
1775 )),
1776 process: Arc::new(Mutex::new(ChildProcess {
1777 child: None,
1778 deliberate: false,
1779 status: None,
1780 terminated_at: None,
1781 #[cfg(unix)]
1782 lifeline: None,
1783 })),
1784 stderr_tail: Arc::new(StderrTail {
1785 state: Mutex::new(StderrState::default()),
1786 drained: Condvar::new(),
1787 }),
1788 started: Instant::now(),
1789 watcher: None,
1790 };
1791 let debug = format!("{child:?}");
1792 assert!(!debug.contains("secretpayload"), "{debug}");
1793 assert!(debug.contains("door_token"), "{debug}");
1794 }
1795
1796 fn start_process_code() -> String {
1798 let source = include_str!("child_launch.rs");
1799 let start = source
1800 .find("pub(crate) fn start_process(")
1801 .expect("start_process is defined here");
1802 let end = start
1803 + source[start..]
1804 .find("\n}\n")
1805 .expect("start_process ends at a closing brace");
1806 source[start..end]
1807 .lines()
1808 .filter(|line| !line.trim_start().starts_with("//"))
1809 .collect::<Vec<_>>()
1810 .join("\n")
1811 }
1812
1813 #[test]
1818 fn start_process_registers_no_fd_hygiene() {
1819 let code = start_process_code();
1820 for name in [
1821 "prepare_enforcement",
1822 "mark_inherited_fds_cloexec",
1823 "close_range",
1824 ] {
1825 assert!(!code.contains(name), "start_process names {name}:\n{code}");
1826 }
1827 let handed = code
1828 .rfind("lifeline.hand_to")
1829 .expect("the lifeline is handed");
1830 let kept = code
1831 .find("keep_across_exec")
1832 .expect("an inherited descriptor is kept");
1833 assert!(kept < handed, "the lifeline is registered last:\n{code}");
1834 assert!(
1835 !code[handed..].contains("pre_exec"),
1836 "nothing is registered after the lifeline:\n{code}"
1837 );
1838 }
1839
1840 #[cfg(unix)]
1841 fn process_with_lifeline(program: &str, args: &[&str]) -> (ChildProcess, i32) {
1842 let lifeline = crate::lifeline::ChildLifeline::new().unwrap();
1843 let write = lifeline.write_fd().unwrap();
1844 let child = Command::new(program).args(args).spawn().unwrap();
1845 (
1846 ChildProcess {
1847 child: Some(child),
1848 deliberate: false,
1849 status: None,
1850 terminated_at: None,
1851 lifeline: Some(lifeline),
1852 },
1853 write,
1854 )
1855 }
1856
1857 #[cfg(unix)]
1859 #[test]
1860 fn ending_a_child_closes_its_lifeline() {
1861 let (process, _) = process_with_lifeline("sleep", &["30"]);
1862 let process = Mutex::new(process);
1863 end_process(&process).unwrap();
1864 assert!(lock(&process).lifeline.is_none());
1865 assert!(lock(&process).status.is_some());
1866 end_process(&process).unwrap();
1867 }
1868
1869 #[cfg(unix)]
1872 #[test]
1873 fn ending_a_child_sends_sigterm_and_waits_for_it_to_wind_down() {
1874 let dir = tempfile::tempdir().unwrap();
1875 let marker = dir.path().join("wound-down");
1876 let script = format!(
1877 "trap 'echo wound-down > {}; exit 7' TERM; sleep 30 & wait",
1878 marker.display()
1879 );
1880 let (process, _) = process_with_lifeline("bash", &["-c", &script]);
1881 let process = Mutex::new(process);
1882 std::thread::sleep(Duration::from_millis(300));
1884 let began = Instant::now();
1885 end_process(&process).unwrap();
1886 assert!(
1887 began.elapsed() < CHILD_END_GRACE,
1888 "took {:?}",
1889 began.elapsed()
1890 );
1891 assert_eq!(
1892 std::fs::read_to_string(&marker).unwrap().trim(),
1893 "wound-down"
1894 );
1895 assert_eq!(
1896 lock(&process).status.and_then(|status| status.code()),
1897 Some(7)
1898 );
1899 assert!(lock(&process).lifeline.is_none());
1900 }
1901
1902 #[cfg(unix)]
1904 #[test]
1905 fn a_child_that_ignores_sigterm_is_killed_at_the_grace() {
1906 use std::os::unix::process::ExitStatusExt as _;
1907 let (process, _) =
1908 process_with_lifeline("bash", &["-c", "trap '' TERM; while :; do sleep 0.1; done"]);
1909 let process = Mutex::new(process);
1910 std::thread::sleep(Duration::from_millis(300));
1911 let began = Instant::now();
1912 end_process(&process).unwrap();
1913 let took = began.elapsed();
1914 assert!(took >= CHILD_END_GRACE, "killed before the grace: {took:?}");
1915 assert!(
1916 took < CHILD_END_GRACE + Duration::from_secs(5),
1917 "took {took:?}"
1918 );
1919 assert_eq!(
1920 lock(&process).status.and_then(|status| status.signal()),
1921 Some(libc::SIGKILL)
1922 );
1923 }
1924
1925 #[cfg(unix)]
1928 #[test]
1929 fn children_signalled_together_wait_one_grace_between_them() {
1930 let wedged = || {
1931 let (process, _) =
1932 process_with_lifeline("bash", &["-c", "trap '' TERM; while :; do sleep 0.1; done"]);
1933 LaunchedChild {
1934 workdir: PathBuf::from("/tmp/child"),
1935 workdir_root: WorkdirRoot::open(&std::env::temp_dir()).unwrap(),
1936 session_id: "ses_child".to_string(),
1937 records_dir: PathBuf::new(),
1938 records_root: WorkdirRoot::open(&std::env::temp_dir()).unwrap(),
1939 capsule_url: String::new(),
1940 argv: Vec::new(),
1941 env: Vec::new(),
1942 delegation_id: None,
1943 formation_id: None,
1944 task_id: None,
1945 door_token: None,
1946 process: Arc::new(Mutex::new(process)),
1947 stderr_tail: Arc::new(StderrTail {
1948 state: Mutex::new(StderrState::default()),
1949 drained: Condvar::new(),
1950 }),
1951 started: Instant::now(),
1952 watcher: None,
1953 }
1954 };
1955 let children = vec![wedged(), wedged()];
1956 std::thread::sleep(Duration::from_millis(300));
1957 let began = Instant::now();
1958 for child in &children {
1959 child.begin_ending();
1960 }
1961 drop(children);
1962 let took = began.elapsed();
1963 assert!(took >= CHILD_END_GRACE, "took {took:?}");
1964 assert!(
1965 took < CHILD_END_GRACE * 2,
1966 "two children waited a grace each: {took:?}"
1967 );
1968 }
1969
1970 #[cfg(unix)]
1972 #[test]
1973 fn a_child_seen_exited_has_its_lifeline_closed() {
1974 let (mut process, _) = process_with_lifeline("true", &[]);
1975 let deadline = Instant::now() + Duration::from_secs(10);
1976 let ending = loop {
1977 if let Some(ending) = process.poll() {
1978 break ending;
1979 }
1980 assert!(Instant::now() < deadline, "`true` never exited");
1981 std::thread::sleep(Duration::from_millis(20));
1982 };
1983 assert!(matches!(ending, Ending::Exited(Some(_))));
1984 assert!(process.lifeline.is_none());
1985 }
1986
1987 #[cfg(unix)]
1992 #[test]
1993 #[allow(unsafe_code)]
1994 fn a_childs_lifeline_closes_when_its_handle_drops_or_its_end_is_seen() {
1995 let launched = |program: &str, args: &[&str]| {
1996 let (process, write) = process_with_lifeline(program, args);
1997 let process = Arc::new(Mutex::new(process));
1998 let child = LaunchedChild {
1999 workdir: PathBuf::from("/tmp/child"),
2000 workdir_root: WorkdirRoot::open(&std::env::temp_dir()).unwrap(),
2001 session_id: "ses_child".to_string(),
2002 records_dir: PathBuf::new(),
2003 records_root: WorkdirRoot::open(&std::env::temp_dir()).unwrap(),
2004 capsule_url: String::new(),
2005 argv: Vec::new(),
2006 env: Vec::new(),
2007 delegation_id: None,
2008 formation_id: None,
2009 task_id: None,
2010 door_token: None,
2011 process: Arc::clone(&process),
2012 stderr_tail: Arc::new(StderrTail {
2013 state: Mutex::new(StderrState::default()),
2014 drained: Condvar::new(),
2015 }),
2016 started: Instant::now(),
2017 watcher: None,
2018 };
2019 (child, process, write)
2020 };
2021 let is_open = |fd: i32| {
2022 unsafe { libc::fcntl(fd, libc::F_GETFD) >= 0 }
2024 };
2025
2026 let (child, process, write) = launched("sleep", &["30"]);
2028 assert!(!child.has_exited());
2029 assert!(is_open(write), "a held child's write end is open");
2030 drop(child);
2031 assert!(
2032 lock(&process).lifeline.is_none(),
2033 "dropping the handle closes it"
2034 );
2035 assert!(
2036 lock(&process).deliberate,
2037 "a running child's drop is deliberate"
2038 );
2039 assert!(lock(&process).status.is_some(), "and reaps the child");
2041
2042 let (child, process, _) = launched("true", &[]);
2044 let deadline = Instant::now() + Duration::from_secs(10);
2045 while !child.has_exited() {
2046 assert!(Instant::now() < deadline, "`true` never exited");
2047 std::thread::sleep(Duration::from_millis(20));
2048 }
2049 assert!(
2050 lock(&process).lifeline.is_none(),
2051 "seeing the end closes it"
2052 );
2053 drop(child);
2054 assert!(
2055 !lock(&process).deliberate,
2056 "a child that ended on its own is not recorded as ended by its parent"
2057 );
2058 }
2059
2060 #[cfg(unix)]
2063 #[test]
2064 fn a_child_its_parent_ends_is_recorded_in_the_shared_words() {
2065 let dir = tempfile::tempdir().unwrap();
2066 let (process, _) = process_with_lifeline("sleep", &["30"]);
2067 let mut child = LaunchedChild {
2068 workdir: dir.path().to_path_buf(),
2069 workdir_root: WorkdirRoot::open(dir.path()).unwrap(),
2070 session_id: "ses_child".to_string(),
2071 records_dir: dir.path().to_path_buf(),
2072 records_root: WorkdirRoot::open(dir.path()).unwrap(),
2073 capsule_url: String::new(),
2074 argv: Vec::new(),
2075 env: Vec::new(),
2076 delegation_id: Some("dlg_watched".to_string()),
2077 formation_id: None,
2078 task_id: None,
2079 door_token: None,
2080 process: Arc::new(Mutex::new(process)),
2081 stderr_tail: Arc::new(StderrTail {
2082 state: Mutex::new(StderrState::default()),
2083 drained: Condvar::new(),
2084 }),
2085 started: Instant::now(),
2086 watcher: None,
2087 };
2088 let watcher = watch_for_completion(
2089 &child,
2090 SpawnerHandle::for_delegation(&spawner(), "dlg_watched".to_string()),
2091 CompletionAddress {
2092 url: "http://127.0.0.1:1".to_string(),
2093 },
2094 "worker",
2095 "0.1.0",
2096 None,
2097 )
2098 .expect("the watcher starts");
2099
2100 child.shutdown().unwrap();
2101 watcher.join().unwrap();
2102
2103 let recorded = delegation::read_completion(&WorkdirRoot::open(dir.path()).unwrap())
2104 .expect("the ending is recorded");
2105 assert_eq!(recorded.status, DelegationStatus::Terminated);
2106 assert_eq!(recorded.reported_by, Reporter::Launcher);
2107 assert!(!recorded.delivered);
2108 assert_eq!(
2109 recorded.detail.as_deref(),
2110 Some(delegation::PARENT_ENDED_DETAIL)
2111 );
2112 }
2113
2114 #[cfg(unix)]
2118 #[test]
2119 fn a_completion_the_child_records_while_being_ended_is_overwritten_by_the_launcher() {
2120 let dir = tempfile::tempdir().unwrap();
2121 let own = DelegationOutcome {
2122 delegation_id: "dlg_watched".to_string(),
2123 capsule_name: "worker".to_string(),
2124 capsule_version: "0.1.0".to_string(),
2125 session_id: "ses_child".to_string(),
2126 status: DelegationStatus::Error,
2127 result_path: None,
2128 workdir: dir.path().display().to_string(),
2129 duration_ms: 1,
2130 detail: None,
2131 reported_by: Reporter::Child,
2132 delivered: false,
2133 delivery_error: None,
2134 };
2135 let staged = dir.path().join("own-record.json");
2136 std::fs::write(&staged, serde_json::to_vec(&own).unwrap()).unwrap();
2137 let script = format!(
2138 "trap 'cp {} {}; exit 0' TERM; sleep 30 & wait",
2139 staged.display(),
2140 dir.path().join(delegation::COMPLETION_FILE).display()
2141 );
2142 let (process, _) = process_with_lifeline("bash", &["-c", &script]);
2143 let mut child = LaunchedChild {
2144 workdir: dir.path().to_path_buf(),
2145 workdir_root: WorkdirRoot::open(dir.path()).unwrap(),
2146 session_id: "ses_child".to_string(),
2147 records_dir: dir.path().to_path_buf(),
2148 records_root: WorkdirRoot::open(dir.path()).unwrap(),
2149 capsule_url: String::new(),
2150 argv: Vec::new(),
2151 env: Vec::new(),
2152 delegation_id: Some("dlg_watched".to_string()),
2153 formation_id: None,
2154 task_id: None,
2155 door_token: None,
2156 process: Arc::new(Mutex::new(process)),
2157 stderr_tail: Arc::new(StderrTail {
2158 state: Mutex::new(StderrState::default()),
2159 drained: Condvar::new(),
2160 }),
2161 started: Instant::now(),
2162 watcher: None,
2163 };
2164 let watcher = watch_for_completion(
2165 &child,
2166 SpawnerHandle::for_delegation(&spawner(), "dlg_watched".to_string()),
2167 CompletionAddress {
2168 url: "http://127.0.0.1:1".to_string(),
2169 },
2170 "worker",
2171 "0.1.0",
2172 None,
2173 )
2174 .expect("the watcher starts");
2175 std::thread::sleep(Duration::from_millis(300));
2176
2177 child.shutdown().unwrap();
2178 watcher.join().unwrap();
2179 let recorded = delegation::read_completion(&WorkdirRoot::open(dir.path()).unwrap())
2180 .expect("the ending is recorded");
2181 assert_eq!(recorded.status, DelegationStatus::Terminated);
2182 assert_eq!(recorded.reported_by, Reporter::Launcher);
2183 assert!(!recorded.delivered);
2184 assert_eq!(recorded.delivery_error, None);
2185 }
2186}