1use std::io::Write;
58use std::path::{Path, PathBuf};
59use std::process::{Child, ExitStatus};
60use std::sync::atomic::{AtomicBool, AtomicI32, AtomicI64, Ordering};
61use std::sync::{mpsc, Once};
62use std::time::{Duration, Instant};
63
64use crate::child_launch::{self, AfterReadiness, ProcessEnv, ProcessLaunch, StartedProcess};
65use crate::delegation::SPAWNER_ENV;
66use crate::door_auth::DoorToken;
67use crate::formation::{FormationId, FormationPeer, FORMATION_PEERS_ENV};
68use crate::formation_credentials::{
69 render_address_line, FormationAuthority, FORMATION_CHANNEL_ENV,
70};
71use crate::lifeline::SPAWNER_LIFELINE_ENV;
72use crate::roster::AdmittedRoster;
73use crate::running::ProcessIdentity;
74use serde::{Deserialize, Serialize};
75
76pub const MEMBER_READY_TIMEOUT: Duration = Duration::from_secs(180);
82
83pub const MEMBER_STOP_GRACE: Duration = Duration::from_secs(25);
89
90const READY_PROBE_INTERVAL: Duration = Duration::from_millis(100);
92
93const POLL_INTERVAL: Duration = Duration::from_millis(50);
96
97const EXIT_AFTER_EOF_WAIT: Duration = Duration::from_secs(2);
99
100pub const PEER_BIND_ADDR: &str = "127.0.0.1";
102
103pub const CAPSULE_SHA256_FLAG: &str = "--capsule-sha256";
105
106pub const STORE_ROOT_FLAG: &str = "--store-root";
109
110pub const FORMATIONS_DIR: &str = "formations";
112
113pub const FORMATION_MARKER_FILE: &str = "formation.json";
116
117#[derive(Debug, Clone)]
119pub struct FormationLaunchOptions {
120 pub formation_id: FormationId,
122 pub project_dir: PathBuf,
125 pub task: Option<String>,
127 pub json: bool,
129 pub verbose: bool,
131 pub no_env_file: bool,
133 pub containment: Option<String>,
135 pub ready_timeout: Duration,
138 pub stop_grace: Duration,
141 pub(crate) formations_dir: Option<PathBuf>,
144}
145
146impl FormationLaunchOptions {
147 pub fn new(formation_id: FormationId, project_dir: PathBuf) -> Self {
149 Self {
150 formation_id,
151 project_dir,
152 task: None,
153 json: false,
154 verbose: false,
155 no_env_file: false,
156 containment: None,
157 ready_timeout: MEMBER_READY_TIMEOUT,
158 stop_grace: MEMBER_STOP_GRACE,
159 formations_dir: None,
160 }
161 }
162}
163
164#[derive(Debug, Clone, PartialEq, Eq)]
166pub(crate) struct PlannedMember {
167 pub(crate) name: String,
168 pub(crate) capsule: String,
169 pub(crate) version: String,
170 pub(crate) sha256: String,
172}
173
174#[derive(Debug, Clone)]
177pub(crate) struct FormationPlan {
178 pub(crate) peers: Vec<PlannedMember>,
179 pub(crate) entry: PlannedMember,
180 pub(crate) callees: Vec<(String, Vec<String>)>,
183}
184
185impl FormationPlan {
186 pub(crate) fn from_roster(roster: &AdmittedRoster) -> Self {
187 let planned = |member: &crate::roster::AdmittedMember| PlannedMember {
188 name: member.name.clone(),
189 capsule: member.capsule.clone(),
190 version: member.version.clone(),
191 sha256: member.sha256.clone(),
192 };
193 let entry = roster.entry();
194 Self {
195 peers: roster
196 .members()
197 .iter()
198 .filter(|member| !member.entry)
199 .map(planned)
200 .collect(),
201 entry: planned(entry),
202 callees: roster
203 .members()
204 .iter()
205 .map(|member| {
206 let callees: Vec<String> = roster
207 .callees(&member.name)
208 .map(|callee| callee.name.clone())
209 .collect();
210 (member.name.clone(), callees)
211 })
212 .filter(|(_, callees)| !callees.is_empty())
213 .collect(),
214 }
215 }
216
217 pub(crate) fn callees_of(&self, member: &str) -> &[String] {
219 self.callees
220 .iter()
221 .find(|(name, _)| name == member)
222 .map(|(_, callees)| callees.as_slice())
223 .unwrap_or_default()
224 }
225
226 fn first_line(
229 &self,
230 authority: &FormationAuthority,
231 launcher: Option<&ProcessIdentity>,
232 member: &str,
233 ) -> String {
234 let callees: Vec<&str> = self.callees_of(member).iter().map(String::as_str).collect();
235 let mut bundle = authority.member_bundle(member, &callees);
236 bundle.launcher = launcher.cloned();
237 bundle.render_line()
238 }
239}
240
241pub struct ReadyPeer {
243 pub name: String,
244 pub capsule: String,
245 pub version: String,
246 pub pid: u32,
247 pub session_id: String,
248 pub url: String,
250 pub workdir: PathBuf,
253 door_token: Option<DoorToken>,
256}
257
258impl ReadyPeer {
259 pub fn door_token(&self) -> Option<&DoorToken> {
261 self.door_token.as_ref()
262 }
263}
264
265impl std::fmt::Debug for ReadyPeer {
266 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
267 f.debug_struct("ReadyPeer")
268 .field("name", &self.name)
269 .field("capsule", &self.capsule)
270 .field("version", &self.version)
271 .field("pid", &self.pid)
272 .field("session_id", &self.session_id)
273 .field("url", &self.url)
274 .field("workdir", &self.workdir)
275 .field(
276 "door_token",
277 &self.door_token.as_ref().map(|_| "<redacted>"),
278 )
279 .finish()
280 }
281}
282
283#[derive(Debug, Clone, PartialEq, Eq)]
285pub enum MemberFailureReason {
286 NotStarted { error: String },
288 ExitedBeforeReporting {
290 status: String,
291 stderr_tail: Vec<String>,
292 },
293 ExitedBeforeDoorAnswered {
295 status: String,
296 stderr_tail: Vec<String>,
297 },
298 DidNotReport { deadline: Duration },
300 UnreadableReport { why: String },
302 ReportedNoDoor,
304 WrongFormation {
306 reported: Option<String>,
307 expected: String,
308 },
309 DoorDidNotAnswer {
311 url: String,
312 session_id: String,
313 deadline: Duration,
314 last_error: String,
315 },
316}
317
318impl std::fmt::Display for MemberFailureReason {
319 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
320 let tail = |f: &mut std::fmt::Formatter<'_>, lines: &[String]| -> std::fmt::Result {
321 if lines.is_empty() {
322 return f.write_str("; it wrote nothing to stderr");
323 }
324 f.write_str("; its last stderr lines were:")?;
325 for line in lines {
326 write!(f, "\n {line}")?;
327 }
328 Ok(())
329 };
330 match self {
331 Self::NotStarted { error } => write!(f, "its process could not be started: {error}"),
332 Self::ExitedBeforeReporting {
333 status,
334 stderr_tail,
335 } => {
336 write!(f, "its process exited with {status} before reporting")?;
337 tail(f, stderr_tail)
338 }
339 Self::ExitedBeforeDoorAnswered {
340 status,
341 stderr_tail,
342 } => {
343 write!(
344 f,
345 "its process reported and then exited with {status} before its door answered"
346 )?;
347 tail(f, stderr_tail)
348 }
349 Self::DidNotReport { deadline } => {
350 write!(f, "it did not report within {}s", deadline.as_secs_f32())
351 }
352 Self::UnreadableReport { why } => {
353 write!(f, "its first line is not a readiness line: {why}")
354 }
355 Self::ReportedNoDoor => f.write_str("it reported no door"),
356 Self::WrongFormation { reported, expected } => match reported {
357 Some(reported) => {
358 write!(f, "it reported formation {reported} instead of {expected}")
359 }
360 None => write!(f, "it reported no formation instead of {expected}"),
361 },
362 Self::DoorDidNotAnswer {
363 url,
364 session_id,
365 deadline,
366 last_error,
367 } => write!(
368 f,
369 "its door at {url} did not answer as session {session_id} within {}s; the last \
370 probe said: {last_error}",
371 deadline.as_secs_f32()
372 ),
373 }
374 }
375}
376
377#[derive(Debug, Clone, PartialEq, Eq)]
379pub struct MemberFailure {
380 pub name: String,
381 pub capsule: String,
382 pub version: String,
383 pub reason: MemberFailureReason,
384}
385
386#[derive(Debug, Clone, PartialEq, Eq)]
388pub struct StoppedMember {
389 pub name: String,
390 pub pid: u32,
391}
392
393#[derive(Debug, Clone, PartialEq, Eq)]
395pub struct MemberLaunchFailure {
396 pub failed: Vec<MemberFailure>,
399 pub stopped: Vec<StoppedMember>,
401 pub interrupted_by: Option<i32>,
404}
405
406impl std::fmt::Display for MemberLaunchFailure {
407 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
408 match self.interrupted_by {
409 Some(signal) if self.failed.is_empty() => write!(
410 f,
411 "the formation launch was interrupted by {} before every member was ready",
412 signal_name(signal)
413 )?,
414 _ => f.write_str("the formation could not be launched")?,
415 }
416 for failure in &self.failed {
417 write!(
418 f,
419 "\n '{}' ({}@{}): {}",
420 failure.name, failure.capsule, failure.version, failure.reason
421 )?;
422 }
423 write!(f, "\n stopped: {}", render_stopped(&self.stopped))
424 }
425}
426
427pub fn render_stopped(stopped: &[StoppedMember]) -> String {
429 if stopped.is_empty() {
430 return "none".to_string();
431 }
432 stopped
433 .iter()
434 .map(|member| format!("{} (pid {})", member.name, member.pid))
435 .collect::<Vec<_>>()
436 .join(", ")
437}
438
439pub fn signal_name(signal: i32) -> String {
441 match signal {
442 libc::SIGINT => "SIGINT".to_string(),
443 libc::SIGTERM => "SIGTERM".to_string(),
444 libc::SIGHUP => "SIGHUP".to_string(),
445 other => format!("signal {other}"),
446 }
447}
448
449#[derive(Debug, Clone, PartialEq, Eq)]
451pub enum FormationEnding {
452 EntryExited(ExitStatus),
454 Signalled(i32),
456}
457
458#[derive(Debug, Clone)]
460pub struct FormationExit {
461 pub ending: FormationEnding,
462 pub stopped: Vec<StoppedMember>,
464}
465
466impl FormationExit {
467 pub fn exit_code(&self) -> i32 {
470 match &self.ending {
471 FormationEnding::EntryExited(status) => crate::shell::exit_code_of(status),
472 FormationEnding::Signalled(signal) => 128 + signal,
473 }
474 }
475}
476
477fn describe_status(status: ExitStatus) -> String {
479 if let Some(code) = status.code() {
480 return format!("status {code}");
481 }
482 #[cfg(unix)]
483 {
484 use std::os::unix::process::ExitStatusExt;
485 if let Some(signal) = status.signal() {
486 return format!("signal {signal}");
487 }
488 }
489 status.to_string()
490}
491
492static CAUGHT_SIGNAL: AtomicI32 = AtomicI32::new(0);
496
497#[cfg(unix)]
498extern "C" fn record_signal(signal: libc::c_int) {
499 CAUGHT_SIGNAL.store(signal, Ordering::SeqCst);
501}
502
503#[cfg(unix)]
509#[allow(unsafe_code)]
510pub fn catch_launcher_signals() -> std::io::Result<()> {
511 for signal in [libc::SIGINT, libc::SIGTERM, libc::SIGHUP] {
512 let installed = unsafe {
516 let mut action: libc::sigaction = std::mem::zeroed();
517 action.sa_sigaction = record_signal as extern "C" fn(libc::c_int) as libc::sighandler_t;
518 action.sa_flags = libc::SA_RESTART;
519 libc::sigemptyset(&mut action.sa_mask);
520 libc::sigaction(signal, &action, std::ptr::null_mut())
521 };
522 if installed != 0 {
523 return Err(std::io::Error::last_os_error());
524 }
525 }
526 Ok(())
527}
528
529#[cfg(not(unix))]
530pub fn catch_launcher_signals() -> std::io::Result<()> {
531 Ok(())
532}
533
534pub fn caught_signal() -> Option<i32> {
536 match CAUGHT_SIGNAL.load(Ordering::SeqCst) {
537 0 => None,
538 signal => Some(signal),
539 }
540}
541
542const REGISTRY_SLOTS: usize = 1024;
547
548static MEMBER_REGISTRY: [AtomicI64; REGISTRY_SLOTS] = [const { AtomicI64::new(0) }; REGISTRY_SLOTS];
552
553struct RegistrySlot(Option<usize>);
555
556impl RegistrySlot {
557 fn register(pid: u32, own_group: bool) -> Self {
558 let pid = i64::from(pid);
559 let value = if own_group { pid } else { -pid };
560 let slot = MEMBER_REGISTRY.iter().position(|slot| {
561 slot.compare_exchange(0, value, Ordering::SeqCst, Ordering::SeqCst)
562 .is_ok()
563 });
564 Self(slot)
565 }
566
567 fn release(&mut self) {
568 if let Some(index) = self.0.take() {
569 MEMBER_REGISTRY[index].store(0, Ordering::SeqCst);
570 }
571 }
572}
573
574impl Drop for RegistrySlot {
575 fn drop(&mut self) {
576 self.release();
577 }
578}
579
580pub(crate) fn kill_registered_members() {
583 for slot in &MEMBER_REGISTRY {
584 match slot.load(Ordering::SeqCst) {
585 0 => {}
586 pid if pid > 0 => signal_group(pid, libc::SIGKILL),
587 pid => signal_pid(-pid, libc::SIGKILL),
588 }
589 }
590}
591
592fn install_panic_hook() {
596 static INSTALLED: Once = Once::new();
597 INSTALLED.call_once(|| {
598 let previous = std::panic::take_hook();
599 std::panic::set_hook(Box::new(move |info| {
600 previous(info);
601 if cfg!(panic = "abort") {
602 kill_registered_members();
603 }
604 }));
605 });
606}
607
608#[cfg(unix)]
609#[allow(unsafe_code)]
610fn signal_pid(pid: i64, signal: libc::c_int) {
611 let Ok(pid) = libc::pid_t::try_from(pid) else {
612 return;
613 };
614 if pid <= 0 {
615 return;
616 }
617 unsafe {
620 libc::kill(pid, signal);
621 }
622}
623
624#[cfg(unix)]
625#[allow(unsafe_code)]
626fn signal_group(pgid: i64, signal: libc::c_int) {
627 let Ok(pgid) = libc::pid_t::try_from(pgid) else {
628 return;
629 };
630 if pgid <= 1 {
631 return;
632 }
633 unsafe {
636 libc::killpg(pgid, signal);
637 }
638}
639
640#[cfg(not(unix))]
641fn signal_pid(_pid: i64, _signal: i32) {}
642
643#[cfg(not(unix))]
644fn signal_group(_pgid: i64, _signal: i32) {}
645
646struct MemberProcess {
650 name: String,
651 pid: u32,
652 child: Child,
653 own_group: bool,
654 #[cfg(unix)]
657 lifeline: Option<crate::lifeline::MemberLifeline>,
658 status: Option<ExitStatus>,
661 slot: RegistrySlot,
662 grace: Duration,
663 channel: Option<std::io::PipeWriter>,
666}
667
668impl MemberProcess {
669 fn new(
670 name: &str,
671 child: Child,
672 own_group: bool,
673 grace: Duration,
674 channel: std::io::PipeWriter,
675 #[cfg(unix)] lifeline: Option<crate::lifeline::MemberLifeline>,
676 ) -> Self {
677 let pid = child.id();
678 Self {
679 name: name.to_string(),
680 pid,
681 child,
682 own_group,
683 #[cfg(unix)]
684 lifeline,
685 status: None,
686 slot: RegistrySlot::register(pid, own_group),
687 grace,
688 channel: Some(channel),
689 }
690 }
691
692 fn send_line(&mut self, line: &str) {
695 if let Some(channel) = self.channel.as_mut() {
696 let _ = channel.write_all(format!("{line}\n").as_bytes());
697 }
698 }
699
700 fn has_exited(&self) -> bool {
702 self.status.is_some() || !crate::running::pid_is_alive(self.pid)
703 }
704
705 fn exit_status_now(&mut self) -> Option<ExitStatus> {
710 if self.status.is_none() && self.has_exited() {
711 self.reap();
712 }
713 self.status
714 }
715
716 fn reap(&mut self) {
717 if self.status.is_some() {
718 return;
719 }
720 if self.own_group {
721 signal_group(i64::from(self.pid), libc::SIGKILL);
722 } else if !self.has_exited() {
723 signal_pid(i64::from(self.pid), libc::SIGKILL);
724 }
725 self.status = self.child.wait().ok();
726 if self.status.is_none() {
727 self.status = Some(ExitStatus::default());
729 }
730 self.slot.release();
731 self.channel = None;
732 }
733
734 fn close_lifeline(&mut self) {
736 #[cfg(unix)]
737 if let Some(lifeline) = &mut self.lifeline {
738 lifeline.close();
739 }
740 }
741
742 fn is_running(&self) -> bool {
744 self.status.is_none() && !self.has_exited()
745 }
746
747 fn stop(&mut self) -> Option<StoppedMember> {
750 let was_running = self.is_running();
751 self.stop_from(was_running)
752 }
753
754 fn stop_from(&mut self, was_running: bool) -> Option<StoppedMember> {
757 if self.status.is_some() {
758 return None;
759 }
760 self.close_lifeline();
761 if was_running {
762 let deadline = Instant::now() + self.grace;
763 while !self.has_exited() && Instant::now() < deadline {
764 std::thread::sleep(POLL_INTERVAL);
765 }
766 }
767 self.reap();
768 was_running.then(|| StoppedMember {
769 name: self.name.clone(),
770 pid: self.pid,
771 })
772 }
773}
774
775impl Drop for MemberProcess {
776 fn drop(&mut self) {
777 let _ = self.stop();
778 }
779}
780
781struct MemberSet(Vec<MemberProcess>);
783
784impl MemberSet {
785 fn stop_all(&mut self) -> Vec<StoppedMember> {
788 let running: Vec<bool> = self.0.iter().map(MemberProcess::is_running).collect();
791 for member in &mut self.0 {
792 member.close_lifeline();
793 }
794 std::thread::scope(|scope| {
795 let stopping: Vec<_> = self
796 .0
797 .iter_mut()
798 .zip(running)
799 .map(|(member, was_running)| scope.spawn(move || member.stop_from(was_running)))
800 .collect();
801 stopping
802 .into_iter()
803 .filter_map(|handle| handle.join().ok().flatten())
804 .collect()
805 })
806 }
807}
808
809impl Drop for MemberSet {
810 fn drop(&mut self) {
811 let _ = self.stop_all();
812 }
813}
814
815fn member_args(member: &PlannedMember, workdir: &Path) -> Vec<String> {
819 vec![
820 "run".to_string(),
821 "--capsule".to_string(),
822 member.capsule.clone(),
823 "--capsule-version".to_string(),
824 member.version.clone(),
825 CAPSULE_SHA256_FLAG.to_string(),
826 member.sha256.clone(),
827 "--workdir".to_string(),
828 workdir.display().to_string(),
829 ]
830}
831
832fn pass_through(args: &mut Vec<String>, options: &FormationLaunchOptions) {
834 if options.no_env_file {
835 args.push("--no-env-file".to_string());
836 }
837 if let Some(class) = &options.containment {
838 args.push("--containment".to_string());
839 args.push(class.clone());
840 }
841}
842
843pub(crate) fn peer_args(
847 member: &PlannedMember,
848 options: &FormationLaunchOptions,
849 workdir: &Path,
850) -> Vec<String> {
851 let mut args = member_args(member, workdir);
852 args.extend([
853 STORE_ROOT_FLAG.to_string(),
854 options.project_dir.display().to_string(),
855 "--json".to_string(),
856 child_launch::READINESS_TOKENS_FLAG.to_string(),
857 "--bind".to_string(),
858 PEER_BIND_ADDR.to_string(),
859 ]);
860 pass_through(&mut args, options);
861 args
862}
863
864pub(crate) fn entry_args(member: &PlannedMember, options: &FormationLaunchOptions) -> Vec<String> {
868 let mut args = member_args(member, &options.project_dir);
869 args.extend([
870 "--lifecycle-task-acceptance".to_string(),
871 "single".to_string(),
872 "--lifecycle-after-task".to_string(),
873 "exit".to_string(),
874 ]);
875 if let Some(task) = &options.task {
876 args.push("--task".to_string());
877 args.push(task.clone());
878 }
879 if options.json {
880 args.push("--json".to_string());
881 }
882 if options.verbose {
883 args.push("--verbose".to_string());
884 }
885 pass_through(&mut args, options);
886 args
887}
888
889fn formations_dir() -> Result<PathBuf, String> {
891 Ok(crate::state_store::murmur_home_dir()?.join(FORMATIONS_DIR))
892}
893
894pub fn formation_member_roots(formation_id: &FormationId) -> Vec<PathBuf> {
901 match formations_dir() {
902 Ok(dir) => formation_member_roots_in(&dir, formation_id),
903 Err(_) => Vec::new(),
904 }
905}
906
907pub fn formation_member_roots_in(
909 formations_dir: &Path,
910 formation_id: &FormationId,
911) -> Vec<PathBuf> {
912 let dir = formations_dir.join(formation_id.as_str());
913 let entries = match std::fs::read_dir(&dir) {
914 Ok(entries) => entries,
915 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Vec::new(),
916 Err(_) => return vec![dir],
917 };
918 let mut roots: Vec<PathBuf> = entries
919 .filter_map(Result::ok)
920 .filter(|entry| entry.file_type().is_ok_and(|kind| kind.is_dir()))
921 .map(|entry| entry.path().join(".murmur"))
922 .filter(|root| root.exists())
923 .collect();
924 roots.sort();
925 roots
926}
927
928#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
934pub struct FormationMarker {
935 pub project_dir: PathBuf,
937 pub launcher_pid: u32,
939 pub launcher_start: String,
942}
943
944impl FormationMarker {
945 pub fn for_this_launcher(project_dir: &Path) -> Self {
947 let pid = std::process::id();
948 Self {
949 project_dir: canonical_project_dir(project_dir),
950 launcher_pid: pid,
951 launcher_start: crate::running::process_start_token(pid).unwrap_or_default(),
952 }
953 }
954
955 pub fn read(formation_dir: &Path) -> Option<Self> {
957 let raw = std::fs::read_to_string(formation_dir.join(FORMATION_MARKER_FILE)).ok()?;
958 serde_json::from_str(&raw).ok()
959 }
960}
961
962pub(crate) fn canonical_project_dir(project_dir: &Path) -> PathBuf {
965 std::fs::canonicalize(project_dir).unwrap_or_else(|_| project_dir.to_path_buf())
966}
967
968fn write_formation_marker(options: &FormationLaunchOptions) -> Result<(), String> {
971 let formations_dir = launch_formations_dir(options)?;
972 let formation_dir = formations_dir.join(options.formation_id.as_str());
973 for dir in [formations_dir.as_path(), formation_dir.as_path()] {
974 crate::state_store::ensure_private_dir(dir, crate::murmur_home::MURMUR_HOME_DIR_MODE)
975 .map_err(|reason| format!("{}: {reason}", dir.display()))?;
976 }
977 let marker = FormationMarker::for_this_launcher(&options.project_dir);
978 let json = serde_json::to_string(&marker).map_err(|error| error.to_string())?;
979 crate::murmur_home::write_private_file(
980 &formation_dir.join(FORMATION_MARKER_FILE),
981 format!("{json}\n").as_bytes(),
982 )
983}
984
985pub fn prune_ended_formations(
992 current: &FormationId,
993 project_dir: &Path,
994 policy: &murmur_artifact::TraceRetainConfig,
995) -> Vec<crate::retention::PrunedFormation> {
996 let (Ok(formations_dir), Ok(running_dir)) =
997 (formations_dir(), crate::running::running_dir_location())
998 else {
999 return Vec::new();
1000 };
1001 prune_ended_formations_in(&formations_dir, &running_dir, current, project_dir, policy)
1002}
1003
1004pub fn prune_ended_formations_in(
1007 formations_dir: &Path,
1008 running_dir: &Path,
1009 current: &FormationId,
1010 project_dir: &Path,
1011 policy: &murmur_artifact::TraceRetainConfig,
1012) -> Vec<crate::retention::PrunedFormation> {
1013 let records = crate::running::scan(running_dir)
1015 .ok()
1016 .map(|scan| scan.records);
1017 crate::retention::prune_formations(
1018 formations_dir,
1019 current,
1020 project_dir,
1021 policy,
1022 crate::retention::now_ms(),
1023 |id, marker| formation_is_live(id, marker, records.as_deref()),
1024 )
1025}
1026
1027fn formation_is_live(
1034 id: &FormationId,
1035 marker: &FormationMarker,
1036 records: Option<&[crate::running::RunningRecord]>,
1037) -> bool {
1038 let not_gone = |state| !matches!(state, crate::running::ProcessState::Gone(_));
1039 let Some(records) = records else {
1040 return true;
1041 };
1042 not_gone(crate::running::process_state_of(
1043 marker.launcher_pid,
1044 &marker.launcher_start,
1045 )) || records.iter().any(|record| {
1046 record.formation_id.as_ref() == Some(id) && not_gone(crate::running::process_state(record))
1047 })
1048}
1049
1050fn make_member_workdir(
1057 formations_dir: &Path,
1058 formation_id: &FormationId,
1059 member: &str,
1060) -> Result<PathBuf, String> {
1061 use std::os::unix::fs::DirBuilderExt;
1062
1063 let formation_dir = formations_dir.join(formation_id.as_str());
1064 let workdir = formation_dir.join(member);
1065 let failed = |path: &Path, reason: String| {
1066 format!(
1067 "its directory {} could not be made: {}: {reason}",
1068 workdir.display(),
1069 path.display()
1070 )
1071 };
1072 for dir in [formations_dir, formation_dir.as_path()] {
1073 crate::state_store::ensure_private_dir(dir, crate::murmur_home::MURMUR_HOME_DIR_MODE)
1074 .map_err(|reason| failed(dir, reason))?;
1075 }
1076 std::fs::DirBuilder::new()
1077 .mode(crate::murmur_home::MURMUR_HOME_DIR_MODE)
1078 .create(&workdir)
1079 .map_err(|error| failed(&workdir, error.to_string()))?;
1080 Ok(workdir)
1082}
1083
1084fn launch_formations_dir(options: &FormationLaunchOptions) -> Result<PathBuf, String> {
1087 match &options.formations_dir {
1088 Some(dir) => Ok(dir.clone()),
1089 None => {
1090 crate::murmur_home::ensure_murmur_home()?;
1091 formations_dir()
1092 }
1093 }
1094}
1095
1096enum PeerOutcome {
1098 Ready(ReadyPeer),
1099 Failed(MemberFailureReason),
1100 Abandoned,
1102}
1103
1104struct Channel {
1107 writer: std::io::PipeWriter,
1108 reader: std::io::PipeReader,
1109}
1110
1111impl Channel {
1112 fn open(first_line: &str) -> Result<Self, MemberFailureReason> {
1115 let (reader, mut writer) = std::io::pipe().map_err(|error| {
1116 not_started(&format!(
1117 "its formation channel could not be created: {error}"
1118 ))
1119 })?;
1120 writer
1121 .write_all(format!("{first_line}\n").as_bytes())
1122 .map_err(|error| {
1123 not_started(&format!(
1124 "its formation channel could not be written: {error}"
1125 ))
1126 })?;
1127 Ok(Self { writer, reader })
1128 }
1129
1130 fn fd(&self) -> i32 {
1132 use std::os::fd::AsRawFd;
1133 self.reader.as_raw_fd()
1134 }
1135
1136 fn member_env(&self, formation_id: &FormationId) -> Vec<(String, String)> {
1140 let (id_name, id_value) = formation_id.env_pair();
1141 vec![
1142 (id_name.to_string(), id_value),
1143 (FORMATION_CHANNEL_ENV.to_string(), self.fd().to_string()),
1144 crate::host_warnings::reported_env_pair(),
1145 ]
1146 }
1147}
1148
1149fn start_peer(
1154 binary: &Path,
1155 member: &PlannedMember,
1156 first_line: &str,
1157 options: &FormationLaunchOptions,
1158 abort: &AtomicBool,
1159 #[cfg(unix)] mut lifeline: crate::lifeline::MemberLifeline,
1160) -> (Option<MemberProcess>, PeerOutcome) {
1161 let workdir = match launch_formations_dir(options)
1162 .and_then(|dir| make_member_workdir(&dir, &options.formation_id, &member.name))
1163 {
1164 Ok(workdir) => workdir,
1165 Err(error) => return (None, PeerOutcome::Failed(not_started(&error))),
1166 };
1167 let args = peer_args(member, options, &workdir);
1168 let prefix = format!("[{}] ", member.name);
1169 let channel = match Channel::open(first_line) {
1170 Ok(channel) => channel,
1171 Err(reason) => return (None, PeerOutcome::Failed(reason)),
1172 };
1173 let set = channel.member_env(&options.formation_id);
1174 let spawned_at = Instant::now();
1175 let deadline = spawned_at + options.ready_timeout;
1176 let started = child_launch::start_process(&ProcessLaunch {
1177 binary,
1178 args: &args,
1179 cwd: None,
1180 env: ProcessEnv::Inherited {
1181 set: &set,
1182 remove: &[FORMATION_PEERS_ENV, SPAWNER_ENV, SPAWNER_LIFELINE_ENV],
1183 },
1184 stdin_line: None,
1185 stderr_prefix: Some(&prefix),
1186 inherit_output: false,
1187 own_process_group: true,
1188 inherit_fd: Some(channel.fd()),
1189 #[cfg(unix)]
1190 lifeline: Some(crate::child_launch::Lifeline::Member(&lifeline)),
1191 });
1192 let Channel { writer, reader } = channel;
1194 drop(reader);
1195 let StartedProcess {
1196 mut child,
1197 stderr_tail,
1198 ..
1199 } = match started {
1200 Ok(started) => started,
1201 Err(error) => {
1202 return (
1203 None,
1204 PeerOutcome::Failed(MemberFailureReason::NotStarted {
1205 error: format!("{}: {error}", binary.display()),
1206 }),
1207 )
1208 }
1209 };
1210 #[cfg(unix)]
1211 lifeline.spawned();
1212 let stdout = child.stdout.take();
1213 let mut process = MemberProcess::new(
1214 &member.name,
1215 child,
1216 true,
1217 options.stop_grace,
1218 writer,
1219 #[cfg(unix)]
1220 Some(lifeline),
1221 );
1222 let Some(stdout) = stdout else {
1223 return (
1224 Some(process),
1225 PeerOutcome::Failed(MemberFailureReason::NotStarted {
1226 error: "the process exposed no standard output".to_string(),
1227 }),
1228 );
1229 };
1230 let lines = child_launch::relay_readiness_line(stdout, AfterReadiness::PrefixedStderr(prefix));
1231
1232 let line = loop {
1234 if abort.load(Ordering::SeqCst) {
1235 return (Some(process), PeerOutcome::Abandoned);
1236 }
1237 let now = Instant::now();
1238 if now >= deadline {
1239 return (
1240 Some(process),
1241 PeerOutcome::Failed(MemberFailureReason::DidNotReport {
1242 deadline: options.ready_timeout,
1243 }),
1244 );
1245 }
1246 match lines.recv_timeout(POLL_INTERVAL.min(deadline - now)) {
1247 Ok(Some(line)) => break line,
1248 Ok(None) | Err(mpsc::RecvTimeoutError::Disconnected) => {
1249 let closed_at = Instant::now();
1252 let status = loop {
1253 if let Some(status) = process.exit_status_now() {
1254 break status;
1255 }
1256 if closed_at.elapsed() >= EXIT_AFTER_EOF_WAIT {
1257 process.reap();
1258 break process.status.unwrap_or_default();
1259 }
1260 std::thread::sleep(POLL_INTERVAL);
1261 };
1262 return (
1263 Some(process),
1264 PeerOutcome::Failed(MemberFailureReason::ExitedBeforeReporting {
1265 status: describe_status(status),
1266 stderr_tail: stderr_tail.lines_at_end(),
1267 }),
1268 );
1269 }
1270 Err(mpsc::RecvTimeoutError::Timeout) => {}
1271 }
1272 };
1273
1274 let report = match read_report(&line, &options.formation_id) {
1275 Ok(report) => report,
1276 Err(reason) => return (Some(process), PeerOutcome::Failed(reason)),
1277 };
1278
1279 let mut last_error;
1281 loop {
1282 if abort.load(Ordering::SeqCst) {
1283 return (Some(process), PeerOutcome::Abandoned);
1284 }
1285 if let Some(status) = process.exit_status_now() {
1286 return (
1287 Some(process),
1288 PeerOutcome::Failed(MemberFailureReason::ExitedBeforeDoorAnswered {
1289 status: describe_status(status),
1290 stderr_tail: stderr_tail.lines_at_end(),
1291 }),
1292 );
1293 }
1294 match door_answers_as(
1295 crate::running::probe_door_session_id(&report.url, report.door_token.as_ref()),
1296 &report.session_id,
1297 ) {
1298 Ok(()) => {
1299 let pid = process.pid;
1300 return (
1301 Some(process),
1302 PeerOutcome::Ready(ReadyPeer {
1303 name: member.name.clone(),
1304 capsule: member.capsule.clone(),
1305 version: member.version.clone(),
1306 pid,
1307 session_id: report.session_id,
1308 url: report.url,
1309 workdir,
1310 door_token: report.door_token,
1311 }),
1312 );
1313 }
1314 Err(why) => last_error = why,
1315 }
1316 let now = Instant::now();
1317 if now >= deadline {
1318 return (
1319 Some(process),
1320 PeerOutcome::Failed(MemberFailureReason::DoorDidNotAnswer {
1321 url: report.url,
1322 session_id: report.session_id,
1323 deadline: options.ready_timeout,
1324 last_error,
1325 }),
1326 );
1327 }
1328 std::thread::sleep(READY_PROBE_INTERVAL.min(deadline - now));
1329 }
1330}
1331
1332fn door_answers_as(probed: Result<String, String>, reported: &str) -> Result<(), String> {
1336 match probed {
1337 Ok(session_id) if session_id == reported => Ok(()),
1338 Ok(other) => Err(format!("the door answers as session {other}")),
1339 Err(error) => Err(error),
1340 }
1341}
1342
1343struct Report {
1345 session_id: String,
1346 url: String,
1348 door_token: Option<DoorToken>,
1349}
1350
1351fn read_report(line: &str, expected: &FormationId) -> Result<Report, MemberFailureReason> {
1353 let report: serde_json::Value =
1354 serde_json::from_str(line).map_err(|error| MemberFailureReason::UnreadableReport {
1355 why: format!("it does not parse as JSON ({error})"),
1356 })?;
1357 let reported = report
1358 .get("formation_id")
1359 .and_then(serde_json::Value::as_str);
1360 if reported != Some(expected.as_str()) {
1361 return Err(MemberFailureReason::WrongFormation {
1362 reported: reported.map(str::to_string),
1363 expected: expected.as_str().to_string(),
1364 });
1365 }
1366 let session_id = report
1367 .get("session_id")
1368 .and_then(serde_json::Value::as_str)
1369 .filter(|session_id| !session_id.is_empty())
1370 .ok_or_else(|| MemberFailureReason::UnreadableReport {
1371 why: "it names no session_id".to_string(),
1372 })?;
1373 let url = report
1374 .get("url")
1375 .and_then(serde_json::Value::as_str)
1376 .unwrap_or_default();
1377 if url.is_empty() {
1378 return Err(MemberFailureReason::ReportedNoDoor);
1379 }
1380 Ok(Report {
1381 session_id: session_id.to_string(),
1382 url: format!("http://{}", url.trim_start_matches("http://")),
1383 door_token: child_launch::readiness_door_token(&report),
1384 })
1385}
1386
1387pub fn launch_formation(
1395 roster: &AdmittedRoster,
1396 options: FormationLaunchOptions,
1397) -> Result<RunningFormation, MemberLaunchFailure> {
1398 launch_plan(FormationPlan::from_roster(roster), options)
1399}
1400
1401pub(crate) fn launch_plan(
1402 plan: FormationPlan,
1403 options: FormationLaunchOptions,
1404) -> Result<RunningFormation, MemberLaunchFailure> {
1405 install_panic_hook();
1406 let binary = match child_launch::mur_binary() {
1407 Ok(binary) => binary,
1408 Err(error) => {
1409 return Err(MemberLaunchFailure {
1410 failed: plan
1411 .peers
1412 .iter()
1413 .chain([&plan.entry])
1414 .map(|member| failure(member, not_started(&error.to_string())))
1415 .collect(),
1416 stopped: Vec::new(),
1417 interrupted_by: None,
1418 })
1419 }
1420 };
1421
1422 let authority = match FormationAuthority::generate(&options.formation_id) {
1423 Ok(authority) => authority,
1424 Err(error) => {
1425 return Err(MemberLaunchFailure {
1426 failed: plan
1427 .peers
1428 .iter()
1429 .chain([&plan.entry])
1430 .map(|member| failure(member, not_started(&error)))
1431 .collect(),
1432 stopped: Vec::new(),
1433 interrupted_by: None,
1434 })
1435 }
1436 };
1437 #[cfg(unix)]
1441 let (peer_lifelines, entry_lifeline) = match member_lifelines(plan.peers.len() + 1) {
1442 Ok(mut lifelines) => {
1443 let entry = lifelines.pop();
1444 (lifelines, entry)
1445 }
1446 Err(error) => {
1447 return Err(MemberLaunchFailure {
1448 failed: plan
1449 .peers
1450 .iter()
1451 .chain([&plan.entry])
1452 .map(|member| {
1453 failure(
1454 member,
1455 not_started(&format!("its lifeline could not be created: {error}")),
1456 )
1457 })
1458 .collect(),
1459 stopped: Vec::new(),
1460 interrupted_by: None,
1461 })
1462 }
1463 };
1464 let launcher = ProcessIdentity::of_this_process();
1468 let first_lines: Vec<String> = plan
1469 .peers
1470 .iter()
1471 .map(|member| plan.first_line(&authority, launcher.as_ref(), &member.name))
1472 .collect();
1473 if !plan.peers.is_empty() {
1476 if let Err(reason) = write_formation_marker(&options) {
1477 crate::runtime_err!(
1478 "[mur run] warning: formation {} has no ownership marker ({reason}); its \
1479 directory will never be removed automatically",
1480 options.formation_id
1481 );
1482 }
1483 }
1484
1485 let abort = AtomicBool::new(false);
1486 let mut outcomes: Vec<Option<(Option<MemberProcess>, PeerOutcome)>> =
1487 plan.peers.iter().map(|_| None).collect();
1488 let mut interrupted_by = caught_signal();
1489 if interrupted_by.is_some() {
1490 abort.store(true, Ordering::SeqCst);
1491 }
1492 std::thread::scope(|scope| {
1493 let (tx, rx) = mpsc::channel();
1494 #[cfg(unix)]
1495 let mut peer_lifelines = peer_lifelines.into_iter();
1496 for (index, member) in plan.peers.iter().enumerate() {
1497 let tx = tx.clone();
1498 let (binary, options, abort) = (&binary, &options, &abort);
1499 let first_line = first_lines[index].as_str();
1500 #[cfg(unix)]
1501 let lifeline = peer_lifelines
1502 .next()
1503 .expect("member_lifelines made one lifeline per peer");
1504 scope.spawn(move || {
1505 let started = start_peer(
1506 binary,
1507 member,
1508 first_line,
1509 options,
1510 abort,
1511 #[cfg(unix)]
1512 lifeline,
1513 );
1514 let _ = tx.send((index, started));
1515 });
1516 }
1517 drop(tx);
1518 loop {
1519 match rx.recv_timeout(POLL_INTERVAL) {
1520 Ok((index, outcome)) => {
1521 if matches!(outcome.1, PeerOutcome::Failed(_)) {
1522 abort.store(true, Ordering::SeqCst);
1523 }
1524 outcomes[index] = Some(outcome);
1525 }
1526 Err(mpsc::RecvTimeoutError::Timeout) => {}
1527 Err(mpsc::RecvTimeoutError::Disconnected) => break,
1528 }
1529 if interrupted_by.is_none() {
1530 if let Some(signal) = caught_signal() {
1531 interrupted_by = Some(signal);
1532 abort.store(true, Ordering::SeqCst);
1533 }
1534 }
1535 }
1536 });
1537
1538 let mut members = MemberSet(Vec::new());
1539 let mut peers = Vec::new();
1540 let mut failed = Vec::new();
1541 for (member, outcome) in plan.peers.iter().zip(outcomes) {
1542 let Some((process, outcome)) = outcome else {
1543 continue;
1544 };
1545 members.0.extend(process);
1546 match outcome {
1547 PeerOutcome::Ready(peer) => peers.push(peer),
1548 PeerOutcome::Failed(reason) => failed.push(failure(member, reason)),
1549 PeerOutcome::Abandoned => {}
1550 }
1551 }
1552
1553 if !failed.is_empty() || interrupted_by.is_some() || peers.len() != plan.peers.len() {
1554 let stopped = members.stop_all();
1555 return Err(MemberLaunchFailure {
1556 failed,
1557 stopped,
1558 interrupted_by,
1559 });
1560 }
1561
1562 for member in &mut members.0 {
1565 let addresses = addresses_of(&peers, plan.callees_of(&member.name));
1566 if !addresses.is_empty() {
1567 member.send_line(&render_address_line(&addresses));
1568 }
1569 }
1570
1571 Ok(RunningFormation {
1572 formation_id: options.formation_id.clone(),
1573 entry: plan.entry.clone(),
1574 plan,
1575 peers,
1576 members,
1577 entry_index: None,
1578 #[cfg(unix)]
1579 entry_lifeline,
1580 binary,
1581 options,
1582 authority,
1583 launcher,
1584 })
1585}
1586
1587fn addresses_of(ready: &[ReadyPeer], callees: &[String]) -> Vec<FormationPeer> {
1589 ready
1590 .iter()
1591 .filter(|peer| callees.contains(&peer.name))
1592 .map(|peer| FormationPeer {
1593 name: peer.name.clone(),
1594 url: peer.url.clone(),
1595 })
1596 .collect()
1597}
1598
1599#[cfg(unix)]
1601fn member_lifelines(count: usize) -> std::io::Result<Vec<crate::lifeline::MemberLifeline>> {
1602 (0..count)
1603 .map(|_| crate::lifeline::MemberLifeline::new())
1604 .collect()
1605}
1606
1607fn failure(member: &PlannedMember, reason: MemberFailureReason) -> MemberFailure {
1608 MemberFailure {
1609 name: member.name.clone(),
1610 capsule: member.capsule.clone(),
1611 version: member.version.clone(),
1612 reason,
1613 }
1614}
1615
1616fn not_started(error: &str) -> MemberFailureReason {
1617 MemberFailureReason::NotStarted {
1618 error: error.to_string(),
1619 }
1620}
1621
1622pub struct RunningFormation {
1628 formation_id: FormationId,
1629 entry: PlannedMember,
1630 plan: FormationPlan,
1631 peers: Vec<ReadyPeer>,
1632 members: MemberSet,
1635 entry_index: Option<usize>,
1637 #[cfg(unix)]
1640 entry_lifeline: Option<crate::lifeline::MemberLifeline>,
1641 binary: PathBuf,
1642 options: FormationLaunchOptions,
1643 authority: FormationAuthority,
1644 launcher: Option<ProcessIdentity>,
1646}
1647
1648impl std::fmt::Debug for RunningFormation {
1649 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1650 f.debug_struct("RunningFormation")
1651 .field("formation_id", &self.formation_id)
1652 .field("entry", &self.entry.name)
1653 .field("peers", &self.peers)
1654 .finish()
1655 }
1656}
1657
1658impl RunningFormation {
1659 pub fn formation_id(&self) -> &FormationId {
1660 &self.formation_id
1661 }
1662
1663 pub fn entry_name(&self) -> &str {
1665 &self.entry.name
1666 }
1667
1668 pub fn peers(&self) -> &[ReadyPeer] {
1670 &self.peers
1671 }
1672
1673 pub fn callees_of(&self, member: &str) -> &[String] {
1676 self.plan.callees_of(member)
1677 }
1678
1679 pub fn entry_pid(&self) -> Option<u32> {
1681 self.entry_index.map(|index| self.members.0[index].pid)
1682 }
1683
1684 pub fn start_entry(&mut self) -> Result<u32, MemberLaunchFailure> {
1688 let args = entry_args(&self.entry, &self.options);
1689 let channel = Channel::open(&self.plan.first_line(
1691 &self.authority,
1692 self.launcher.as_ref(),
1693 &self.entry.name,
1694 ))
1695 .and_then(|mut channel| {
1696 let addresses = addresses_of(&self.peers, self.plan.callees_of(&self.entry.name));
1697 if !addresses.is_empty() {
1698 channel
1699 .writer
1700 .write_all(format!("{}\n", render_address_line(&addresses)).as_bytes())
1701 .map_err(|error| {
1702 not_started(&format!(
1703 "its formation channel could not be written: {error}"
1704 ))
1705 })?;
1706 }
1707 Ok(channel)
1708 });
1709 let channel = match channel {
1710 Ok(channel) => channel,
1711 Err(reason) => {
1712 let stopped = self.members.stop_all();
1713 return Err(MemberLaunchFailure {
1714 failed: vec![failure(&self.entry, reason)],
1715 stopped,
1716 interrupted_by: None,
1717 });
1718 }
1719 };
1720 let set = channel.member_env(&self.formation_id);
1721 let started = child_launch::start_process(&ProcessLaunch {
1722 binary: &self.binary,
1723 args: &args,
1724 cwd: None,
1725 env: ProcessEnv::Inherited {
1726 set: &set,
1727 remove: &[SPAWNER_ENV, FORMATION_PEERS_ENV, SPAWNER_LIFELINE_ENV],
1728 },
1729 stdin_line: None,
1730 stderr_prefix: None,
1731 inherit_output: true,
1732 own_process_group: false,
1733 inherit_fd: Some(channel.fd()),
1734 #[cfg(unix)]
1735 lifeline: self
1736 .entry_lifeline
1737 .as_ref()
1738 .map(crate::child_launch::Lifeline::Member),
1739 });
1740 let Channel { writer, reader } = channel;
1741 drop(reader);
1742 match started {
1743 Ok(started) => {
1744 #[cfg(unix)]
1745 let lifeline = self.entry_lifeline.take().map(|mut lifeline| {
1746 lifeline.spawned();
1747 lifeline
1748 });
1749 let process = MemberProcess::new(
1750 &self.entry.name,
1751 started.child,
1752 false,
1753 self.options.stop_grace,
1754 writer,
1755 #[cfg(unix)]
1756 lifeline,
1757 );
1758 let pid = process.pid;
1759 self.members.0.push(process);
1760 self.entry_index = Some(self.members.0.len() - 1);
1761 Ok(pid)
1762 }
1763 Err(error) => {
1764 let stopped = self.members.stop_all();
1765 Err(MemberLaunchFailure {
1766 failed: vec![failure(
1767 &self.entry,
1768 not_started(&format!("{}: {error}", self.binary.display())),
1769 )],
1770 stopped,
1771 interrupted_by: None,
1772 })
1773 }
1774 }
1775 }
1776
1777 pub fn wait(mut self) -> FormationExit {
1780 let ending = match self.entry_index {
1781 None => caught_signal()
1782 .map(FormationEnding::Signalled)
1783 .unwrap_or(FormationEnding::EntryExited(ExitStatus::default())),
1784 Some(index) => loop {
1785 if let Some(status) = self.members.0[index].exit_status_now() {
1786 break FormationEnding::EntryExited(status);
1787 }
1788 if let Some(signal) = caught_signal() {
1789 break FormationEnding::Signalled(signal);
1790 }
1791 std::thread::sleep(POLL_INTERVAL);
1792 },
1793 };
1794 let stopped = self.members.stop_all();
1795 FormationExit { ending, stopped }
1796 }
1797}
1798
1799#[cfg(test)]
1800mod tests {
1801 use super::*;
1802 use std::io::{Read, Write};
1803 use std::net::TcpListener;
1804 use std::sync::atomic::AtomicUsize;
1805 use std::sync::Arc;
1806
1807 const LIVENESS_BOUND: Duration = Duration::from_secs(120);
1811
1812 fn load_average() -> String {
1814 std::fs::read_to_string("/proc/loadavg")
1815 .unwrap_or_default()
1816 .split(' ')
1817 .take(3)
1818 .collect::<Vec<_>>()
1819 .join(" ")
1820 }
1821
1822 #[test]
1825 fn the_production_bounds_are_180s_and_a_grace_past_the_runtime_teardown() {
1826 assert_eq!(MEMBER_READY_TIMEOUT, Duration::from_secs(180));
1827 assert_eq!(MEMBER_STOP_GRACE, Duration::from_secs(25));
1828 assert!(MEMBER_STOP_GRACE > crate::runtime::TERMINATE_TEARDOWN_DEADLINE);
1829 }
1830
1831 fn member(name: &str) -> PlannedMember {
1832 PlannedMember {
1833 name: name.to_string(),
1834 capsule: name.to_string(),
1835 version: "0.1.0".to_string(),
1836 sha256: "ab".repeat(32),
1837 }
1838 }
1839
1840 #[test]
1843 fn every_member_is_told_host_warnings_were_reported() {
1844 let channel = Channel::open("first").expect("a channel opens");
1845 let env = channel.member_env(&FormationId::mint());
1846 let values: Vec<&str> = env
1847 .iter()
1848 .filter(|(key, _)| key == crate::host_warnings::HOST_WARNINGS_REPORTED_ENV)
1849 .map(|(_, value)| value.as_str())
1850 .collect();
1851 assert_eq!(values, vec!["1"], "{env:?}");
1852 }
1853
1854 fn options(deadline: Duration) -> FormationLaunchOptions {
1855 let mut options =
1856 FormationLaunchOptions::new(FormationId::mint(), PathBuf::from("/tmp/project"));
1857 options.ready_timeout = deadline;
1858 options.stop_grace = Duration::from_secs(2);
1859 options.formations_dir = Some(tempfile::tempdir().unwrap().keep());
1861 options
1862 }
1863
1864 #[test]
1865 fn member_argv_is_exactly_the_documented_shape() {
1866 let mut options = options(MEMBER_READY_TIMEOUT);
1867 let coder = member("coder");
1868 let workdir = PathBuf::from("/home/u/.murmur/formations/frm_x/coder");
1869 assert_eq!(
1870 peer_args(&coder, &options, &workdir),
1871 [
1872 "run",
1873 "--capsule",
1874 "coder",
1875 "--capsule-version",
1876 "0.1.0",
1877 "--capsule-sha256",
1878 &"ab".repeat(32),
1879 "--workdir",
1880 "/home/u/.murmur/formations/frm_x/coder",
1881 "--store-root",
1882 "/tmp/project",
1883 "--json",
1884 "--readiness-tokens",
1885 "--bind",
1886 "127.0.0.1",
1887 ]
1888 );
1889 options.task = Some("probe".to_string());
1890 options.json = true;
1891 options.verbose = true;
1892 options.no_env_file = true;
1893 options.containment = Some("scoped".to_string());
1894 let entry = entry_args(&member("planner"), &options);
1895 assert_eq!(&entry[7..9], ["--workdir", "/tmp/project"]);
1896 assert_eq!(
1897 &entry[9..],
1898 [
1899 "--lifecycle-task-acceptance",
1900 "single",
1901 "--lifecycle-after-task",
1902 "exit",
1903 "--task",
1904 "probe",
1905 "--json",
1906 "--verbose",
1907 "--no-env-file",
1908 "--containment",
1909 "scoped",
1910 ]
1911 );
1912 assert!(
1913 !entry.iter().any(|arg| arg == "--bind"
1914 || arg == STORE_ROOT_FLAG
1915 || arg == child_launch::READINESS_TOKENS_FLAG),
1916 "{entry:?}"
1917 );
1918 let peer = peer_args(&coder, &options, &workdir);
1919 assert!(peer.ends_with(&[
1920 "--no-env-file".to_string(),
1921 "--containment".to_string(),
1922 "scoped".to_string()
1923 ]));
1924 assert!(!peer.iter().any(|arg| arg == "--task" || arg == "--verbose"));
1925 }
1926
1927 fn binary_guard() -> std::sync::MutexGuard<'static, ()> {
1930 crate::formation::FORMATION_ENV_LOCK
1931 .lock()
1932 .unwrap_or_else(|poisoned| poisoned.into_inner())
1933 }
1934
1935 fn stand_in(dir: &Path, line: Option<&str>) -> PathBuf {
1938 let script = dir.join("mur");
1939 let print = match line {
1940 Some(line) => {
1941 std::fs::write(dir.join("line"), format!("{line}\n")).unwrap();
1942 format!("cat '{}'\n", dir.join("line").display())
1943 }
1944 None => String::new(),
1945 };
1946 let body = format!(
1949 "#!/bin/sh\ntrap 'echo TERM >> \"{dir}/sigterm\"' TERM\n{print}\
1950 cat \"/dev/fd/$MURMUR_FORMATION_LIFELINE\" >/dev/null\n",
1951 dir = dir.display()
1952 );
1953 std::fs::write(&script, body).unwrap();
1954 use std::os::unix::fs::PermissionsExt;
1955 std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
1956 script
1957 }
1958
1959 fn plan_of(peer: &str) -> FormationPlan {
1960 FormationPlan {
1961 peers: vec![member(peer)],
1962 entry: member("planner"),
1963 callees: vec![("planner".to_string(), vec![peer.to_string()])],
1964 }
1965 }
1966
1967 fn refused(line: Option<&str>, deadline: Duration) -> (MemberLaunchFailure, FormationId) {
1970 let _guard = binary_guard();
1971 let dir = tempfile::tempdir().unwrap();
1972 let options = options(deadline);
1973 let id = options.formation_id.clone();
1974 let line = line.map(|line| line.replace("{formation}", id.as_str()));
1975 std::env::set_var(
1976 child_launch::MUR_BINARY_ENV,
1977 stand_in(dir.path(), line.as_deref()),
1978 );
1979 let result = launch_plan(plan_of("coder"), options);
1980 std::env::remove_var(child_launch::MUR_BINARY_ENV);
1981 match result {
1982 Ok(_) => panic!("the launch was expected to be refused"),
1983 Err(failure) => (failure, id),
1984 }
1985 }
1986
1987 fn assert_reaped(failure: &MemberLaunchFailure) {
1988 assert_eq!(failure.stopped.len(), 1, "{failure:?}");
1989 let pid = failure.stopped[0].pid;
1990 assert_eq!(failure.stopped[0].name, "coder");
1991 assert!(
1992 !crate::running::pid_is_alive(pid),
1993 "the stand-in {pid} is still running"
1994 );
1995 assert!(
1996 !Path::new(&format!("/proc/{pid}")).exists() || cfg!(not(target_os = "linux")),
1997 "the stand-in {pid} was not reaped"
1998 );
1999 }
2000
2001 fn only_reason(failure: &MemberLaunchFailure) -> &MemberFailureReason {
2002 assert_eq!(failure.failed.len(), 1, "{failure:?}");
2003 assert_eq!(failure.failed[0].name, "coder");
2004 &failure.failed[0].reason
2005 }
2006
2007 fn silent_port() -> u16 {
2009 crate::pinned_port::claim()
2010 }
2011
2012 #[test]
2013 fn a_door_nothing_answers_on_is_refused_at_the_deadline() {
2014 let deadline = Duration::from_secs(10);
2015 let port = silent_port();
2016 let started = Instant::now();
2017 let (failure, _) = refused(
2018 Some(&format!(
2019 r#"{{"formation_id":"{{formation}}","session_id":"ses_standin","url":"localhost:{port}"}}"#
2020 )),
2021 deadline,
2022 );
2023 assert!(started.elapsed() >= deadline);
2024 match only_reason(&failure) {
2025 MemberFailureReason::DoorDidNotAnswer {
2026 url, session_id, ..
2027 } => {
2028 assert_eq!(url, &format!("http://localhost:{port}"));
2029 assert_eq!(session_id, "ses_standin");
2030 }
2031 MemberFailureReason::DidNotReport { .. } => panic!(
2034 "the stand-in's readiness line did not arrive within {deadline:?} (load average {}); \
2035 host load is the likely cause, since the stand-in prints it at once: {failure:?}",
2036 load_average()
2037 ),
2038 other => panic!("{other:?}"),
2039 }
2040 let message = failure.to_string();
2041 assert!(
2042 message.contains("did not answer as session ses_standin"),
2043 "{message}"
2044 );
2045 assert!(message.contains("'coder' (coder@0.1.0)"), "{message}");
2046 assert_reaped(&failure);
2047 }
2048
2049 fn card_door(mut answer: impl FnMut() -> Option<&'static str> + Send + 'static) -> u16 {
2053 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
2054 let port = listener.local_addr().unwrap().port();
2055 std::thread::spawn(move || {
2056 for stream in listener.incoming().flatten() {
2057 let mut stream = stream;
2058 let mut buffer = [0u8; 4096];
2059 let _ = stream.read(&mut buffer);
2060 let body = match answer() {
2061 Some(session_id) => serde_json::json!({
2062 "name": "coder",
2063 "capabilities": {"extensions": [{
2064 "uri": crate::identity::CAPSULE_EXTENSION_URI,
2065 "params": {"sessionId": session_id},
2066 }]},
2067 })
2068 .to_string(),
2069 None => "{}".to_string(),
2070 };
2071 let _ = write!(
2072 stream,
2073 "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\n\
2074 connection: close\r\n\r\n{body}",
2075 body.len()
2076 );
2077 }
2078 });
2079 port
2080 }
2081
2082 fn door(session_id: &'static str, answer_after: Duration) -> u16 {
2085 let opened = Instant::now();
2086 card_door(move || (opened.elapsed() >= answer_after).then_some(session_id))
2087 }
2088
2089 const WRONG_ANSWERS: usize = 3;
2091
2092 fn door_naming_another_session_first() -> (u16, Arc<AtomicUsize>) {
2096 let served = Arc::new(AtomicUsize::new(0));
2097 let counted = Arc::clone(&served);
2098 let port = card_door(move || {
2099 if counted.fetch_add(1, Ordering::SeqCst) < WRONG_ANSWERS {
2100 Some("ses_somebody_else")
2101 } else {
2102 Some("ses_standin")
2103 }
2104 });
2105 (port, served)
2106 }
2107
2108 #[test]
2111 fn a_door_naming_another_session_is_not_ready() {
2112 let _guard = binary_guard();
2113 let (port, served) = door_naming_another_session_first();
2114 let dir = tempfile::tempdir().unwrap();
2115 let options = options(LIVENESS_BOUND);
2116 let line = format!(
2117 r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
2118 options.formation_id
2119 );
2120 std::env::set_var(
2121 child_launch::MUR_BINARY_ENV,
2122 stand_in(dir.path(), Some(&line)),
2123 );
2124 let result = launch_plan(plan_of("coder"), options);
2125 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2126 let formation = match result {
2127 Ok(formation) => formation,
2128 Err(failure) => match only_reason(&failure) {
2129 MemberFailureReason::DidNotReport { .. } => panic!(
2130 "the stand-in's readiness line did not arrive within {LIVENESS_BOUND:?} \
2131 (load average {}); host load is the likely cause: {failure:?}",
2132 load_average()
2133 ),
2134 _ => panic!(
2135 "the launch was refused although the door answered as ses_standin from its \
2136 request {} on: {failure:?}",
2137 WRONG_ANSWERS + 1
2138 ),
2139 },
2140 };
2141 let peer = &formation.peers()[0];
2142 assert_eq!(peer.session_id, "ses_standin");
2143 let served = served.load(Ordering::SeqCst);
2144 assert!(
2145 served > WRONG_ANSWERS,
2146 "an answer naming another session was counted as ready: the door served {served} \
2147 request(s), and its first {WRONG_ANSWERS} named ses_somebody_else"
2148 );
2149 let pid = peer.pid;
2150 drop(formation);
2151 assert!(
2152 !crate::running::pid_is_alive(pid),
2153 "dropping the owner stops the peer"
2154 );
2155 }
2156
2157 #[test]
2158 fn a_door_counts_as_ready_only_when_it_answers_as_the_reported_session() {
2159 let port = door("ses_somebody_else", Duration::ZERO);
2160 let probed = crate::running::probe_door_session_id(&format!("localhost:{port}"), None);
2161 match door_answers_as(probed, "ses_standin") {
2162 Err(why) => assert!(
2163 why.contains("the door answers as session ses_somebody_else"),
2164 "{why}"
2165 ),
2166 Ok(()) => panic!("a door answering as ses_somebody_else counted as ses_standin"),
2167 }
2168 assert_eq!(
2169 door_answers_as(Ok("ses_standin".to_string()), "ses_standin"),
2170 Ok(())
2171 );
2172 assert_eq!(
2173 door_answers_as(Err("connection refused".to_string()), "ses_standin"),
2174 Err("connection refused".to_string())
2175 );
2176 }
2177
2178 #[test]
2179 fn an_empty_url_is_refused_as_no_door() {
2180 let (failure, _) = refused(
2181 Some(r#"{"formation_id":"{formation}","session_id":"ses_standin","url":""}"#),
2182 Duration::from_secs(10),
2183 );
2184 assert_eq!(only_reason(&failure), &MemberFailureReason::ReportedNoDoor);
2185 assert!(failure.to_string().contains("it reported no door"));
2186 assert_reaped(&failure);
2187 }
2188
2189 #[test]
2190 fn a_wrong_formation_or_none_is_refused_naming_what_was_reported() {
2191 let other = FormationId::mint();
2192 let (failure, id) = refused(
2193 Some(&format!(
2194 r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:1"}}"#,
2195 other.as_str()
2196 )),
2197 Duration::from_secs(10),
2198 );
2199 assert_eq!(
2200 only_reason(&failure),
2201 &MemberFailureReason::WrongFormation {
2202 reported: Some(other.as_str().to_string()),
2203 expected: id.as_str().to_string(),
2204 }
2205 );
2206 assert!(failure
2207 .to_string()
2208 .contains(&format!("it reported formation {other} instead of {id}")));
2209 assert_reaped(&failure);
2210
2211 let (failure, id) = refused(
2212 Some(r#"{"session_id":"ses_standin","url":"localhost:1"}"#),
2213 Duration::from_secs(10),
2214 );
2215 assert!(failure
2216 .to_string()
2217 .contains(&format!("it reported no formation instead of {id}")));
2218 assert_reaped(&failure);
2219 }
2220
2221 #[test]
2222 fn no_line_at_all_is_refused_at_the_deadline() {
2223 let started = Instant::now();
2224 let (failure, _) = refused(None, Duration::from_millis(600));
2225 assert!(started.elapsed() >= Duration::from_millis(600));
2226 assert_eq!(
2227 only_reason(&failure),
2228 &MemberFailureReason::DidNotReport {
2229 deadline: Duration::from_millis(600)
2230 }
2231 );
2232 assert_reaped(&failure);
2233 }
2234
2235 #[test]
2236 fn a_door_that_starts_answering_before_the_deadline_is_ready() {
2237 let _guard = binary_guard();
2238 let port = door("ses_standin", Duration::from_millis(400));
2239 let dir = tempfile::tempdir().unwrap();
2240 let options = options(Duration::from_secs(20));
2241 let line = format!(
2242 r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
2243 options.formation_id
2244 );
2245 std::env::set_var(
2246 child_launch::MUR_BINARY_ENV,
2247 stand_in(dir.path(), Some(&line)),
2248 );
2249 let result = launch_plan(plan_of("coder"), options);
2250 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2251 let formation = result.expect("the door answered before the deadline");
2252 assert_eq!(formation.peers().len(), 1);
2253 let peer = &formation.peers()[0];
2254 assert_eq!(peer.session_id, "ses_standin");
2255 assert_eq!(peer.url, format!("http://localhost:{port}"));
2256 assert_eq!(formation.callees_of("planner"), ["coder"]);
2257 assert!(formation.callees_of("coder").is_empty());
2258 let pid = peer.pid;
2259 assert!(crate::running::pid_is_alive(pid));
2260 let dropped = Instant::now();
2261 drop(formation);
2262 assert!(
2263 !crate::running::pid_is_alive(pid),
2264 "dropping the owner stops the peer"
2265 );
2266 assert!(
2267 dropped.elapsed() < Duration::from_secs(2),
2268 "the peer ended on its lifeline's EOF, inside the grace: {:?}",
2269 dropped.elapsed()
2270 );
2271 assert!(
2272 !dir.path().join("sigterm").exists(),
2273 "the launcher sent the peer SIGTERM"
2274 );
2275 }
2276
2277 #[test]
2279 fn a_member_deaf_to_its_lifeline_is_killed_after_the_grace() {
2280 let _guard = binary_guard();
2281 let port = door("ses_standin", Duration::ZERO);
2282 let dir = tempfile::tempdir().unwrap();
2283 let mut options = options(Duration::from_secs(20));
2284 options.stop_grace = Duration::from_millis(600);
2285 let line = format!(
2286 r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
2287 options.formation_id
2288 );
2289 let script = dir.path().join("mur");
2290 std::fs::write(dir.path().join("line"), format!("{line}\n")).unwrap();
2291 std::fs::write(
2292 &script,
2293 format!(
2294 "#!/bin/sh\ncat '{}'\nexec sleep 600\n",
2295 dir.path().join("line").display()
2296 ),
2297 )
2298 .unwrap();
2299 use std::os::unix::fs::PermissionsExt;
2300 std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
2301 std::env::set_var(child_launch::MUR_BINARY_ENV, &script);
2302 let result = launch_plan(plan_of("coder"), options);
2303 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2304 let formation = result.expect("the door answers");
2305 let pid = formation.peers()[0].pid;
2306 let dropped = Instant::now();
2307 drop(formation);
2308 assert!(dropped.elapsed() >= Duration::from_millis(600));
2309 assert!(!crate::running::pid_is_alive(pid));
2310 }
2311
2312 #[test]
2317 fn each_member_reads_only_its_own_credentials_and_callees_on_its_channel() {
2318 use crate::formation_credentials::{parse_address_line, FormationBundle};
2319 let _guard = binary_guard();
2320 let dir = tempfile::tempdir().unwrap();
2321 let coder_port = door("ses_coder", Duration::ZERO);
2322 let reviewer_port = door("ses_reviewer", Duration::ZERO);
2323 let options = options(Duration::from_secs(20));
2324 let id = options.formation_id.clone();
2325 let line = |session: &str, port: u16| {
2326 format!(
2327 r#"{{"formation_id":"{id}","session_id":"{session}","url":"localhost:{port}"}}"#
2328 )
2329 };
2330 let out = dir.path().display().to_string();
2331 let script = dir.path().join("mur");
2332 std::fs::write(
2333 &script,
2334 format!(
2335 "#!/bin/sh\n\
2336 case \" $* \" in\n\
2337 *' --capsule coder '*) name=coder; line='{coder}';;\n\
2338 *' --capsule reviewer '*) name=reviewer; line='{reviewer}';;\n\
2339 *) name=planner; line='';;\n\
2340 esac\n\
2341 env > '{out}/'$name.env\n\
2342 printf '%s\\n' \"$*\" > '{out}/'$name.argv\n\
2343 cat /dev/fd/$MURMUR_FORMATION_CHANNEL >> '{out}/'$name.channel &\n\
2344 [ -n \"$line\" ] && printf '%s\\n' \"$line\"\n\
2345 exec sleep 600\n",
2346 coder = line("ses_coder", coder_port),
2347 reviewer = line("ses_reviewer", reviewer_port),
2348 ),
2349 )
2350 .unwrap();
2351 use std::os::unix::fs::PermissionsExt;
2352 std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
2353 std::env::set_var(child_launch::MUR_BINARY_ENV, &script);
2354 let plan = FormationPlan {
2355 peers: vec![member("coder"), member("reviewer")],
2356 entry: member("planner"),
2357 callees: vec![
2360 ("planner".to_string(), vec!["coder".to_string()]),
2361 ("coder".to_string(), vec!["reviewer".to_string()]),
2362 ],
2363 };
2364 let result = launch_plan(plan, options);
2365 let mut formation = result.expect("both stand-ins came up");
2366 formation.start_entry().expect("the entry stand-in starts");
2367 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2368
2369 let read_lines = |name: &str, count: usize| -> Vec<String> {
2370 let path = dir.path().join(format!("{name}.channel"));
2371 let deadline = Instant::now() + Duration::from_secs(10);
2372 loop {
2373 let lines: Vec<String> = std::fs::read_to_string(&path)
2374 .unwrap_or_default()
2375 .lines()
2376 .map(str::to_string)
2377 .collect();
2378 if lines.len() >= count || Instant::now() >= deadline {
2379 return lines;
2380 }
2381 std::thread::sleep(POLL_INTERVAL);
2382 }
2383 };
2384 let coder = read_lines("coder", 2);
2385 let planner = read_lines("planner", 2);
2386 let reviewer = read_lines("reviewer", 1);
2387 std::thread::sleep(Duration::from_millis(300));
2388 assert_eq!(read_lines("reviewer", 1).len(), 1, "reviewer calls nobody");
2389
2390 let bundle = |line: &str| FormationBundle::parse_line(line).expect("a member's bundle");
2391 let names = |bundle: &FormationBundle| -> Vec<String> {
2392 bundle.calls.iter().map(|(name, _)| name.clone()).collect()
2393 };
2394 let (coder_bundle, planner_bundle, reviewer_bundle) =
2395 (bundle(&coder[0]), bundle(&planner[0]), bundle(&reviewer[0]));
2396 assert_eq!(coder_bundle.member, "coder");
2397 assert_eq!(names(&coder_bundle), ["reviewer"]);
2398 assert_eq!(names(&planner_bundle), ["coder"]);
2399 assert!(names(&reviewer_bundle).is_empty());
2400 let launcher = crate::running::ProcessIdentity::of_this_process();
2403 assert!(launcher.is_some());
2404 for bundle in [&coder_bundle, &planner_bundle, &reviewer_bundle] {
2405 assert_eq!(bundle.formation_id, id);
2406 assert_eq!(bundle.verify_key, coder_bundle.verify_key);
2407 assert_eq!(bundle.launcher, launcher, "{}", bundle.member);
2408 }
2409 assert!(!planner.join("\n").contains("reviewer"), "{planner:?}");
2410 let addresses = |line: &str| -> Vec<(String, String)> {
2411 parse_address_line(line)
2412 .unwrap()
2413 .into_iter()
2414 .map(|peer| (peer.name, peer.url))
2415 .collect()
2416 };
2417 assert_eq!(
2418 addresses(&coder[1]),
2419 [(
2420 "reviewer".to_string(),
2421 format!("http://localhost:{reviewer_port}")
2422 )]
2423 );
2424 assert_eq!(
2425 addresses(&planner[1]),
2426 [(
2427 "coder".to_string(),
2428 format!("http://localhost:{coder_port}")
2429 )]
2430 );
2431
2432 for name in ["coder", "reviewer", "planner"] {
2433 let env = std::fs::read_to_string(dir.path().join(format!("{name}.env"))).unwrap();
2434 let argv = std::fs::read_to_string(dir.path().join(format!("{name}.argv"))).unwrap();
2435 assert!(!env.contains("mft1.") && !argv.contains("mft1."), "{name}");
2436 assert!(!env.contains(FORMATION_PEERS_ENV), "{name}: {env}");
2437 assert!(
2438 env.contains(&format!("{FORMATION_CHANNEL_ENV}=")),
2439 "{name}: {env}"
2440 );
2441 }
2442 let pids: Vec<u32> = formation.peers().iter().map(|peer| peer.pid).collect();
2443 drop(formation);
2444 for pid in pids {
2445 assert!(!crate::running::pid_is_alive(pid), "{pid}");
2446 }
2447 }
2448
2449 #[test]
2450 fn a_registered_member_is_killed_by_the_panic_hooks_sweep() {
2451 use std::os::unix::process::CommandExt;
2452 let _guard = binary_guard();
2454 let mut child = std::process::Command::new("sleep")
2455 .arg("600")
2456 .process_group(0)
2457 .spawn()
2458 .unwrap();
2459 let slot = RegistrySlot::register(child.id(), true);
2460 assert!(slot.0.is_some());
2461 kill_registered_members();
2462 let status = child.wait().unwrap();
2463 use std::os::unix::process::ExitStatusExt;
2464 assert_eq!(status.signal(), Some(libc::SIGKILL));
2465 drop(slot);
2466 assert!(MEMBER_REGISTRY
2467 .iter()
2468 .all(|slot| slot.load(Ordering::SeqCst) != i64::from(child.id())));
2469 }
2470
2471 #[test]
2472 fn a_launch_failure_names_each_member_and_every_stopped_one() {
2473 let failure = MemberLaunchFailure {
2474 failed: vec![failure(
2475 &member("broken"),
2476 MemberFailureReason::ExitedBeforeReporting {
2477 status: "status 1".to_string(),
2478 stderr_tail: vec!["error[E-RUN-008]: missing".to_string()],
2479 },
2480 )],
2481 stopped: vec![
2482 StoppedMember {
2483 name: "coder".to_string(),
2484 pid: 7,
2485 },
2486 StoppedMember {
2487 name: "reviewer".to_string(),
2488 pid: 8,
2489 },
2490 ],
2491 interrupted_by: None,
2492 };
2493 assert_eq!(
2494 failure.to_string(),
2495 "the formation could not be launched\n 'broken' (broken@0.1.0): its process exited \
2496 with status 1 before reporting; its last stderr lines were:\n \
2497 error[E-RUN-008]: missing\n stopped: coder (pid 7), reviewer (pid 8)"
2498 );
2499 }
2500
2501 #[test]
2502 fn ready_peer_debug_prints_no_door_token() {
2503 let peer = ReadyPeer {
2504 name: "coder".to_string(),
2505 capsule: "coder".to_string(),
2506 version: "1.2.0".to_string(),
2507 pid: 1,
2508 session_id: "ses_x".to_string(),
2509 url: "http://localhost:1".to_string(),
2510 workdir: PathBuf::from("/home/u/.murmur/formations/frm_x/coder"),
2511 door_token: Some(DoorToken::new("mdt1.secretpayload.mac".to_string())),
2512 };
2513 let debug = format!("{peer:?}");
2514 assert!(!debug.contains("secretpayload"), "{debug}");
2515 assert!(debug.contains("<redacted>"), "{debug}");
2516 }
2517
2518 #[test]
2520 fn a_member_directory_is_made_owner_only_under_its_formation() {
2521 use std::os::unix::fs::PermissionsExt;
2522
2523 let home = tempfile::tempdir().unwrap();
2524 let formations = home.path().join(FORMATIONS_DIR);
2525 let id = FormationId::mint();
2526 let workdir = make_member_workdir(&formations, &id, "worker").unwrap();
2527 assert_eq!(workdir, formations.join(id.as_str()).join("worker"));
2528 for dir in [&formations, &formations.join(id.as_str()), &workdir] {
2529 let mode = std::fs::metadata(dir).unwrap().permissions().mode() & 0o777;
2530 assert_eq!(mode, 0o700, "{}", dir.display());
2531 }
2532 }
2533
2534 #[test]
2537 fn an_existing_member_path_is_refused_and_named() {
2538 let home = tempfile::tempdir().unwrap();
2539 let formations = home.path().join(FORMATIONS_DIR);
2540 let id = FormationId::mint();
2541 make_member_workdir(&formations, &id, "worker").unwrap();
2542 let again = make_member_workdir(&formations, &id, "worker").unwrap_err();
2543 let path = formations.join(id.as_str()).join("worker");
2544 assert!(again.contains(&path.display().to_string()), "{again}");
2545 std::os::unix::fs::symlink(home.path(), formations.join(id.as_str()).join("linked"))
2546 .unwrap();
2547 assert!(make_member_workdir(&formations, &id, "linked").is_err());
2548 }
2549
2550 #[test]
2552 fn a_peer_whose_directory_cannot_be_made_refuses_the_launch() {
2553 let _guard = binary_guard();
2554 let dir = tempfile::tempdir().unwrap();
2555 let options = options(Duration::from_secs(5));
2556 let formations = options.formations_dir.clone().unwrap();
2557 let taken = formations.join(options.formation_id.as_str()).join("coder");
2558 std::fs::create_dir_all(&taken).unwrap();
2559 std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2560 let result = launch_plan(plan_of("coder"), options);
2561 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2562 let Err(failure) = result else {
2563 panic!("the launch was expected to be refused");
2564 };
2565 assert!(failure.stopped.is_empty(), "{failure:?}");
2566 match only_reason(&failure) {
2567 MemberFailureReason::NotStarted { error } => {
2568 assert!(error.contains(&taken.display().to_string()), "{error}")
2569 }
2570 other => panic!("expected NotStarted, got {other:?}"),
2571 }
2572 }
2573
2574 #[test]
2578 fn member_roots_are_the_member_directories_that_hold_sessions() {
2579 assert!(formation_member_roots(&FormationId::mint()).is_empty());
2580
2581 let id = FormationId::mint();
2582 let home = tempfile::tempdir().unwrap();
2583 let formations = home.path().join(FORMATIONS_DIR);
2584 for member in ["worker", "reviewer", "idle"] {
2585 assert_eq!(
2586 make_member_workdir(&formations, &id, member).unwrap(),
2587 formations.join(id.as_str()).join(member)
2588 );
2589 }
2590 for member in ["worker", "reviewer"] {
2591 std::fs::create_dir(formations.join(id.as_str()).join(member).join(".murmur")).unwrap();
2592 }
2593 assert_eq!(
2594 formation_member_roots_in(&formations, &id),
2595 [
2596 formations.join(id.as_str()).join("reviewer/.murmur"),
2597 formations.join(id.as_str()).join("worker/.murmur"),
2598 ]
2599 );
2600 }
2601
2602 #[test]
2604 fn formation_marker_round_trips_as_three_keys() {
2605 let marker = FormationMarker {
2606 project_dir: PathBuf::from("/srv/project"),
2607 launcher_pid: 4242,
2608 launcher_start: "12345".to_string(),
2609 };
2610 let value = serde_json::to_value(&marker).unwrap();
2611 assert_eq!(
2612 value,
2613 serde_json::json!({
2614 "project_dir": "/srv/project",
2615 "launcher_pid": 4242,
2616 "launcher_start": "12345",
2617 })
2618 );
2619 let dir = tempfile::tempdir().unwrap();
2620 std::fs::write(dir.path().join(FORMATION_MARKER_FILE), value.to_string()).unwrap();
2621 assert_eq!(FormationMarker::read(dir.path()), Some(marker));
2622 std::fs::write(dir.path().join(FORMATION_MARKER_FILE), "{").unwrap();
2623 assert_eq!(FormationMarker::read(dir.path()), None);
2624 assert_eq!(FormationMarker::read(&dir.path().join("missing")), None);
2625 }
2626
2627 #[test]
2631 fn formation_marker_is_written_owner_only_before_the_first_peer_directory() {
2632 use std::os::unix::fs::PermissionsExt;
2633
2634 let _guard = binary_guard();
2635 let dir = tempfile::tempdir().unwrap();
2636 let project = tempfile::tempdir().unwrap();
2637 let mut options = options(Duration::from_secs(1));
2638 options.project_dir = project.path().join(".");
2639 let formation_dir = options
2640 .formations_dir
2641 .clone()
2642 .unwrap()
2643 .join(options.formation_id.as_str());
2644 std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2645 let result = launch_plan(plan_of("coder"), options);
2646 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2647 assert!(result.is_err(), "the stand-in never reports itself");
2648
2649 let path = formation_dir.join(FORMATION_MARKER_FILE);
2650 let metadata = std::fs::metadata(&path).unwrap();
2651 assert_eq!(metadata.permissions().mode() & 0o777, 0o600);
2652 let marker = FormationMarker::read(&formation_dir).unwrap();
2653 let pid = std::process::id();
2654 assert_eq!(
2655 marker,
2656 FormationMarker {
2657 project_dir: std::fs::canonicalize(project.path()).unwrap(),
2658 launcher_pid: pid,
2659 launcher_start: crate::running::process_start_token(pid).unwrap_or_default(),
2660 }
2661 );
2662 let peer_dir = std::fs::metadata(formation_dir.join("coder")).unwrap();
2663 assert!(
2664 metadata.modified().unwrap() <= peer_dir.modified().unwrap(),
2665 "the marker was written after the peer's directory was made"
2666 );
2667 }
2668
2669 #[test]
2671 fn formation_marker_is_never_written_without_peers() {
2672 let _guard = binary_guard();
2673 let dir = tempfile::tempdir().unwrap();
2674 let options = options(Duration::from_secs(5));
2675 let formations = options.formations_dir.clone().unwrap();
2676 std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2677 let plan = FormationPlan {
2678 peers: Vec::new(),
2679 entry: member("planner"),
2680 callees: Vec::new(),
2681 };
2682 let result = launch_plan(plan, options);
2683 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2684 let formation = result.expect("a formation with no peers has nothing to wait for");
2685 drop(formation);
2686 assert!(
2687 std::fs::read_dir(&formations).map_or(true, |mut entries| entries.next().is_none()),
2688 "{}",
2689 formations.display()
2690 );
2691 }
2692
2693 fn marker_of(pid: u32, start: &str) -> FormationMarker {
2694 FormationMarker {
2695 project_dir: PathBuf::from("/srv/project"),
2696 launcher_pid: pid,
2697 launcher_start: start.to_string(),
2698 }
2699 }
2700
2701 fn reaped_pid() -> u32 {
2703 let mut child = std::process::Command::new("true").spawn().unwrap();
2704 let pid = child.id();
2705 child.wait().unwrap();
2706 pid
2707 }
2708
2709 fn running_record(pid: u32, formation: Option<&FormationId>) -> crate::running::RunningRecord {
2710 crate::running::RunningRecord {
2711 session_id: "ses_0199c4e2f1b7712a9d3e4f5061728394".to_string(),
2712 url: "127.0.0.1:1".to_string(),
2713 pid,
2714 process_start: crate::running::process_start_token(pid).unwrap_or_default(),
2715 capsule_name: "worker".to_string(),
2716 capsule_version: "0.1.0".to_string(),
2717 workdir: PathBuf::from("/tmp/worker"),
2718 outlives_launcher: false,
2719 started_at: "2026-01-01T00:00:00Z".to_string(),
2720 door_token: None,
2721 credentials: Default::default(),
2722 formation_id: formation.cloned(),
2723 formation_lifeline: false,
2724 formation_launcher: None,
2725 spawned_by: None,
2726 }
2727 }
2728
2729 #[test]
2732 fn prune_formations_liveness_reads_the_launcher_from_the_marker() {
2733 let id = FormationId::mint();
2734 let own = std::process::id();
2735 let token = crate::running::process_start_token(own).unwrap();
2736 assert!(formation_is_live(&id, &marker_of(own, &token), Some(&[])));
2737 assert!(!formation_is_live(
2738 &id,
2739 &marker_of(reaped_pid(), &token),
2740 Some(&[])
2741 ));
2742 }
2743
2744 #[test]
2748 fn prune_formations_liveness_reads_running_records_that_name_the_formation() {
2749 let id = FormationId::mint();
2750 let dead = marker_of(reaped_pid(), "1");
2751 let mut sleeper = std::process::Command::new("sleep")
2752 .arg("30")
2753 .spawn()
2754 .unwrap();
2755 let live = running_record(sleeper.id(), Some(&id));
2756 let other = running_record(sleeper.id(), Some(&FormationId::mint()));
2757 let unnamed = running_record(sleeper.id(), None);
2758 let gone = running_record(reaped_pid(), Some(&id));
2759
2760 assert!(formation_is_live(&id, &dead, Some(&[live])));
2761 assert!(!formation_is_live(
2762 &id,
2763 &dead,
2764 Some(&[other, unnamed, gone])
2765 ));
2766 let _ = sleeper.kill();
2767 let _ = sleeper.wait();
2768 }
2769
2770 #[test]
2773 fn prune_formations_liveness_treats_an_unlistable_running_directory_as_all_live() {
2774 let id = FormationId::mint();
2775 assert!(formation_is_live(&id, &marker_of(reaped_pid(), "1"), None));
2776
2777 let home = tempfile::tempdir().unwrap();
2778 let project = tempfile::tempdir().unwrap();
2779 let formations = home.path().join(FORMATIONS_DIR);
2780 let old = FormationId::parse(&format!("frm_{:012x}{:020x}", 1_000_000u64, 1)).unwrap();
2781 let old_dir = formations.join(old.as_str());
2782 std::fs::create_dir_all(old_dir.join("worker")).unwrap();
2783 let marker = FormationMarker {
2784 project_dir: std::fs::canonicalize(project.path()).unwrap(),
2785 ..marker_of(reaped_pid(), "1")
2786 };
2787 std::fs::write(
2788 old_dir.join(FORMATION_MARKER_FILE),
2789 serde_json::to_string(&marker).unwrap(),
2790 )
2791 .unwrap();
2792 let policy = murmur_artifact::TraceRetainConfig {
2793 max_sessions: Some(1),
2794 max_age_secs: Some(1),
2795 };
2796
2797 let running = home.path().join("running");
2799 std::fs::write(&running, "").unwrap();
2800 let current = FormationId::mint();
2801 assert!(prune_ended_formations_in(
2802 &formations,
2803 &running,
2804 ¤t,
2805 project.path(),
2806 &policy
2807 )
2808 .is_empty());
2809 assert!(old_dir.is_dir());
2810
2811 std::fs::remove_file(&running).unwrap();
2813 let pruned =
2814 prune_ended_formations_in(&formations, &running, ¤t, project.path(), &policy);
2815 assert_eq!(pruned.len(), 1, "{pruned:?}");
2816 assert_eq!(pruned[0].formation_id, old);
2817 assert_eq!(pruned[0].reason, crate::trace::RETENTION_REASON_MAX_AGE);
2818 assert!(!old_dir.exists());
2819 }
2820}