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, DEFAULT_DIR_MODE};
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_accessible_root: WorkdirRoot,
152 pub capsule_name: String,
153 pub capsule_version: String,
154 pub grant: SpawnApproval,
156 pub child_env_allow: Vec<String>,
160 pub roost_url: String,
162 pub spawner: Option<Spawner>,
171 pub completion_deadline: Option<Duration>,
179 pub formation_id: Option<FormationId>,
183}
184
185enum Ending {
187 Exited(Option<ExitStatus>),
190 Deliberate,
193}
194
195struct ChildProcess {
202 child: Option<Child>,
204 deliberate: bool,
208 status: Option<ExitStatus>,
210 terminated_at: Option<Instant>,
214 #[cfg(unix)]
219 lifeline: Option<crate::lifeline::ChildLifeline>,
220}
221
222impl ChildProcess {
223 fn close_lifeline(&mut self) {
225 #[cfg(unix)]
226 if let Some(mut lifeline) = self.lifeline.take() {
227 lifeline.close();
228 }
229 }
230
231 fn poll(&mut self) -> Option<Ending> {
236 if self.deliberate {
237 return (!self.is_being_ended()).then_some(Ending::Deliberate);
240 }
241 let Some(child) = self.child.as_mut() else {
242 return Some(Ending::Exited(self.status));
245 };
246 let ending = match child.try_wait() {
247 Ok(Some(status)) => {
248 self.status = Some(status);
249 Ending::Exited(Some(status))
250 }
251 Ok(None) => return None,
252 Err(_) => Ending::Exited(self.status),
255 };
256 self.close_lifeline();
257 Some(ending)
258 }
259
260 #[allow(unsafe_code)]
267 fn terminate(&mut self) {
268 if self.status.is_none() {
269 self.deliberate = true;
270 }
271 if self.terminated_at.is_some() {
272 return;
273 }
274 let Some(child) = self.child.as_mut() else {
275 return;
276 };
277 if let Ok(Some(status)) = child.try_wait() {
278 self.status = Some(status);
279 return;
280 }
281 #[cfg(unix)]
282 if let Ok(pid) = libc::pid_t::try_from(child.id()) {
283 unsafe {
287 libc::kill(pid, libc::SIGTERM);
288 }
289 }
290 #[cfg(not(unix))]
291 let _ = child.kill();
292 self.terminated_at = Some(Instant::now());
293 }
294
295 fn is_being_ended(&self) -> bool {
298 self.deliberate && self.status.is_none() && self.terminated_at.is_some()
299 }
300}
301
302fn end_process(process: &Mutex<ChildProcess>) -> Result<(), std::io::Error> {
312 let (mut child, signalled) = {
313 let mut process = lock(process);
314 process.terminate();
315 let Some(child) = process.child.take() else {
316 if process.status.is_some() || process.terminated_at.is_none() {
319 process.close_lifeline();
320 }
321 return Ok(());
322 };
323 (child, process.terminated_at)
324 };
325 let waited = match signalled {
326 None => child.wait(),
327 Some(signalled) => loop {
328 match child.try_wait() {
329 Ok(Some(status)) => break Ok(status),
330 Ok(None) if signalled.elapsed() < CHILD_END_GRACE => {
331 std::thread::sleep(CHILD_END_POLL);
332 }
333 _ => {
337 let _ = child.kill();
338 break child.wait();
339 }
340 }
341 },
342 };
343 let mut process = lock(process);
344 process.close_lifeline();
345 match waited {
346 Ok(status) => {
347 process.status = Some(status);
348 Ok(())
349 }
350 Err(error) => {
352 process.terminated_at = None;
353 Err(error)
354 }
355 }
356}
357
358pub struct LaunchedChild {
360 pub workdir: PathBuf,
363 pub(crate) workdir_rel: PathBuf,
366 pub(crate) parent_root: WorkdirRoot,
368 pub session_id: String,
370 pub records_dir: PathBuf,
374 pub(crate) records_root: Option<WorkdirRoot>,
378 pub capsule_url: String,
381 pub argv: Vec<String>,
386 pub env: Vec<(String, String)>,
389 pub delegation_id: Option<String>,
394 pub door_token: Option<crate::door_auth::DoorToken>,
398 pub formation_id: Option<FormationId>,
401 process: Arc<Mutex<ChildProcess>>,
402 stderr_tail: Arc<StderrTail>,
404 started: Instant,
406 watcher: Option<std::thread::JoinHandle<()>>,
408}
409
410impl std::fmt::Debug for LaunchedChild {
411 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
412 f.debug_struct("LaunchedChild")
413 .field("workdir", &self.workdir)
414 .field("session_id", &self.session_id)
415 .field("records_dir", &self.records_dir)
416 .field("capsule_url", &self.capsule_url)
417 .field("delegation_id", &self.delegation_id)
418 .field("formation_id", &self.formation_id)
419 .field(
420 "door_token",
421 &self.door_token.as_ref().map(|_| "<redacted>"),
422 )
423 .field("pid", &self.pid())
424 .finish()
425 }
426}
427
428impl LaunchedChild {
429 pub fn pid(&self) -> u32 {
431 lock(&self.process)
432 .child
433 .as_ref()
434 .map(Child::id)
435 .unwrap_or(0)
436 }
437
438 pub fn stderr_tail(&self) -> Vec<String> {
443 self.stderr_tail.lines()
444 }
445
446 pub fn begin_ending(&self) {
452 lock(&self.process).terminate();
453 }
454
455 pub fn shutdown(&mut self) -> Result<(), RuntimeError> {
461 end_process(&self.process).map_err(|error| {
462 RuntimeError::Runtime(format!("failed to reap child capsule: {error}"))
463 })
464 }
465
466 pub fn is_being_ended(&self) -> bool {
472 lock(&self.process).is_being_ended()
473 }
474
475 pub fn watcher_finished(&self) -> bool {
482 self.watcher
483 .as_ref()
484 .is_none_or(std::thread::JoinHandle::is_finished)
485 }
486
487 pub fn has_exited(&self) -> bool {
493 lock(&self.process).poll().is_some()
494 }
495}
496
497impl Drop for LaunchedChild {
498 fn drop(&mut self) {
502 let _ = end_process(&self.process);
503 }
504}
505
506pub(crate) fn workdir_relative_to(workdir: &Path, root: &Path) -> String {
511 workdir
512 .strip_prefix(root)
513 .unwrap_or(workdir)
514 .to_string_lossy()
515 .replace('\\', "/")
516}
517
518fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
521 mutex
522 .lock()
523 .unwrap_or_else(|poisoned| poisoned.into_inner())
524}
525
526pub fn child_workdir_for(parent_accessible_workdir: &Path, capsule_name: &str) -> PathBuf {
534 parent_accessible_workdir.join(child_workdir_rel(capsule_name))
535}
536
537pub(crate) const CHILDREN_DIR: &str = ".murmur/children";
540
541fn child_workdir_rel(capsule_name: &str) -> PathBuf {
544 let suffix = mac_token::random_hex(8).unwrap_or_else(|_| {
545 format!(
548 "{:016x}",
549 std::time::SystemTime::now()
550 .duration_since(std::time::UNIX_EPOCH)
551 .map(|elapsed| elapsed.as_nanos() as u64)
552 .unwrap_or(0)
553 )
554 });
555 Path::new(CHILDREN_DIR).join(format!(
556 "{}-{suffix}",
557 murmur_artifact::tool_name_of(capsule_name)
558 ))
559}
560
561pub fn launch_child_capsule(request: ChildLaunchRequest) -> Result<LaunchedChild, RuntimeError> {
566 #[cfg(unix)]
569 let mut lifeline = crate::lifeline::ChildLifeline::new().map_err(|error| {
570 RuntimeError::Runtime(format!(
571 "the child capsule's spawner lifeline could not be created: {error}"
572 ))
573 })?;
574 let workdir_rel = child_workdir_rel(&request.capsule_name);
575 let identity = create_child_dir(&request.parent_accessible_root, &workdir_rel)?;
576 let workdir = request.parent_accessible_root.host_path(&workdir_rel);
577
578 let binary = mur_binary()?;
579 let argv: Vec<String> = vec![
580 binary.display().to_string(),
581 "run".to_string(),
582 "--capsule".to_string(),
583 request.capsule_name.clone(),
584 "--capsule-version".to_string(),
585 request.capsule_version.clone(),
586 "--workdir".to_string(),
587 workdir.display().to_string(),
588 WORKDIR_IDENTITY_FLAG.to_string(),
589 format_workdir_identity(identity),
590 "--json".to_string(),
591 READINESS_TOKENS_FLAG.to_string(),
592 "--no-env-file".to_string(),
593 "--spawn-grant-stdin".to_string(),
594 ];
595 let handle = request
598 .spawner
599 .as_ref()
600 .map(|spawner| SpawnerHandle::for_delegation(spawner, delegation::new_delegation_id()));
601 #[cfg_attr(not(unix), allow(unused_mut))]
602 let mut env = child_environment(&request, handle.as_ref());
603 #[cfg(unix)]
606 if let Some((name, value)) = lifeline.env_pair() {
607 let at = env
608 .iter()
609 .position(|(key, _)| key == SPAWNER_ENV || key == FORMATION_ID_ENV)
610 .unwrap_or(env.len());
611 env.insert(at, (name.to_string(), value));
612 }
613 let started = Instant::now();
614
615 let StartedProcess {
619 child,
620 stderr_tail,
621 stdin_written: write_result,
622 } = start_process(&ProcessLaunch {
623 binary: &binary,
624 args: &argv[1..],
625 cwd: Some(&workdir),
626 env: ProcessEnv::Cleared(&env),
627 stdin_line: Some(request.grant.expose()),
628 stderr_prefix: None,
629 inherit_output: false,
630 own_process_group: false,
631 inherit_fd: None,
632 #[cfg(unix)]
633 lifeline: Some(Lifeline::Child(&lifeline)),
634 })
635 .map_err(|error| {
636 RuntimeError::Runtime(format!(
637 "failed to start the child capsule process '{}': {error}",
638 binary.display()
639 ))
640 })?;
641 #[cfg(unix)]
642 lifeline.spawned();
643 let write_result =
644 write_result.map_err(|error| format!("failed to hand the child its launch grant: {error}"));
645 let mut launched = LaunchedChild {
646 workdir,
647 workdir_rel,
648 parent_root: request.parent_accessible_root.clone(),
649 session_id: String::new(),
650 records_dir: PathBuf::new(),
651 records_root: None,
652 capsule_url: String::new(),
653 argv,
654 env,
655 delegation_id: handle.as_ref().map(|handle| handle.delegation_id.clone()),
656 door_token: None,
657 formation_id: None,
658 process: Arc::new(Mutex::new(ChildProcess {
659 child: Some(child),
660 deliberate: false,
661 status: None,
662 terminated_at: None,
663 #[cfg(unix)]
664 lifeline: Some(lifeline),
665 })),
666 stderr_tail: Arc::clone(&stderr_tail),
667 started,
668 watcher: None,
669 };
670 if let Err(reason) = write_result {
671 return Err(RuntimeError::Runtime(reason));
672 }
673
674 let line = first_json_line(&mut launched, &stderr_tail)?;
675 let report: serde_json::Value = serde_json::from_str(&line).map_err(|error| {
676 RuntimeError::Runtime(format!(
677 "the child capsule's first --json line did not parse: {error}"
678 ))
679 })?;
680 let session_id = report
681 .get("session_id")
682 .and_then(serde_json::Value::as_str)
683 .ok_or_else(|| {
684 RuntimeError::Runtime(
685 "the child capsule's --json line carried no session_id".to_string(),
686 )
687 })?;
688 let url = report
689 .get("url")
690 .and_then(serde_json::Value::as_str)
691 .unwrap_or_default();
692
693 launched.session_id = session_id.to_string();
694 launched.records_dir = readiness_records_dir(&report, session_id)?;
695 launched.records_root = Some(
696 WorkdirRoot::open_sharing_latch_absolute(&launched.records_dir, &launched.parent_root)
697 .map_err(|error| {
698 RuntimeError::Runtime(format!(
699 "the child capsule's records directory could not be opened: {error}"
700 ))
701 })?,
702 );
703 launched.door_token = readiness_door_token(&report);
704 launched.formation_id = report
705 .get("formation_id")
706 .and_then(serde_json::Value::as_str)
707 .and_then(|id| FormationId::parse(id).ok());
708 launched.capsule_url = if url.is_empty() {
711 String::new()
712 } else {
713 format!("http://{url}")
714 };
715
716 if let Some(handle) = handle {
720 if let Some(address) = handle.report_to.clone() {
721 launched.watcher = watch_for_completion(
722 &launched,
723 handle,
724 address,
725 &request.capsule_name,
726 &request.capsule_version,
727 request.completion_deadline,
728 );
729 }
730 }
731 Ok(launched)
732}
733
734pub(crate) fn readiness_records_dir(
739 report: &serde_json::Value,
740 session_id: &str,
741) -> Result<PathBuf, RuntimeError> {
742 let refused = |why: String| {
743 RuntimeError::Runtime(format!(
744 "the child capsule's --json line {why}, so its outcome has nowhere to be read from"
745 ))
746 };
747 let reported = report
748 .get("records_dir")
749 .and_then(serde_json::Value::as_str)
750 .map(PathBuf::from)
751 .ok_or_else(|| refused("carried no records_dir".to_string()))?;
752 let records_root = crate::records_dir::records_dir_root()
753 .map_err(|error| refused(format!("named a records directory and this host's {error}")))?;
754 let well_formed = reported.is_absolute()
755 && reported.starts_with(&records_root)
756 && reported.file_name().and_then(|name| name.to_str()) == Some(session_id)
757 && !reported
758 .components()
759 .any(|part| matches!(part, std::path::Component::ParentDir));
760 if !well_formed {
761 return Err(refused(format!(
762 "named {} as its records directory, which is not {}/<workdir root>/{session_id}",
763 reported.display(),
764 records_root.display()
765 )));
766 }
767 Ok(reported)
768}
769
770pub(crate) fn readiness_door_token(
774 report: &serde_json::Value,
775) -> Option<crate::door_auth::DoorToken> {
776 report
777 .pointer("/tokens/operator")
778 .and_then(serde_json::Value::as_str)
779 .filter(|token| !token.is_empty())
780 .map(|token| crate::door_auth::DoorToken::new(token.to_string()))
781}
782
783fn watch_for_completion(
812 launched: &LaunchedChild,
813 handle: SpawnerHandle,
814 address: CompletionAddress,
815 capsule_name: &str,
816 capsule_version: &str,
817 deadline: Option<Duration>,
818) -> Option<std::thread::JoinHandle<()>> {
819 let process = Arc::clone(&launched.process);
820 let stderr_tail = Arc::clone(&launched.stderr_tail);
821 let workdir = launched.workdir.clone();
822 let Some(records_root) = launched.records_root.clone() else {
823 crate::runtime_err!(
824 "[capsule-runtime] delegation {}: the child reported no records directory, so its \
825 completion cannot be watched",
826 handle.delegation_id
827 );
828 return None;
829 };
830 let session_id = launched.session_id.clone();
831 let capsule_name = capsule_name.to_string();
832 let capsule_version = capsule_version.to_string();
833 let started = launched.started;
834
835 let watching_from = Instant::now();
836 let spawned = std::thread::Builder::new().spawn(move || {
837 let expires_at = deadline.map(|bound| watching_from + bound);
838 let mut expired = false;
841 let ending = loop {
842 if expires_at.is_some_and(|at| Instant::now() >= at) {
843 let _ = end_process(&process);
844 expired = true;
845 break Ending::Deliberate;
846 }
847 if let Some(ending) = lock(&process).poll() {
848 break ending;
849 }
850 std::thread::sleep(CHILD_WATCH_INTERVAL);
851 };
852 let duration_ms = started.elapsed().as_millis().try_into().unwrap_or(u64::MAX);
853
854 let recorded = match ending {
858 Ending::Exited(_) => delegation::read_completion(&records_root),
859 Ending::Deliberate => None,
860 };
861 if let Some(recorded) = recorded {
862 if recorded.delivered {
863 return;
864 }
865 let retried =
866 delegation::report_completion(&handle, &address, recorded, &records_root);
867 if !retried.delivered {
868 crate::runtime_err!(
870 "[capsule-runtime] delegation {}: the child's completion is undelivered after one retry",
871 handle.delegation_id
872 );
873 }
874 return;
875 }
876
877 let outcome = |status: DelegationStatus, detail: String| DelegationOutcome {
878 delegation_id: handle.delegation_id.clone(),
879 capsule_name: capsule_name.clone(),
880 capsule_version: capsule_version.clone(),
881 session_id: session_id.clone(),
882 status,
883 result_path: None,
886 workdir: workdir.display().to_string(),
887 duration_ms,
888 detail: Some(detail),
889 reported_by: Reporter::Launcher,
890 delivered: false,
891 delivery_error: None,
892 };
893
894 match ending {
895 Ending::Deliberate if expired => {
899 let bound = deadline.map(|bound| bound.as_secs()).unwrap_or_default();
900 delegation::report_completion(
901 &handle,
902 &address,
903 outcome(
904 DelegationStatus::Terminated,
905 format!(
906 "the capsule was still running after {bound}s and was ended at the \
907 delegation deadline"
908 ),
909 ),
910 &records_root,
911 );
912 }
913 Ending::Deliberate => {
914 delegation::record_terminated(
915 &records_root,
916 outcome(
917 DelegationStatus::Terminated,
918 delegation::PARENT_ENDED_DETAIL.to_string(),
919 ),
920 );
921 }
922 Ending::Exited(status) => {
923 let status_text = match status {
924 Some(status) => status.to_string(),
925 None => "unknown exit status".to_string(),
926 };
927 let tail = stderr_tail.lines_at_end().join("\n");
928 let detail = if tail.is_empty() {
929 format!(
930 "the child process ended without recording a completion ({status_text})"
931 )
932 } else {
933 format!(
934 "the child process ended without recording a completion ({status_text}); \
935 its last stderr lines were:\n{tail}"
936 )
937 };
938 delegation::report_completion(
939 &handle,
940 &address,
941 outcome(DelegationStatus::Crashed, detail),
942 &records_root,
943 );
944 }
945 }
946 });
947 match spawned {
948 Ok(watcher) => Some(watcher),
949 Err(error) => {
950 crate::runtime_err!(
951 "[capsule-runtime] delegation {}: the child's completion watcher could not be \
952 started: {error}",
953 launched.delegation_id.as_deref().unwrap_or_default()
954 );
955 None
956 }
957 }
958}
959
960pub(crate) fn child_environment(
985 request: &ChildLaunchRequest,
986 handle: Option<&SpawnerHandle>,
987) -> Vec<(String, String)> {
988 let mut env: Vec<(String, String)> = Vec::new();
989 for key in &request.child_env_allow {
990 if let Ok(value) = std::env::var(key) {
991 env.push((key.clone(), value));
992 }
993 }
994 env.retain(|(key, _)| {
995 !matches!(key.as_str(), "PATH" | "HOME" | "MURMUR_ROOST_URL")
996 && !crate::shell::ACCOUNT_ENV.contains(&key.as_str())
997 && key != SPAWNER_ENV
998 && key != FORMATION_ID_ENV
999 && key != SPAWNER_LIFELINE_ENV
1000 && key != HOST_WARNINGS_REPORTED_ENV
1001 && !crate::formation::is_member_grant_env(key)
1002 });
1003
1004 if let Ok(path) = std::env::var("PATH") {
1005 env.push(("PATH".to_string(), path));
1006 }
1007 if let Ok(home) = std::env::var("HOME") {
1008 env.push(("HOME".to_string(), home));
1009 }
1010 env.extend(
1011 crate::shell::HostAccountEnv::from_host()
1012 .set()
1013 .iter()
1014 .cloned(),
1015 );
1016 env.push(("MURMUR_ROOST_URL".to_string(), request.roost_url.clone()));
1017 env.push(crate::host_warnings::reported_env_pair());
1018 if let Some(handle) = handle {
1022 env.push((SPAWNER_ENV.to_string(), handle.to_env_value()));
1023 }
1024 if let Some(formation_id) = &request.formation_id {
1026 let (name, value) = formation_id.env_pair();
1027 env.push((name.to_string(), value));
1028 }
1029 env
1030}
1031
1032pub(crate) fn mur_binary() -> Result<PathBuf, RuntimeError> {
1034 if let Some(path) = std::env::var_os(MUR_BINARY_ENV) {
1035 if !path.is_empty() {
1036 return Ok(PathBuf::from(path));
1037 }
1038 }
1039 std::env::current_exe().map_err(|error| {
1040 RuntimeError::Runtime(format!(
1041 "cannot locate the mur binary to launch a child capsule with: {error}"
1042 ))
1043 })
1044}
1045
1046fn create_child_dir(root: &WorkdirRoot, rel: &Path) -> Result<(u64, u64), RuntimeError> {
1051 root.create_dir_all(CHILDREN_DIR, DEFAULT_DIR_MODE)?;
1052 root.create_dir_all(rel, CHILD_DIR_MODE)?;
1053 let created = root.open_beneath(&rel.to_string_lossy())?.ok_or_else(|| {
1054 RuntimeError::Runtime(format!(
1055 "the child capsule's directory {} vanished after its creation",
1056 root.display_path(rel)
1057 ))
1058 })?;
1059 Ok(created.identity()?)
1060}
1061
1062pub(crate) struct StderrTail {
1066 state: Mutex<StderrState>,
1067 drained: Condvar,
1069}
1070
1071#[derive(Default)]
1072struct StderrState {
1073 lines: Vec<String>,
1074 at_eof: bool,
1076}
1077
1078impl StderrTail {
1079 pub(crate) fn lines(&self) -> Vec<String> {
1082 lock(&self.state).lines.clone()
1083 }
1084
1085 pub(crate) fn lines_at_end(&self) -> Vec<String> {
1093 let (state, _) = self
1094 .drained
1095 .wait_timeout_while(lock(&self.state), CHILD_STDERR_DRAIN_TIMEOUT, |state| {
1096 !state.at_eof
1097 })
1098 .unwrap_or_else(|poisoned| poisoned.into_inner());
1099 state.lines.clone()
1100 }
1101
1102 fn finish(&self) {
1104 lock(&self.state).at_eof = true;
1105 self.drained.notify_all();
1106 }
1107}
1108
1109fn drain_stderr(child: &mut Child, prefix: Option<&str>) -> Arc<StderrTail> {
1115 let tail = Arc::new(StderrTail {
1116 state: Mutex::new(StderrState::default()),
1117 drained: Condvar::new(),
1118 });
1119 let Some(stderr) = child.stderr.take() else {
1120 tail.finish();
1123 return tail;
1124 };
1125 let collector = Arc::clone(&tail);
1126 let prefix = prefix.unwrap_or_default().to_string();
1127 std::thread::spawn(move || {
1128 for line in BufReader::new(stderr).lines().map_while(Result::ok) {
1129 crate::runtime_err!("{prefix}{line}");
1130 let mut state = lock(&collector.state);
1131 if state.lines.len() == CHILD_STDERR_TAIL_LINES {
1132 state.lines.remove(0);
1133 }
1134 state.lines.push(line);
1135 }
1136 collector.finish();
1137 });
1138 tail
1139}
1140
1141fn first_json_line(
1146 launched: &mut LaunchedChild,
1147 stderr_tail: &StderrTail,
1148) -> Result<String, RuntimeError> {
1149 let stdout = lock(&launched.process)
1150 .child
1151 .as_mut()
1152 .and_then(|child| child.stdout.take())
1153 .ok_or_else(|| {
1154 RuntimeError::Runtime("the child process exposed no standard output".to_string())
1155 })?;
1156
1157 let rx = relay_readiness_line(stdout, AfterReadiness::Stdout);
1158
1159 match rx.recv_timeout(CHILD_READY_TIMEOUT) {
1160 Ok(Some(line)) => Ok(line),
1161 Ok(None) => {
1162 let status = lock(&launched.process)
1163 .child
1164 .as_mut()
1165 .and_then(|child| child.wait().ok())
1166 .map(|status| status.to_string())
1167 .unwrap_or_else(|| "unknown".to_string());
1168 let reason = stderr_tail.lines_at_end().join("\n");
1172 Err(RuntimeError::Runtime(format!(
1173 "the child capsule '{}' exited without reporting a launch line ({status}): {reason}",
1174 launched.workdir.display()
1175 )))
1176 }
1177 Err(_) => Err(RuntimeError::Runtime(format!(
1178 "the child capsule '{}' did not report a launch line within {}s",
1179 launched.workdir.display(),
1180 CHILD_READY_TIMEOUT.as_secs()
1181 ))),
1182 }
1183}
1184
1185pub(crate) enum ProcessEnv<'a> {
1187 Cleared(&'a [(String, String)]),
1190 Inherited {
1193 set: &'a [(String, String)],
1194 remove: &'a [&'a str],
1195 },
1196}
1197
1198pub(crate) enum AfterReadiness {
1200 Stdout,
1202 PrefixedStderr(String),
1204}
1205
1206pub(crate) struct ProcessLaunch<'a> {
1208 pub(crate) binary: &'a Path,
1209 pub(crate) args: &'a [String],
1211 pub(crate) cwd: Option<&'a Path>,
1213 pub(crate) env: ProcessEnv<'a>,
1214 pub(crate) stdin_line: Option<&'a str>,
1217 pub(crate) stderr_prefix: Option<&'a str>,
1219 pub(crate) inherit_output: bool,
1222 pub(crate) own_process_group: bool,
1226 pub(crate) inherit_fd: Option<i32>,
1230 #[cfg(unix)]
1233 pub(crate) lifeline: Option<Lifeline<'a>>,
1234}
1235
1236#[cfg(unix)]
1238#[derive(Clone, Copy)]
1239pub(crate) enum Lifeline<'a> {
1240 Member(&'a crate::lifeline::MemberLifeline),
1241 Child(&'a crate::lifeline::ChildLifeline),
1242}
1243
1244#[cfg(unix)]
1245impl Lifeline<'_> {
1246 fn hand_to(self, command: &mut Command) {
1247 match self {
1248 Lifeline::Member(lifeline) => lifeline.hand_to(command),
1249 Lifeline::Child(lifeline) => lifeline.hand_to(command),
1250 }
1251 }
1252}
1253
1254pub(crate) struct StartedProcess {
1256 pub(crate) child: Child,
1258 pub(crate) stderr_tail: Arc<StderrTail>,
1259 pub(crate) stdin_written: Result<(), String>,
1261}
1262
1263pub(crate) fn start_process(launch: &ProcessLaunch<'_>) -> std::io::Result<StartedProcess> {
1266 let mut command = Command::new(launch.binary);
1267 command
1268 .args(launch.args)
1269 .stdin(if launch.stdin_line.is_some() {
1270 Stdio::piped()
1271 } else {
1272 Stdio::null()
1273 })
1274 .stdout(if launch.inherit_output {
1275 Stdio::inherit()
1276 } else {
1277 Stdio::piped()
1278 })
1279 .stderr(if launch.inherit_output {
1280 Stdio::inherit()
1281 } else {
1282 Stdio::piped()
1283 });
1284 if let Some(cwd) = launch.cwd {
1285 command.current_dir(cwd);
1286 }
1287 match launch.env {
1288 ProcessEnv::Cleared(pairs) => {
1289 command.env_clear();
1290 for (key, value) in pairs {
1291 command.env(key, value);
1292 }
1293 }
1294 ProcessEnv::Inherited { set, remove } => {
1295 for name in remove {
1296 command.env_remove(name);
1297 }
1298 for (key, value) in set {
1299 command.env(key, value);
1300 }
1301 }
1302 }
1303 #[cfg(unix)]
1304 if launch.own_process_group {
1305 use std::os::unix::process::CommandExt;
1306 command.process_group(0);
1307 }
1308 #[cfg(unix)]
1309 if let Some(fd) = launch.inherit_fd {
1310 keep_across_exec(&mut command, fd);
1311 }
1312 #[cfg(unix)]
1317 if let Some(lifeline) = launch.lifeline {
1318 lifeline.hand_to(&mut command);
1319 }
1320
1321 let mut child = command.spawn()?;
1322 let stdin_written = match launch.stdin_line {
1323 None => Ok(()),
1324 Some(line) => child
1325 .stdin
1326 .take()
1327 .ok_or_else(|| "the process exposed no standard input".to_string())
1328 .and_then(|mut stdin| writeln!(stdin, "{line}").map_err(|error| error.to_string())),
1329 };
1330 let stderr_tail = drain_stderr(&mut child, launch.stderr_prefix);
1331 Ok(StartedProcess {
1332 child,
1333 stderr_tail,
1334 stdin_written,
1335 })
1336}
1337
1338#[cfg(unix)]
1341#[allow(unsafe_code)]
1342fn keep_across_exec(command: &mut Command, fd: i32) {
1343 use std::os::unix::process::CommandExt;
1344 unsafe {
1348 command.pre_exec(move || {
1349 let flags = libc::fcntl(fd, libc::F_GETFD);
1350 if flags == -1 || libc::fcntl(fd, libc::F_SETFD, flags & !libc::FD_CLOEXEC) == -1 {
1351 return Err(std::io::Error::last_os_error());
1352 }
1353 Ok(())
1354 });
1355 }
1356}
1357
1358pub(crate) fn relay_readiness_line(
1369 stdout: std::process::ChildStdout,
1370 after: AfterReadiness,
1371) -> mpsc::Receiver<Option<String>> {
1372 let (tx, rx) = mpsc::channel::<Option<String>>();
1373 std::thread::spawn(move || {
1374 let mut reader = BufReader::new(stdout);
1375 let mut line = String::new();
1376 let first = match reader.read_line(&mut line) {
1377 Ok(0) | Err(_) => None,
1378 Ok(_) => Some(line.trim_end().to_string()),
1379 };
1380 let _ = tx.send(first);
1381 let mut rest = String::new();
1382 while let Ok(read) = reader.read_line(&mut rest) {
1383 if read == 0 {
1384 break;
1385 }
1386 match &after {
1387 AfterReadiness::Stdout => {
1388 crate::diagnostic::raw_to_stdout(&rest);
1389 }
1390 AfterReadiness::PrefixedStderr(prefix) => {
1391 crate::runtime_err!("{prefix}{}", rest.trim_end_matches(['\r', '\n']));
1392 }
1393 }
1394 rest.clear();
1395 }
1396 });
1397 rx
1398}
1399
1400#[cfg(test)]
1401mod tests {
1402 use super::*;
1403
1404 #[test]
1405 fn a_child_directory_is_unique_per_delegation() {
1406 let parent = PathBuf::from("/tmp/parent");
1407 let first = child_workdir_for(&parent, "worker");
1408 let second = child_workdir_for(&parent, "worker");
1409
1410 assert_ne!(first, second);
1411 assert_eq!(
1412 first.parent(),
1413 Some(parent.join(".murmur").join("children").as_path())
1414 );
1415 assert!(first
1416 .file_name()
1417 .unwrap()
1418 .to_string_lossy()
1419 .starts_with("worker-"));
1420 assert_eq!(
1422 first.file_name().unwrap().to_string_lossy().len(),
1423 "worker-".len() + 16
1424 );
1425 }
1426
1427 #[test]
1430 fn a_child_directory_is_named_by_the_bare_part_of_a_full_name() {
1431 let parent = PathBuf::from("/tmp/parent");
1432 let scoped = child_workdir_for(&parent, "local/worker");
1433
1434 assert_eq!(
1435 scoped.parent(),
1436 Some(parent.join(".murmur").join("children").as_path())
1437 );
1438 let name = scoped.file_name().unwrap().to_string_lossy().into_owned();
1439 assert!(name.starts_with("worker-"), "{name}");
1440 assert_eq!(name.len(), "worker-".len() + 16);
1441 }
1442
1443 fn request(env_allow: &[&str], spawner: Option<Spawner>) -> ChildLaunchRequest {
1444 ChildLaunchRequest {
1445 parent_accessible_root: WorkdirRoot::open(&std::env::temp_dir()).unwrap(),
1446 capsule_name: "worker".to_string(),
1447 capsule_version: "0.1.0".to_string(),
1448 grant: SpawnApproval::new("msa1.token".to_string()),
1449 child_env_allow: env_allow.iter().map(|name| name.to_string()).collect(),
1450 roost_url: "http://127.0.0.1:7700".to_string(),
1451 spawner,
1452 completion_deadline: None,
1453 formation_id: None,
1454 }
1455 }
1456
1457 fn decoy() -> String {
1464 SpawnerHandle {
1465 session_id: "ses_decoy".to_string(),
1466 context_id: "ctx_decoy".to_string(),
1467 delegation_id: "dlg_decoy".to_string(),
1468 report_to: None,
1469 }
1470 .to_env_value()
1471 }
1472
1473 fn spawner() -> Spawner {
1474 Spawner {
1475 session_id: "ses_parent".to_string(),
1476 context_id: "ctx_parent".to_string(),
1477 report_to: Some(CompletionAddress {
1478 url: "http://127.0.0.1:7000".to_string(),
1479 }),
1480 }
1481 }
1482
1483 #[test]
1484 fn a_childs_environment_is_built_from_its_own_declaration() {
1485 std::env::set_var("MURMUR_CHILD_LAUNCH_TEST_A", "a");
1486 std::env::set_var("MURMUR_CHILD_LAUNCH_TEST_B", "b");
1487
1488 let env = child_environment(&request(&["MURMUR_CHILD_LAUNCH_TEST_A"], None), None);
1489 let names: Vec<&str> = env.iter().map(|(key, _)| key.as_str()).collect();
1490
1491 assert!(names.contains(&"MURMUR_CHILD_LAUNCH_TEST_A"));
1492 assert!(!names.contains(&"MURMUR_CHILD_LAUNCH_TEST_B"));
1493 assert!(names.contains(&"MURMUR_ROOST_URL"));
1494 assert!(!env.iter().any(|(_, value)| value.contains("msa1.token")));
1495 }
1496
1497 #[test]
1498 fn a_runtime_owned_name_cannot_be_displaced_by_a_declaration() {
1499 std::env::set_var("MURMUR_ROOST_URL", "http://127.0.0.1:1");
1500
1501 let env = child_environment(&request(&["MURMUR_ROOST_URL"], None), None);
1502 let urls: Vec<&String> = env
1503 .iter()
1504 .filter(|(key, _)| key == "MURMUR_ROOST_URL")
1505 .map(|(_, value)| value)
1506 .collect();
1507
1508 assert_eq!(urls, vec!["http://127.0.0.1:7700"]);
1509 }
1510
1511 #[test]
1514 fn the_spawner_handle_is_injected_last_and_cannot_be_displaced() {
1515 std::env::set_var(SPAWNER_ENV, decoy());
1516 let handle = SpawnerHandle::for_delegation(&spawner(), "dlg_0001".to_string());
1517
1518 let env = child_environment(&request(&[SPAWNER_ENV], Some(spawner())), Some(&handle));
1519 let injected: Vec<&String> = env
1520 .iter()
1521 .filter(|(key, _)| key == SPAWNER_ENV)
1522 .map(|(_, value)| value)
1523 .collect();
1524
1525 assert_eq!(injected.len(), 1, "{env:?}");
1526 assert_eq!(
1527 SpawnerHandle::parse(injected[0]).expect("the injected value is a handle"),
1528 handle
1529 );
1530 assert_eq!(
1531 env.last().map(|(key, _)| key.as_str()),
1532 Some(SPAWNER_ENV),
1533 "the handle is applied in the runtime-owned tail: {env:?}"
1534 );
1535 std::env::remove_var(SPAWNER_ENV);
1536 }
1537
1538 #[test]
1540 fn a_launch_without_a_spawner_injects_no_handle() {
1541 std::env::set_var(SPAWNER_ENV, decoy());
1542
1543 let env = child_environment(&request(&[SPAWNER_ENV], None), None);
1544
1545 assert!(!env.iter().any(|(key, _)| key == SPAWNER_ENV), "{env:?}");
1546 std::env::remove_var(SPAWNER_ENV);
1547 }
1548
1549 #[test]
1552 fn the_formation_id_is_injected_last_and_cannot_be_displaced() {
1553 let _guard = crate::formation::FORMATION_ENV_LOCK
1554 .lock()
1555 .unwrap_or_else(|poisoned| poisoned.into_inner());
1556 let decoy = FormationId::mint();
1557 std::env::set_var(FORMATION_ID_ENV, decoy.as_str());
1558 let injected = |env: &[(String, String)]| -> Vec<String> {
1559 env.iter()
1560 .filter(|(key, _)| key == FORMATION_ID_ENV)
1561 .map(|(_, value)| value.clone())
1562 .collect()
1563 };
1564
1565 let outside = child_environment(&request(&[FORMATION_ID_ENV], None), None);
1566 assert!(injected(&outside).is_empty(), "{outside:?}");
1567
1568 let other = FormationId::mint();
1569 let handle = SpawnerHandle::for_delegation(&spawner(), "dlg_0001".to_string());
1570 for (spawner, handle) in [(Some(spawner()), Some(&handle)), (None, None)] {
1571 let mut member = request(&[FORMATION_ID_ENV], spawner);
1572 member.formation_id = Some(other.clone());
1573 let env = child_environment(&member, handle);
1574 assert_eq!(injected(&env), vec![other.as_str().to_string()], "{env:?}");
1575 assert_eq!(
1576 env.last(),
1577 Some(&(FORMATION_ID_ENV.to_string(), other.as_str().to_string())),
1578 "the formation id is the last runtime-owned name: {env:?}"
1579 );
1580 }
1581 std::env::remove_var(FORMATION_ID_ENV);
1582 }
1583
1584 #[test]
1587 fn every_child_is_told_host_warnings_were_reported() {
1588 std::env::set_var(HOST_WARNINGS_REPORTED_ENV, "0");
1589 let handle = SpawnerHandle::for_delegation(&spawner(), "dlg_0001".to_string());
1590 for (spawner, handle) in [(Some(spawner()), Some(&handle)), (None, None)] {
1591 for allow in [&[][..], &[HOST_WARNINGS_REPORTED_ENV][..]] {
1592 let env = child_environment(&request(allow, spawner.clone()), handle);
1593 let values: Vec<&str> = env
1594 .iter()
1595 .filter(|(key, _)| key == HOST_WARNINGS_REPORTED_ENV)
1596 .map(|(_, value)| value.as_str())
1597 .collect();
1598 assert_eq!(values, vec!["1"], "{env:?}");
1599 }
1600 }
1601 std::env::remove_var(HOST_WARNINGS_REPORTED_ENV);
1602 }
1603
1604 fn values_of<'a>(env: &'a [(String, String)], name: &str) -> Vec<&'a str> {
1605 env.iter()
1606 .filter(|(key, _)| key == name)
1607 .map(|(_, value)| value.as_str())
1608 .collect()
1609 }
1610
1611 #[test]
1614 fn a_child_carries_the_parents_account_names() {
1615 let _guard = crate::shell::ACCOUNT_ENV_LOCK
1616 .lock()
1617 .unwrap_or_else(|poisoned| poisoned.into_inner());
1618 std::env::set_var("USER", "murmur-parent-user");
1619 std::env::set_var("LOGNAME", "murmur-parent-login");
1620 for allow in [&[][..], &["USER"][..], &["USER", "LOGNAME"][..]] {
1621 let env = child_environment(&request(allow, None), None);
1622 assert_eq!(
1623 values_of(&env, "USER"),
1624 vec!["murmur-parent-user"],
1625 "{env:?}"
1626 );
1627 assert_eq!(
1628 values_of(&env, "LOGNAME"),
1629 vec!["murmur-parent-login"],
1630 "{env:?}"
1631 );
1632 }
1633 }
1634
1635 #[test]
1637 fn a_child_of_a_parent_without_account_names_has_none() {
1638 let _guard = crate::shell::ACCOUNT_ENV_LOCK
1639 .lock()
1640 .unwrap_or_else(|poisoned| poisoned.into_inner());
1641 let saved: Vec<_> = crate::shell::ACCOUNT_ENV
1642 .iter()
1643 .map(|name| (*name, std::env::var_os(name)))
1644 .collect();
1645 for name in crate::shell::ACCOUNT_ENV {
1646 std::env::remove_var(name);
1647 }
1648 let env = child_environment(&request(&["USER", "LOGNAME"], None), None);
1649 for (name, value) in saved {
1650 if let Some(value) = value {
1651 std::env::set_var(name, value);
1652 }
1653 }
1654 assert!(values_of(&env, "USER").is_empty(), "{env:?}");
1655 assert!(values_of(&env, "LOGNAME").is_empty(), "{env:?}");
1656 }
1657
1658 #[test]
1662 fn a_tail_read_after_the_end_waits_for_the_drain_to_catch_up() {
1663 let tail = Arc::new(StderrTail {
1664 state: Mutex::new(StderrState::default()),
1665 drained: Condvar::new(),
1666 });
1667
1668 let writer = Arc::clone(&tail);
1669 std::thread::spawn(move || {
1670 std::thread::sleep(Duration::from_millis(200));
1671 lock(&writer.state)
1672 .lines
1673 .push("error[E-CAP-003]: not achievable on this host".to_string());
1674 writer.finish();
1675 });
1676
1677 assert!(tail.lines().is_empty());
1678 assert_eq!(
1679 tail.lines_at_end(),
1680 vec!["error[E-CAP-003]: not achievable on this host".to_string()]
1681 );
1682 }
1683
1684 #[test]
1687 fn a_tail_with_no_pipe_behind_it_is_at_its_end_already() {
1688 let tail = StderrTail {
1689 state: Mutex::new(StderrState::default()),
1690 drained: Condvar::new(),
1691 };
1692 tail.finish();
1693
1694 let started = Instant::now();
1695 assert!(tail.lines_at_end().is_empty());
1696 assert!(
1697 started.elapsed() < CHILD_STDERR_DRAIN_TIMEOUT,
1698 "{started:?}"
1699 );
1700 }
1701
1702 #[test]
1703 fn child_launch_reads_the_operator_token_from_the_readiness_line() {
1704 let report = serde_json::json!({
1705 "url": "localhost:1",
1706 "session_id": "ses_x",
1707 "tokens": {"operator": "mdt1.a.b", "watcher": "mdt1.c.d"},
1708 });
1709 assert_eq!(
1710 readiness_door_token(&report).map(|token| token.expose().to_string()),
1711 Some("mdt1.a.b".to_string())
1712 );
1713 let public = serde_json::json!({"url": "localhost:1", "session_id": "ses_x"});
1714 assert_eq!(readiness_door_token(&public), None);
1715 }
1716
1717 #[test]
1718 fn child_launch_debug_prints_no_door_token() {
1719 let child = LaunchedChild {
1720 workdir: PathBuf::from("/tmp/child"),
1721 workdir_rel: PathBuf::new(),
1722 parent_root: WorkdirRoot::open(&std::env::temp_dir()).unwrap(),
1723 session_id: "ses_child".to_string(),
1724 records_dir: PathBuf::new(),
1725 records_root: None,
1726 capsule_url: "http://localhost:1".to_string(),
1727 argv: Vec::new(),
1728 env: Vec::new(),
1729 delegation_id: None,
1730 formation_id: None,
1731 door_token: Some(crate::door_auth::DoorToken::new(
1732 "mdt1.secretpayload.secretmac".to_string(),
1733 )),
1734 process: Arc::new(Mutex::new(ChildProcess {
1735 child: None,
1736 deliberate: false,
1737 status: None,
1738 terminated_at: None,
1739 #[cfg(unix)]
1740 lifeline: None,
1741 })),
1742 stderr_tail: Arc::new(StderrTail {
1743 state: Mutex::new(StderrState::default()),
1744 drained: Condvar::new(),
1745 }),
1746 started: Instant::now(),
1747 watcher: None,
1748 };
1749 let debug = format!("{child:?}");
1750 assert!(!debug.contains("secretpayload"), "{debug}");
1751 assert!(debug.contains("door_token"), "{debug}");
1752 }
1753
1754 fn start_process_code() -> String {
1756 let source = include_str!("child_launch.rs");
1757 let start = source
1758 .find("pub(crate) fn start_process(")
1759 .expect("start_process is defined here");
1760 let end = start
1761 + source[start..]
1762 .find("\n}\n")
1763 .expect("start_process ends at a closing brace");
1764 source[start..end]
1765 .lines()
1766 .filter(|line| !line.trim_start().starts_with("//"))
1767 .collect::<Vec<_>>()
1768 .join("\n")
1769 }
1770
1771 #[test]
1776 fn start_process_registers_no_fd_hygiene() {
1777 let code = start_process_code();
1778 for name in [
1779 "prepare_enforcement",
1780 "mark_inherited_fds_cloexec",
1781 "close_range",
1782 ] {
1783 assert!(!code.contains(name), "start_process names {name}:\n{code}");
1784 }
1785 let handed = code
1786 .rfind("lifeline.hand_to")
1787 .expect("the lifeline is handed");
1788 let kept = code
1789 .find("keep_across_exec")
1790 .expect("an inherited descriptor is kept");
1791 assert!(kept < handed, "the lifeline is registered last:\n{code}");
1792 assert!(
1793 !code[handed..].contains("pre_exec"),
1794 "nothing is registered after the lifeline:\n{code}"
1795 );
1796 }
1797
1798 #[cfg(unix)]
1799 fn process_with_lifeline(program: &str, args: &[&str]) -> (ChildProcess, i32) {
1800 let lifeline = crate::lifeline::ChildLifeline::new().unwrap();
1801 let write = lifeline.write_fd().unwrap();
1802 let child = Command::new(program).args(args).spawn().unwrap();
1803 (
1804 ChildProcess {
1805 child: Some(child),
1806 deliberate: false,
1807 status: None,
1808 terminated_at: None,
1809 lifeline: Some(lifeline),
1810 },
1811 write,
1812 )
1813 }
1814
1815 #[cfg(unix)]
1817 #[test]
1818 fn ending_a_child_closes_its_lifeline() {
1819 let (process, _) = process_with_lifeline("sleep", &["30"]);
1820 let process = Mutex::new(process);
1821 end_process(&process).unwrap();
1822 assert!(lock(&process).lifeline.is_none());
1823 assert!(lock(&process).status.is_some());
1824 end_process(&process).unwrap();
1825 }
1826
1827 #[cfg(unix)]
1830 #[test]
1831 fn ending_a_child_sends_sigterm_and_waits_for_it_to_wind_down() {
1832 let dir = tempfile::tempdir().unwrap();
1833 let marker = dir.path().join("wound-down");
1834 let script = format!(
1835 "trap 'echo wound-down > {}; exit 7' TERM; sleep 30 & wait",
1836 marker.display()
1837 );
1838 let (process, _) = process_with_lifeline("bash", &["-c", &script]);
1839 let process = Mutex::new(process);
1840 std::thread::sleep(Duration::from_millis(300));
1842 let began = Instant::now();
1843 end_process(&process).unwrap();
1844 assert!(
1845 began.elapsed() < CHILD_END_GRACE,
1846 "took {:?}",
1847 began.elapsed()
1848 );
1849 assert_eq!(
1850 std::fs::read_to_string(&marker).unwrap().trim(),
1851 "wound-down"
1852 );
1853 assert_eq!(
1854 lock(&process).status.and_then(|status| status.code()),
1855 Some(7)
1856 );
1857 assert!(lock(&process).lifeline.is_none());
1858 }
1859
1860 #[cfg(unix)]
1862 #[test]
1863 fn a_child_that_ignores_sigterm_is_killed_at_the_grace() {
1864 use std::os::unix::process::ExitStatusExt as _;
1865 let (process, _) =
1866 process_with_lifeline("bash", &["-c", "trap '' TERM; while :; do sleep 0.1; done"]);
1867 let process = Mutex::new(process);
1868 std::thread::sleep(Duration::from_millis(300));
1869 let began = Instant::now();
1870 end_process(&process).unwrap();
1871 let took = began.elapsed();
1872 assert!(took >= CHILD_END_GRACE, "killed before the grace: {took:?}");
1873 assert!(
1874 took < CHILD_END_GRACE + Duration::from_secs(5),
1875 "took {took:?}"
1876 );
1877 assert_eq!(
1878 lock(&process).status.and_then(|status| status.signal()),
1879 Some(libc::SIGKILL)
1880 );
1881 }
1882
1883 #[cfg(unix)]
1886 #[test]
1887 fn children_signalled_together_wait_one_grace_between_them() {
1888 let wedged = || {
1889 let (process, _) =
1890 process_with_lifeline("bash", &["-c", "trap '' TERM; while :; do sleep 0.1; done"]);
1891 LaunchedChild {
1892 workdir: PathBuf::from("/tmp/child"),
1893 workdir_rel: PathBuf::new(),
1894 parent_root: WorkdirRoot::open(&std::env::temp_dir()).unwrap(),
1895 session_id: "ses_child".to_string(),
1896 records_dir: PathBuf::new(),
1897 records_root: None,
1898 capsule_url: String::new(),
1899 argv: Vec::new(),
1900 env: Vec::new(),
1901 delegation_id: None,
1902 formation_id: None,
1903 door_token: None,
1904 process: Arc::new(Mutex::new(process)),
1905 stderr_tail: Arc::new(StderrTail {
1906 state: Mutex::new(StderrState::default()),
1907 drained: Condvar::new(),
1908 }),
1909 started: Instant::now(),
1910 watcher: None,
1911 }
1912 };
1913 let children = vec![wedged(), wedged()];
1914 std::thread::sleep(Duration::from_millis(300));
1915 let began = Instant::now();
1916 for child in &children {
1917 child.begin_ending();
1918 }
1919 drop(children);
1920 let took = began.elapsed();
1921 assert!(took >= CHILD_END_GRACE, "took {took:?}");
1922 assert!(
1923 took < CHILD_END_GRACE * 2,
1924 "two children waited a grace each: {took:?}"
1925 );
1926 }
1927
1928 #[cfg(unix)]
1930 #[test]
1931 fn a_child_seen_exited_has_its_lifeline_closed() {
1932 let (mut process, _) = process_with_lifeline("true", &[]);
1933 let deadline = Instant::now() + Duration::from_secs(10);
1934 let ending = loop {
1935 if let Some(ending) = process.poll() {
1936 break ending;
1937 }
1938 assert!(Instant::now() < deadline, "`true` never exited");
1939 std::thread::sleep(Duration::from_millis(20));
1940 };
1941 assert!(matches!(ending, Ending::Exited(Some(_))));
1942 assert!(process.lifeline.is_none());
1943 }
1944
1945 #[cfg(unix)]
1950 #[test]
1951 #[allow(unsafe_code)]
1952 fn a_childs_lifeline_closes_when_its_handle_drops_or_its_end_is_seen() {
1953 let launched = |program: &str, args: &[&str]| {
1954 let (process, write) = process_with_lifeline(program, args);
1955 let process = Arc::new(Mutex::new(process));
1956 let child = LaunchedChild {
1957 workdir: PathBuf::from("/tmp/child"),
1958 workdir_rel: PathBuf::new(),
1959 parent_root: WorkdirRoot::open(&std::env::temp_dir()).unwrap(),
1960 session_id: "ses_child".to_string(),
1961 records_dir: PathBuf::new(),
1962 records_root: None,
1963 capsule_url: String::new(),
1964 argv: Vec::new(),
1965 env: Vec::new(),
1966 delegation_id: None,
1967 formation_id: None,
1968 door_token: None,
1969 process: Arc::clone(&process),
1970 stderr_tail: Arc::new(StderrTail {
1971 state: Mutex::new(StderrState::default()),
1972 drained: Condvar::new(),
1973 }),
1974 started: Instant::now(),
1975 watcher: None,
1976 };
1977 (child, process, write)
1978 };
1979 let is_open = |fd: i32| {
1980 unsafe { libc::fcntl(fd, libc::F_GETFD) >= 0 }
1982 };
1983
1984 let (child, process, write) = launched("sleep", &["30"]);
1986 assert!(!child.has_exited());
1987 assert!(is_open(write), "a held child's write end is open");
1988 drop(child);
1989 assert!(
1990 lock(&process).lifeline.is_none(),
1991 "dropping the handle closes it"
1992 );
1993 assert!(
1994 lock(&process).deliberate,
1995 "a running child's drop is deliberate"
1996 );
1997 assert!(lock(&process).status.is_some(), "and reaps the child");
1999
2000 let (child, process, _) = launched("true", &[]);
2002 let deadline = Instant::now() + Duration::from_secs(10);
2003 while !child.has_exited() {
2004 assert!(Instant::now() < deadline, "`true` never exited");
2005 std::thread::sleep(Duration::from_millis(20));
2006 }
2007 assert!(
2008 lock(&process).lifeline.is_none(),
2009 "seeing the end closes it"
2010 );
2011 drop(child);
2012 assert!(
2013 !lock(&process).deliberate,
2014 "a child that ended on its own is not recorded as ended by its parent"
2015 );
2016 }
2017
2018 #[cfg(unix)]
2021 #[test]
2022 fn a_child_its_parent_ends_is_recorded_in_the_shared_words() {
2023 let dir = tempfile::tempdir().unwrap();
2024 let (process, _) = process_with_lifeline("sleep", &["30"]);
2025 let mut child = LaunchedChild {
2026 workdir: dir.path().to_path_buf(),
2027 workdir_rel: PathBuf::new(),
2028 parent_root: WorkdirRoot::open(dir.path()).unwrap(),
2029 session_id: "ses_child".to_string(),
2030 records_dir: dir.path().to_path_buf(),
2031 records_root: Some(WorkdirRoot::open(dir.path()).unwrap()),
2032 capsule_url: String::new(),
2033 argv: Vec::new(),
2034 env: Vec::new(),
2035 delegation_id: Some("dlg_watched".to_string()),
2036 formation_id: None,
2037 door_token: None,
2038 process: Arc::new(Mutex::new(process)),
2039 stderr_tail: Arc::new(StderrTail {
2040 state: Mutex::new(StderrState::default()),
2041 drained: Condvar::new(),
2042 }),
2043 started: Instant::now(),
2044 watcher: None,
2045 };
2046 let watcher = watch_for_completion(
2047 &child,
2048 SpawnerHandle::for_delegation(&spawner(), "dlg_watched".to_string()),
2049 CompletionAddress {
2050 url: "http://127.0.0.1:1".to_string(),
2051 },
2052 "worker",
2053 "0.1.0",
2054 None,
2055 )
2056 .expect("the watcher starts");
2057
2058 child.shutdown().unwrap();
2059 watcher.join().unwrap();
2060
2061 let recorded = delegation::read_completion(&WorkdirRoot::open(dir.path()).unwrap())
2062 .expect("the ending is recorded");
2063 assert_eq!(recorded.status, DelegationStatus::Terminated);
2064 assert_eq!(recorded.reported_by, Reporter::Launcher);
2065 assert!(!recorded.delivered);
2066 assert_eq!(
2067 recorded.detail.as_deref(),
2068 Some(delegation::PARENT_ENDED_DETAIL)
2069 );
2070 }
2071
2072 #[cfg(unix)]
2076 #[test]
2077 fn a_completion_the_child_records_while_being_ended_is_overwritten_by_the_launcher() {
2078 let dir = tempfile::tempdir().unwrap();
2079 let own = DelegationOutcome {
2080 delegation_id: "dlg_watched".to_string(),
2081 capsule_name: "worker".to_string(),
2082 capsule_version: "0.1.0".to_string(),
2083 session_id: "ses_child".to_string(),
2084 status: DelegationStatus::Error,
2085 result_path: None,
2086 workdir: dir.path().display().to_string(),
2087 duration_ms: 1,
2088 detail: None,
2089 reported_by: Reporter::Child,
2090 delivered: false,
2091 delivery_error: None,
2092 };
2093 let staged = dir.path().join("own-record.json");
2094 std::fs::write(&staged, serde_json::to_vec(&own).unwrap()).unwrap();
2095 let script = format!(
2096 "trap 'cp {} {}; exit 0' TERM; sleep 30 & wait",
2097 staged.display(),
2098 dir.path().join(delegation::COMPLETION_FILE).display()
2099 );
2100 let (process, _) = process_with_lifeline("bash", &["-c", &script]);
2101 let mut child = LaunchedChild {
2102 workdir: dir.path().to_path_buf(),
2103 workdir_rel: PathBuf::new(),
2104 parent_root: WorkdirRoot::open(dir.path()).unwrap(),
2105 session_id: "ses_child".to_string(),
2106 records_dir: dir.path().to_path_buf(),
2107 records_root: Some(WorkdirRoot::open(dir.path()).unwrap()),
2108 capsule_url: String::new(),
2109 argv: Vec::new(),
2110 env: Vec::new(),
2111 delegation_id: Some("dlg_watched".to_string()),
2112 formation_id: None,
2113 door_token: None,
2114 process: Arc::new(Mutex::new(process)),
2115 stderr_tail: Arc::new(StderrTail {
2116 state: Mutex::new(StderrState::default()),
2117 drained: Condvar::new(),
2118 }),
2119 started: Instant::now(),
2120 watcher: None,
2121 };
2122 let watcher = watch_for_completion(
2123 &child,
2124 SpawnerHandle::for_delegation(&spawner(), "dlg_watched".to_string()),
2125 CompletionAddress {
2126 url: "http://127.0.0.1:1".to_string(),
2127 },
2128 "worker",
2129 "0.1.0",
2130 None,
2131 )
2132 .expect("the watcher starts");
2133 std::thread::sleep(Duration::from_millis(300));
2134
2135 child.shutdown().unwrap();
2136 watcher.join().unwrap();
2137 let recorded = delegation::read_completion(&WorkdirRoot::open(dir.path()).unwrap())
2138 .expect("the ending is recorded");
2139 assert_eq!(recorded.status, DelegationStatus::Terminated);
2140 assert_eq!(recorded.reported_by, Reporter::Launcher);
2141 assert!(!recorded.delivered);
2142 assert_eq!(recorded.delivery_error, None);
2143 }
2144}