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(
910 formations_dir: &Path,
911 formation_id: &FormationId,
912) -> Vec<PathBuf> {
913 let dir = formations_dir.join(formation_id.as_str());
914 let entries = match std::fs::read_dir(&dir) {
915 Ok(entries) => entries,
916 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Vec::new(),
917 Err(_) => return vec![dir],
918 };
919 let mut roots: Vec<PathBuf> = entries
920 .filter_map(Result::ok)
921 .filter(|entry| entry.file_type().is_ok_and(|kind| kind.is_dir()))
922 .filter_map(|entry| crate::records_dir::records_group(&entry.path()).ok())
923 .filter(|root| root.exists())
924 .collect();
925 roots.sort();
926 roots
927}
928
929#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
935pub struct FormationMarker {
936 pub project_dir: PathBuf,
938 pub launcher_pid: u32,
940 pub launcher_start: String,
943}
944
945impl FormationMarker {
946 pub fn for_this_launcher(project_dir: &Path) -> Self {
948 let pid = std::process::id();
949 Self {
950 project_dir: canonical_project_dir(project_dir),
951 launcher_pid: pid,
952 launcher_start: crate::running::process_start_token(pid).unwrap_or_default(),
953 }
954 }
955
956 pub fn read(formation_dir: &Path) -> Option<Self> {
958 let raw = std::fs::read_to_string(formation_dir.join(FORMATION_MARKER_FILE)).ok()?;
959 serde_json::from_str(&raw).ok()
960 }
961}
962
963pub(crate) fn canonical_project_dir(project_dir: &Path) -> PathBuf {
966 std::fs::canonicalize(project_dir).unwrap_or_else(|_| project_dir.to_path_buf())
967}
968
969fn write_formation_marker(options: &FormationLaunchOptions) -> Result<(), String> {
972 let formations_dir = launch_formations_dir(options)?;
973 let formation_dir = formations_dir.join(options.formation_id.as_str());
974 for dir in [formations_dir.as_path(), formation_dir.as_path()] {
975 crate::state_store::ensure_private_dir(dir, crate::murmur_home::MURMUR_HOME_DIR_MODE)
976 .map_err(|reason| format!("{}: {reason}", dir.display()))?;
977 }
978 let marker = FormationMarker::for_this_launcher(&options.project_dir);
979 let json = serde_json::to_string(&marker).map_err(|error| error.to_string())?;
980 crate::murmur_home::write_private_file(
981 &formation_dir.join(FORMATION_MARKER_FILE),
982 format!("{json}\n").as_bytes(),
983 )
984}
985
986pub fn prune_ended_formations(
993 current: &FormationId,
994 project_dir: &Path,
995 policy: &murmur_artifact::TraceRetainConfig,
996) -> Vec<crate::retention::PrunedFormation> {
997 let (Ok(formations_dir), Ok(running_dir)) =
998 (formations_dir(), crate::running::running_dir_location())
999 else {
1000 return Vec::new();
1001 };
1002 prune_ended_formations_in(&formations_dir, &running_dir, current, project_dir, policy)
1003}
1004
1005pub fn prune_ended_formations_in(
1008 formations_dir: &Path,
1009 running_dir: &Path,
1010 current: &FormationId,
1011 project_dir: &Path,
1012 policy: &murmur_artifact::TraceRetainConfig,
1013) -> Vec<crate::retention::PrunedFormation> {
1014 let records = crate::running::scan(running_dir)
1016 .ok()
1017 .map(|scan| scan.records);
1018 crate::retention::prune_formations(
1019 formations_dir,
1020 current,
1021 project_dir,
1022 policy,
1023 crate::retention::now_ms(),
1024 |id, marker| formation_is_live(id, marker, records.as_deref()),
1025 )
1026}
1027
1028fn formation_is_live(
1035 id: &FormationId,
1036 marker: &FormationMarker,
1037 records: Option<&[crate::running::RunningRecord]>,
1038) -> bool {
1039 let not_gone = |state| !matches!(state, crate::running::ProcessState::Gone(_));
1040 let Some(records) = records else {
1041 return true;
1042 };
1043 not_gone(crate::running::process_state_of(
1044 marker.launcher_pid,
1045 &marker.launcher_start,
1046 )) || records.iter().any(|record| {
1047 record.formation_id.as_ref() == Some(id) && not_gone(crate::running::process_state(record))
1048 })
1049}
1050
1051fn make_member_workdir(
1058 formations_dir: &Path,
1059 formation_id: &FormationId,
1060 member: &str,
1061) -> Result<PathBuf, String> {
1062 use std::os::unix::fs::DirBuilderExt;
1063
1064 let formation_dir = formations_dir.join(formation_id.as_str());
1065 let workdir = formation_dir.join(member);
1066 let failed = |path: &Path, reason: String| {
1067 format!(
1068 "its directory {} could not be made: {}: {reason}",
1069 workdir.display(),
1070 path.display()
1071 )
1072 };
1073 for dir in [formations_dir, formation_dir.as_path()] {
1074 crate::state_store::ensure_private_dir(dir, crate::murmur_home::MURMUR_HOME_DIR_MODE)
1075 .map_err(|reason| failed(dir, reason))?;
1076 }
1077 std::fs::DirBuilder::new()
1078 .mode(crate::murmur_home::MURMUR_HOME_DIR_MODE)
1079 .create(&workdir)
1080 .map_err(|error| failed(&workdir, error.to_string()))?;
1081 Ok(workdir)
1083}
1084
1085fn launch_formations_dir(options: &FormationLaunchOptions) -> Result<PathBuf, String> {
1088 match &options.formations_dir {
1089 Some(dir) => Ok(dir.clone()),
1090 None => {
1091 crate::murmur_home::ensure_murmur_home()?;
1092 formations_dir()
1093 }
1094 }
1095}
1096
1097enum PeerOutcome {
1099 Ready(ReadyPeer),
1100 Failed(MemberFailureReason),
1101 Abandoned,
1103}
1104
1105struct Channel {
1108 writer: std::io::PipeWriter,
1109 reader: std::io::PipeReader,
1110}
1111
1112impl Channel {
1113 fn open(first_line: &str) -> Result<Self, MemberFailureReason> {
1116 let (reader, mut writer) = std::io::pipe().map_err(|error| {
1117 not_started(&format!(
1118 "its formation channel could not be created: {error}"
1119 ))
1120 })?;
1121 writer
1122 .write_all(format!("{first_line}\n").as_bytes())
1123 .map_err(|error| {
1124 not_started(&format!(
1125 "its formation channel could not be written: {error}"
1126 ))
1127 })?;
1128 Ok(Self { writer, reader })
1129 }
1130
1131 fn fd(&self) -> i32 {
1133 use std::os::fd::AsRawFd;
1134 self.reader.as_raw_fd()
1135 }
1136
1137 fn member_env(&self, formation_id: &FormationId) -> Vec<(String, String)> {
1141 let (id_name, id_value) = formation_id.env_pair();
1142 vec![
1143 (id_name.to_string(), id_value),
1144 (FORMATION_CHANNEL_ENV.to_string(), self.fd().to_string()),
1145 crate::host_warnings::reported_env_pair(),
1146 ]
1147 }
1148}
1149
1150fn start_peer(
1155 binary: &Path,
1156 member: &PlannedMember,
1157 first_line: &str,
1158 options: &FormationLaunchOptions,
1159 abort: &AtomicBool,
1160 #[cfg(unix)] mut lifeline: crate::lifeline::MemberLifeline,
1161) -> (Option<MemberProcess>, PeerOutcome) {
1162 let workdir = match launch_formations_dir(options)
1163 .and_then(|dir| make_member_workdir(&dir, &options.formation_id, &member.name))
1164 {
1165 Ok(workdir) => workdir,
1166 Err(error) => return (None, PeerOutcome::Failed(not_started(&error))),
1167 };
1168 let args = peer_args(member, options, &workdir);
1169 let prefix = format!("[{}] ", member.name);
1170 let channel = match Channel::open(first_line) {
1171 Ok(channel) => channel,
1172 Err(reason) => return (None, PeerOutcome::Failed(reason)),
1173 };
1174 let set = channel.member_env(&options.formation_id);
1175 let spawned_at = Instant::now();
1176 let deadline = spawned_at + options.ready_timeout;
1177 let started = child_launch::start_process(&ProcessLaunch {
1178 binary,
1179 args: &args,
1180 cwd: None,
1181 env: ProcessEnv::Inherited {
1182 set: &set,
1183 remove: &[FORMATION_PEERS_ENV, SPAWNER_ENV, SPAWNER_LIFELINE_ENV],
1184 },
1185 stdin_line: None,
1186 stderr_prefix: Some(&prefix),
1187 inherit_output: false,
1188 own_process_group: true,
1189 inherit_fd: Some(channel.fd()),
1190 #[cfg(unix)]
1191 lifeline: Some(crate::child_launch::Lifeline::Member(&lifeline)),
1192 });
1193 let Channel { writer, reader } = channel;
1195 drop(reader);
1196 let StartedProcess {
1197 mut child,
1198 stderr_tail,
1199 ..
1200 } = match started {
1201 Ok(started) => started,
1202 Err(error) => {
1203 return (
1204 None,
1205 PeerOutcome::Failed(MemberFailureReason::NotStarted {
1206 error: format!("{}: {error}", binary.display()),
1207 }),
1208 )
1209 }
1210 };
1211 #[cfg(unix)]
1212 lifeline.spawned();
1213 let stdout = child.stdout.take();
1214 let mut process = MemberProcess::new(
1215 &member.name,
1216 child,
1217 true,
1218 options.stop_grace,
1219 writer,
1220 #[cfg(unix)]
1221 Some(lifeline),
1222 );
1223 let Some(stdout) = stdout else {
1224 return (
1225 Some(process),
1226 PeerOutcome::Failed(MemberFailureReason::NotStarted {
1227 error: "the process exposed no standard output".to_string(),
1228 }),
1229 );
1230 };
1231 let lines = child_launch::relay_readiness_line(stdout, AfterReadiness::PrefixedStderr(prefix));
1232
1233 let line = loop {
1235 if abort.load(Ordering::SeqCst) {
1236 return (Some(process), PeerOutcome::Abandoned);
1237 }
1238 let now = Instant::now();
1239 if now >= deadline {
1240 return (
1241 Some(process),
1242 PeerOutcome::Failed(MemberFailureReason::DidNotReport {
1243 deadline: options.ready_timeout,
1244 }),
1245 );
1246 }
1247 match lines.recv_timeout(POLL_INTERVAL.min(deadline - now)) {
1248 Ok(Some(line)) => break line,
1249 Ok(None) | Err(mpsc::RecvTimeoutError::Disconnected) => {
1250 let closed_at = Instant::now();
1253 let status = loop {
1254 if let Some(status) = process.exit_status_now() {
1255 break status;
1256 }
1257 if closed_at.elapsed() >= EXIT_AFTER_EOF_WAIT {
1258 process.reap();
1259 break process.status.unwrap_or_default();
1260 }
1261 std::thread::sleep(POLL_INTERVAL);
1262 };
1263 return (
1264 Some(process),
1265 PeerOutcome::Failed(MemberFailureReason::ExitedBeforeReporting {
1266 status: describe_status(status),
1267 stderr_tail: stderr_tail.lines_at_end(),
1268 }),
1269 );
1270 }
1271 Err(mpsc::RecvTimeoutError::Timeout) => {}
1272 }
1273 };
1274
1275 let report = match read_report(&line, &options.formation_id) {
1276 Ok(report) => report,
1277 Err(reason) => return (Some(process), PeerOutcome::Failed(reason)),
1278 };
1279
1280 let mut last_error;
1282 loop {
1283 if abort.load(Ordering::SeqCst) {
1284 return (Some(process), PeerOutcome::Abandoned);
1285 }
1286 if let Some(status) = process.exit_status_now() {
1287 return (
1288 Some(process),
1289 PeerOutcome::Failed(MemberFailureReason::ExitedBeforeDoorAnswered {
1290 status: describe_status(status),
1291 stderr_tail: stderr_tail.lines_at_end(),
1292 }),
1293 );
1294 }
1295 match door_answers_as(
1296 crate::running::probe_door_session_id(&report.url, report.door_token.as_ref()),
1297 &report.session_id,
1298 ) {
1299 Ok(()) => {
1300 let pid = process.pid;
1301 return (
1302 Some(process),
1303 PeerOutcome::Ready(ReadyPeer {
1304 name: member.name.clone(),
1305 capsule: member.capsule.clone(),
1306 version: member.version.clone(),
1307 pid,
1308 session_id: report.session_id,
1309 url: report.url,
1310 workdir,
1311 door_token: report.door_token,
1312 }),
1313 );
1314 }
1315 Err(why) => last_error = why,
1316 }
1317 let now = Instant::now();
1318 if now >= deadline {
1319 return (
1320 Some(process),
1321 PeerOutcome::Failed(MemberFailureReason::DoorDidNotAnswer {
1322 url: report.url,
1323 session_id: report.session_id,
1324 deadline: options.ready_timeout,
1325 last_error,
1326 }),
1327 );
1328 }
1329 std::thread::sleep(READY_PROBE_INTERVAL.min(deadline - now));
1330 }
1331}
1332
1333fn door_answers_as(probed: Result<String, String>, reported: &str) -> Result<(), String> {
1337 match probed {
1338 Ok(session_id) if session_id == reported => Ok(()),
1339 Ok(other) => Err(format!("the door answers as session {other}")),
1340 Err(error) => Err(error),
1341 }
1342}
1343
1344struct Report {
1346 session_id: String,
1347 url: String,
1349 door_token: Option<DoorToken>,
1350}
1351
1352fn read_report(line: &str, expected: &FormationId) -> Result<Report, MemberFailureReason> {
1354 let report: serde_json::Value =
1355 serde_json::from_str(line).map_err(|error| MemberFailureReason::UnreadableReport {
1356 why: format!("it does not parse as JSON ({error})"),
1357 })?;
1358 let reported = report
1359 .get("formation_id")
1360 .and_then(serde_json::Value::as_str);
1361 if reported != Some(expected.as_str()) {
1362 return Err(MemberFailureReason::WrongFormation {
1363 reported: reported.map(str::to_string),
1364 expected: expected.as_str().to_string(),
1365 });
1366 }
1367 let session_id = report
1368 .get("session_id")
1369 .and_then(serde_json::Value::as_str)
1370 .filter(|session_id| !session_id.is_empty())
1371 .ok_or_else(|| MemberFailureReason::UnreadableReport {
1372 why: "it names no session_id".to_string(),
1373 })?;
1374 let url = report
1375 .get("url")
1376 .and_then(serde_json::Value::as_str)
1377 .unwrap_or_default();
1378 if url.is_empty() {
1379 return Err(MemberFailureReason::ReportedNoDoor);
1380 }
1381 Ok(Report {
1382 session_id: session_id.to_string(),
1383 url: format!("http://{}", url.trim_start_matches("http://")),
1384 door_token: child_launch::readiness_door_token(&report),
1385 })
1386}
1387
1388pub fn launch_formation(
1396 roster: &AdmittedRoster,
1397 options: FormationLaunchOptions,
1398) -> Result<RunningFormation, MemberLaunchFailure> {
1399 launch_plan(FormationPlan::from_roster(roster), options)
1400}
1401
1402pub(crate) fn launch_plan(
1403 plan: FormationPlan,
1404 options: FormationLaunchOptions,
1405) -> Result<RunningFormation, MemberLaunchFailure> {
1406 install_panic_hook();
1407 let binary = match child_launch::mur_binary() {
1408 Ok(binary) => binary,
1409 Err(error) => {
1410 return Err(MemberLaunchFailure {
1411 failed: plan
1412 .peers
1413 .iter()
1414 .chain([&plan.entry])
1415 .map(|member| failure(member, not_started(&error.to_string())))
1416 .collect(),
1417 stopped: Vec::new(),
1418 interrupted_by: None,
1419 })
1420 }
1421 };
1422
1423 let authority = match FormationAuthority::generate(&options.formation_id) {
1424 Ok(authority) => authority,
1425 Err(error) => {
1426 return Err(MemberLaunchFailure {
1427 failed: plan
1428 .peers
1429 .iter()
1430 .chain([&plan.entry])
1431 .map(|member| failure(member, not_started(&error)))
1432 .collect(),
1433 stopped: Vec::new(),
1434 interrupted_by: None,
1435 })
1436 }
1437 };
1438 #[cfg(unix)]
1442 let (peer_lifelines, entry_lifeline) = match member_lifelines(plan.peers.len() + 1) {
1443 Ok(mut lifelines) => {
1444 let entry = lifelines.pop();
1445 (lifelines, entry)
1446 }
1447 Err(error) => {
1448 return Err(MemberLaunchFailure {
1449 failed: plan
1450 .peers
1451 .iter()
1452 .chain([&plan.entry])
1453 .map(|member| {
1454 failure(
1455 member,
1456 not_started(&format!("its lifeline could not be created: {error}")),
1457 )
1458 })
1459 .collect(),
1460 stopped: Vec::new(),
1461 interrupted_by: None,
1462 })
1463 }
1464 };
1465 let launcher = ProcessIdentity::of_this_process();
1469 let first_lines: Vec<String> = plan
1470 .peers
1471 .iter()
1472 .map(|member| plan.first_line(&authority, launcher.as_ref(), &member.name))
1473 .collect();
1474 if !plan.peers.is_empty() {
1477 if let Err(reason) = write_formation_marker(&options) {
1478 crate::runtime_err!(
1479 "[mur run] warning: formation {} has no ownership marker ({reason}); its \
1480 directory will never be removed automatically",
1481 options.formation_id
1482 );
1483 }
1484 }
1485
1486 let abort = AtomicBool::new(false);
1487 let mut outcomes: Vec<Option<(Option<MemberProcess>, PeerOutcome)>> =
1488 plan.peers.iter().map(|_| None).collect();
1489 let mut interrupted_by = caught_signal();
1490 if interrupted_by.is_some() {
1491 abort.store(true, Ordering::SeqCst);
1492 }
1493 std::thread::scope(|scope| {
1494 let (tx, rx) = mpsc::channel();
1495 #[cfg(unix)]
1496 let mut peer_lifelines = peer_lifelines.into_iter();
1497 for (index, member) in plan.peers.iter().enumerate() {
1498 let tx = tx.clone();
1499 let (binary, options, abort) = (&binary, &options, &abort);
1500 let first_line = first_lines[index].as_str();
1501 #[cfg(unix)]
1502 let lifeline = peer_lifelines
1503 .next()
1504 .expect("member_lifelines made one lifeline per peer");
1505 scope.spawn(move || {
1506 let started = start_peer(
1507 binary,
1508 member,
1509 first_line,
1510 options,
1511 abort,
1512 #[cfg(unix)]
1513 lifeline,
1514 );
1515 let _ = tx.send((index, started));
1516 });
1517 }
1518 drop(tx);
1519 loop {
1520 match rx.recv_timeout(POLL_INTERVAL) {
1521 Ok((index, outcome)) => {
1522 if matches!(outcome.1, PeerOutcome::Failed(_)) {
1523 abort.store(true, Ordering::SeqCst);
1524 }
1525 outcomes[index] = Some(outcome);
1526 }
1527 Err(mpsc::RecvTimeoutError::Timeout) => {}
1528 Err(mpsc::RecvTimeoutError::Disconnected) => break,
1529 }
1530 if interrupted_by.is_none() {
1531 if let Some(signal) = caught_signal() {
1532 interrupted_by = Some(signal);
1533 abort.store(true, Ordering::SeqCst);
1534 }
1535 }
1536 }
1537 });
1538
1539 let mut members = MemberSet(Vec::new());
1540 let mut peers = Vec::new();
1541 let mut failed = Vec::new();
1542 for (member, outcome) in plan.peers.iter().zip(outcomes) {
1543 let Some((process, outcome)) = outcome else {
1544 continue;
1545 };
1546 members.0.extend(process);
1547 match outcome {
1548 PeerOutcome::Ready(peer) => peers.push(peer),
1549 PeerOutcome::Failed(reason) => failed.push(failure(member, reason)),
1550 PeerOutcome::Abandoned => {}
1551 }
1552 }
1553
1554 if !failed.is_empty() || interrupted_by.is_some() || peers.len() != plan.peers.len() {
1555 let stopped = members.stop_all();
1556 return Err(MemberLaunchFailure {
1557 failed,
1558 stopped,
1559 interrupted_by,
1560 });
1561 }
1562
1563 for member in &mut members.0 {
1566 let addresses = addresses_of(&peers, plan.callees_of(&member.name));
1567 if !addresses.is_empty() {
1568 member.send_line(&render_address_line(&addresses));
1569 }
1570 }
1571
1572 Ok(RunningFormation {
1573 formation_id: options.formation_id.clone(),
1574 entry: plan.entry.clone(),
1575 plan,
1576 peers,
1577 members,
1578 entry_index: None,
1579 #[cfg(unix)]
1580 entry_lifeline,
1581 binary,
1582 options,
1583 authority,
1584 launcher,
1585 })
1586}
1587
1588fn addresses_of(ready: &[ReadyPeer], callees: &[String]) -> Vec<FormationPeer> {
1590 ready
1591 .iter()
1592 .filter(|peer| callees.contains(&peer.name))
1593 .map(|peer| FormationPeer {
1594 name: peer.name.clone(),
1595 url: peer.url.clone(),
1596 })
1597 .collect()
1598}
1599
1600#[cfg(unix)]
1602fn member_lifelines(count: usize) -> std::io::Result<Vec<crate::lifeline::MemberLifeline>> {
1603 (0..count)
1604 .map(|_| crate::lifeline::MemberLifeline::new())
1605 .collect()
1606}
1607
1608fn failure(member: &PlannedMember, reason: MemberFailureReason) -> MemberFailure {
1609 MemberFailure {
1610 name: member.name.clone(),
1611 capsule: member.capsule.clone(),
1612 version: member.version.clone(),
1613 reason,
1614 }
1615}
1616
1617fn not_started(error: &str) -> MemberFailureReason {
1618 MemberFailureReason::NotStarted {
1619 error: error.to_string(),
1620 }
1621}
1622
1623pub struct RunningFormation {
1629 formation_id: FormationId,
1630 entry: PlannedMember,
1631 plan: FormationPlan,
1632 peers: Vec<ReadyPeer>,
1633 members: MemberSet,
1636 entry_index: Option<usize>,
1638 #[cfg(unix)]
1641 entry_lifeline: Option<crate::lifeline::MemberLifeline>,
1642 binary: PathBuf,
1643 options: FormationLaunchOptions,
1644 authority: FormationAuthority,
1645 launcher: Option<ProcessIdentity>,
1647}
1648
1649impl std::fmt::Debug for RunningFormation {
1650 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1651 f.debug_struct("RunningFormation")
1652 .field("formation_id", &self.formation_id)
1653 .field("entry", &self.entry.name)
1654 .field("peers", &self.peers)
1655 .finish()
1656 }
1657}
1658
1659impl RunningFormation {
1660 pub fn formation_id(&self) -> &FormationId {
1661 &self.formation_id
1662 }
1663
1664 pub fn entry_name(&self) -> &str {
1666 &self.entry.name
1667 }
1668
1669 pub fn peers(&self) -> &[ReadyPeer] {
1671 &self.peers
1672 }
1673
1674 pub fn callees_of(&self, member: &str) -> &[String] {
1677 self.plan.callees_of(member)
1678 }
1679
1680 pub fn entry_pid(&self) -> Option<u32> {
1682 self.entry_index.map(|index| self.members.0[index].pid)
1683 }
1684
1685 pub fn start_entry(&mut self) -> Result<u32, MemberLaunchFailure> {
1689 let args = entry_args(&self.entry, &self.options);
1690 let channel = Channel::open(&self.plan.first_line(
1692 &self.authority,
1693 self.launcher.as_ref(),
1694 &self.entry.name,
1695 ))
1696 .and_then(|mut channel| {
1697 let addresses = addresses_of(&self.peers, self.plan.callees_of(&self.entry.name));
1698 if !addresses.is_empty() {
1699 channel
1700 .writer
1701 .write_all(format!("{}\n", render_address_line(&addresses)).as_bytes())
1702 .map_err(|error| {
1703 not_started(&format!(
1704 "its formation channel could not be written: {error}"
1705 ))
1706 })?;
1707 }
1708 Ok(channel)
1709 });
1710 let channel = match channel {
1711 Ok(channel) => channel,
1712 Err(reason) => {
1713 let stopped = self.members.stop_all();
1714 return Err(MemberLaunchFailure {
1715 failed: vec![failure(&self.entry, reason)],
1716 stopped,
1717 interrupted_by: None,
1718 });
1719 }
1720 };
1721 let set = channel.member_env(&self.formation_id);
1722 let started = child_launch::start_process(&ProcessLaunch {
1723 binary: &self.binary,
1724 args: &args,
1725 cwd: None,
1726 env: ProcessEnv::Inherited {
1727 set: &set,
1728 remove: &[SPAWNER_ENV, FORMATION_PEERS_ENV, SPAWNER_LIFELINE_ENV],
1729 },
1730 stdin_line: None,
1731 stderr_prefix: None,
1732 inherit_output: true,
1733 own_process_group: false,
1734 inherit_fd: Some(channel.fd()),
1735 #[cfg(unix)]
1736 lifeline: self
1737 .entry_lifeline
1738 .as_ref()
1739 .map(crate::child_launch::Lifeline::Member),
1740 });
1741 let Channel { writer, reader } = channel;
1742 drop(reader);
1743 match started {
1744 Ok(started) => {
1745 #[cfg(unix)]
1746 let lifeline = self.entry_lifeline.take().map(|mut lifeline| {
1747 lifeline.spawned();
1748 lifeline
1749 });
1750 let process = MemberProcess::new(
1751 &self.entry.name,
1752 started.child,
1753 false,
1754 self.options.stop_grace,
1755 writer,
1756 #[cfg(unix)]
1757 lifeline,
1758 );
1759 let pid = process.pid;
1760 self.members.0.push(process);
1761 self.entry_index = Some(self.members.0.len() - 1);
1762 Ok(pid)
1763 }
1764 Err(error) => {
1765 let stopped = self.members.stop_all();
1766 Err(MemberLaunchFailure {
1767 failed: vec![failure(
1768 &self.entry,
1769 not_started(&format!("{}: {error}", self.binary.display())),
1770 )],
1771 stopped,
1772 interrupted_by: None,
1773 })
1774 }
1775 }
1776 }
1777
1778 pub fn wait(mut self) -> FormationExit {
1781 let ending = match self.entry_index {
1782 None => caught_signal()
1783 .map(FormationEnding::Signalled)
1784 .unwrap_or(FormationEnding::EntryExited(ExitStatus::default())),
1785 Some(index) => loop {
1786 if let Some(status) = self.members.0[index].exit_status_now() {
1787 break FormationEnding::EntryExited(status);
1788 }
1789 if let Some(signal) = caught_signal() {
1790 break FormationEnding::Signalled(signal);
1791 }
1792 std::thread::sleep(POLL_INTERVAL);
1793 },
1794 };
1795 let stopped = self.members.stop_all();
1796 FormationExit { ending, stopped }
1797 }
1798}
1799
1800#[cfg(test)]
1801mod tests {
1802 use super::*;
1803 use std::io::{Read, Write};
1804 use std::net::TcpListener;
1805 use std::sync::atomic::AtomicUsize;
1806 use std::sync::Arc;
1807
1808 const LIVENESS_BOUND: Duration = Duration::from_secs(120);
1812
1813 fn load_average() -> String {
1815 std::fs::read_to_string("/proc/loadavg")
1816 .unwrap_or_default()
1817 .split(' ')
1818 .take(3)
1819 .collect::<Vec<_>>()
1820 .join(" ")
1821 }
1822
1823 #[test]
1826 fn the_production_bounds_are_180s_and_a_grace_past_the_runtime_teardown() {
1827 assert_eq!(MEMBER_READY_TIMEOUT, Duration::from_secs(180));
1828 assert_eq!(MEMBER_STOP_GRACE, Duration::from_secs(25));
1829 assert!(MEMBER_STOP_GRACE > crate::runtime::TERMINATE_TEARDOWN_DEADLINE);
1830 }
1831
1832 fn member(name: &str) -> PlannedMember {
1833 PlannedMember {
1834 name: name.to_string(),
1835 capsule: name.to_string(),
1836 version: "0.1.0".to_string(),
1837 sha256: "ab".repeat(32),
1838 }
1839 }
1840
1841 #[test]
1844 fn every_member_is_told_host_warnings_were_reported() {
1845 let channel = Channel::open("first").expect("a channel opens");
1846 let env = channel.member_env(&FormationId::mint());
1847 let values: Vec<&str> = env
1848 .iter()
1849 .filter(|(key, _)| key == crate::host_warnings::HOST_WARNINGS_REPORTED_ENV)
1850 .map(|(_, value)| value.as_str())
1851 .collect();
1852 assert_eq!(values, vec!["1"], "{env:?}");
1853 }
1854
1855 fn options(deadline: Duration) -> FormationLaunchOptions {
1856 let mut options =
1857 FormationLaunchOptions::new(FormationId::mint(), PathBuf::from("/tmp/project"));
1858 options.ready_timeout = deadline;
1859 options.stop_grace = Duration::from_secs(2);
1860 options.formations_dir = Some(tempfile::tempdir().unwrap().keep());
1862 options
1863 }
1864
1865 #[test]
1866 fn member_argv_is_exactly_the_documented_shape() {
1867 let mut options = options(MEMBER_READY_TIMEOUT);
1868 let coder = member("coder");
1869 let workdir = PathBuf::from("/home/u/.murmur/formations/frm_x/coder");
1870 assert_eq!(
1871 peer_args(&coder, &options, &workdir),
1872 [
1873 "run",
1874 "--capsule",
1875 "coder",
1876 "--capsule-version",
1877 "0.1.0",
1878 "--capsule-sha256",
1879 &"ab".repeat(32),
1880 "--workdir",
1881 "/home/u/.murmur/formations/frm_x/coder",
1882 "--store-root",
1883 "/tmp/project",
1884 "--json",
1885 "--readiness-tokens",
1886 "--bind",
1887 "127.0.0.1",
1888 ]
1889 );
1890 options.task = Some("probe".to_string());
1891 options.json = true;
1892 options.verbose = true;
1893 options.no_env_file = true;
1894 options.containment = Some("scoped".to_string());
1895 let entry = entry_args(&member("planner"), &options);
1896 assert_eq!(&entry[7..9], ["--workdir", "/tmp/project"]);
1897 assert_eq!(
1898 &entry[9..],
1899 [
1900 "--lifecycle-task-acceptance",
1901 "single",
1902 "--lifecycle-after-task",
1903 "exit",
1904 "--task",
1905 "probe",
1906 "--json",
1907 "--verbose",
1908 "--no-env-file",
1909 "--containment",
1910 "scoped",
1911 ]
1912 );
1913 assert!(
1914 !entry.iter().any(|arg| arg == "--bind"
1915 || arg == STORE_ROOT_FLAG
1916 || arg == child_launch::READINESS_TOKENS_FLAG),
1917 "{entry:?}"
1918 );
1919 let peer = peer_args(&coder, &options, &workdir);
1920 assert!(peer.ends_with(&[
1921 "--no-env-file".to_string(),
1922 "--containment".to_string(),
1923 "scoped".to_string()
1924 ]));
1925 assert!(!peer.iter().any(|arg| arg == "--task" || arg == "--verbose"));
1926 }
1927
1928 fn binary_guard() -> std::sync::MutexGuard<'static, ()> {
1931 crate::formation::FORMATION_ENV_LOCK
1932 .lock()
1933 .unwrap_or_else(|poisoned| poisoned.into_inner())
1934 }
1935
1936 fn stand_in(dir: &Path, line: Option<&str>) -> PathBuf {
1939 let script = dir.join("mur");
1940 let print = match line {
1941 Some(line) => {
1942 std::fs::write(dir.join("line"), format!("{line}\n")).unwrap();
1943 format!("cat '{}'\n", dir.join("line").display())
1944 }
1945 None => String::new(),
1946 };
1947 let body = format!(
1950 "#!/bin/sh\ntrap 'echo TERM >> \"{dir}/sigterm\"' TERM\n{print}\
1951 cat \"/dev/fd/$MURMUR_FORMATION_LIFELINE\" >/dev/null\n",
1952 dir = dir.display()
1953 );
1954 std::fs::write(&script, body).unwrap();
1955 use std::os::unix::fs::PermissionsExt;
1956 std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
1957 script
1958 }
1959
1960 fn plan_of(peer: &str) -> FormationPlan {
1961 FormationPlan {
1962 peers: vec![member(peer)],
1963 entry: member("planner"),
1964 callees: vec![("planner".to_string(), vec![peer.to_string()])],
1965 }
1966 }
1967
1968 fn refused(line: Option<&str>, deadline: Duration) -> (MemberLaunchFailure, FormationId) {
1971 let _guard = binary_guard();
1972 let dir = tempfile::tempdir().unwrap();
1973 let options = options(deadline);
1974 let id = options.formation_id.clone();
1975 let line = line.map(|line| line.replace("{formation}", id.as_str()));
1976 std::env::set_var(
1977 child_launch::MUR_BINARY_ENV,
1978 stand_in(dir.path(), line.as_deref()),
1979 );
1980 let result = launch_plan(plan_of("coder"), options);
1981 std::env::remove_var(child_launch::MUR_BINARY_ENV);
1982 match result {
1983 Ok(_) => panic!("the launch was expected to be refused"),
1984 Err(failure) => (failure, id),
1985 }
1986 }
1987
1988 fn assert_reaped(failure: &MemberLaunchFailure) {
1989 assert_eq!(failure.stopped.len(), 1, "{failure:?}");
1990 let pid = failure.stopped[0].pid;
1991 assert_eq!(failure.stopped[0].name, "coder");
1992 assert!(
1993 !crate::running::pid_is_alive(pid),
1994 "the stand-in {pid} is still running"
1995 );
1996 assert!(
1997 !Path::new(&format!("/proc/{pid}")).exists() || cfg!(not(target_os = "linux")),
1998 "the stand-in {pid} was not reaped"
1999 );
2000 }
2001
2002 fn only_reason(failure: &MemberLaunchFailure) -> &MemberFailureReason {
2003 assert_eq!(failure.failed.len(), 1, "{failure:?}");
2004 assert_eq!(failure.failed[0].name, "coder");
2005 &failure.failed[0].reason
2006 }
2007
2008 fn silent_port() -> u16 {
2010 crate::pinned_port::claim()
2011 }
2012
2013 #[test]
2014 fn a_door_nothing_answers_on_is_refused_at_the_deadline() {
2015 let deadline = Duration::from_secs(10);
2016 let port = silent_port();
2017 let started = Instant::now();
2018 let (failure, _) = refused(
2019 Some(&format!(
2020 r#"{{"formation_id":"{{formation}}","session_id":"ses_standin","url":"localhost:{port}"}}"#
2021 )),
2022 deadline,
2023 );
2024 assert!(started.elapsed() >= deadline);
2025 match only_reason(&failure) {
2026 MemberFailureReason::DoorDidNotAnswer {
2027 url, session_id, ..
2028 } => {
2029 assert_eq!(url, &format!("http://localhost:{port}"));
2030 assert_eq!(session_id, "ses_standin");
2031 }
2032 MemberFailureReason::DidNotReport { .. } => panic!(
2035 "the stand-in's readiness line did not arrive within {deadline:?} (load average {}); \
2036 host load is the likely cause, since the stand-in prints it at once: {failure:?}",
2037 load_average()
2038 ),
2039 other => panic!("{other:?}"),
2040 }
2041 let message = failure.to_string();
2042 assert!(
2043 message.contains("did not answer as session ses_standin"),
2044 "{message}"
2045 );
2046 assert!(message.contains("'coder' (coder@0.1.0)"), "{message}");
2047 assert_reaped(&failure);
2048 }
2049
2050 fn card_door(mut answer: impl FnMut() -> Option<&'static str> + Send + 'static) -> u16 {
2054 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
2055 let port = listener.local_addr().unwrap().port();
2056 std::thread::spawn(move || {
2057 for stream in listener.incoming().flatten() {
2058 let mut stream = stream;
2059 let mut buffer = [0u8; 4096];
2060 let _ = stream.read(&mut buffer);
2061 let body = match answer() {
2062 Some(session_id) => serde_json::json!({
2063 "name": "coder",
2064 "capabilities": {"extensions": [{
2065 "uri": crate::identity::CAPSULE_EXTENSION_URI,
2066 "params": {"sessionId": session_id},
2067 }]},
2068 })
2069 .to_string(),
2070 None => "{}".to_string(),
2071 };
2072 let _ = write!(
2073 stream,
2074 "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\n\
2075 connection: close\r\n\r\n{body}",
2076 body.len()
2077 );
2078 }
2079 });
2080 port
2081 }
2082
2083 fn door(session_id: &'static str, answer_after: Duration) -> u16 {
2086 let opened = Instant::now();
2087 card_door(move || (opened.elapsed() >= answer_after).then_some(session_id))
2088 }
2089
2090 const WRONG_ANSWERS: usize = 3;
2092
2093 fn door_naming_another_session_first() -> (u16, Arc<AtomicUsize>) {
2097 let served = Arc::new(AtomicUsize::new(0));
2098 let counted = Arc::clone(&served);
2099 let port = card_door(move || {
2100 if counted.fetch_add(1, Ordering::SeqCst) < WRONG_ANSWERS {
2101 Some("ses_somebody_else")
2102 } else {
2103 Some("ses_standin")
2104 }
2105 });
2106 (port, served)
2107 }
2108
2109 #[test]
2112 fn a_door_naming_another_session_is_not_ready() {
2113 let _guard = binary_guard();
2114 let (port, served) = door_naming_another_session_first();
2115 let dir = tempfile::tempdir().unwrap();
2116 let options = options(LIVENESS_BOUND);
2117 let line = format!(
2118 r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
2119 options.formation_id
2120 );
2121 std::env::set_var(
2122 child_launch::MUR_BINARY_ENV,
2123 stand_in(dir.path(), Some(&line)),
2124 );
2125 let result = launch_plan(plan_of("coder"), options);
2126 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2127 let formation = match result {
2128 Ok(formation) => formation,
2129 Err(failure) => match only_reason(&failure) {
2130 MemberFailureReason::DidNotReport { .. } => panic!(
2131 "the stand-in's readiness line did not arrive within {LIVENESS_BOUND:?} \
2132 (load average {}); host load is the likely cause: {failure:?}",
2133 load_average()
2134 ),
2135 _ => panic!(
2136 "the launch was refused although the door answered as ses_standin from its \
2137 request {} on: {failure:?}",
2138 WRONG_ANSWERS + 1
2139 ),
2140 },
2141 };
2142 let peer = &formation.peers()[0];
2143 assert_eq!(peer.session_id, "ses_standin");
2144 let served = served.load(Ordering::SeqCst);
2145 assert!(
2146 served > WRONG_ANSWERS,
2147 "an answer naming another session was counted as ready: the door served {served} \
2148 request(s), and its first {WRONG_ANSWERS} named ses_somebody_else"
2149 );
2150 let pid = peer.pid;
2151 drop(formation);
2152 assert!(
2153 !crate::running::pid_is_alive(pid),
2154 "dropping the owner stops the peer"
2155 );
2156 }
2157
2158 #[test]
2159 fn a_door_counts_as_ready_only_when_it_answers_as_the_reported_session() {
2160 let port = door("ses_somebody_else", Duration::ZERO);
2161 let probed = crate::running::probe_door_session_id(&format!("localhost:{port}"), None);
2162 match door_answers_as(probed, "ses_standin") {
2163 Err(why) => assert!(
2164 why.contains("the door answers as session ses_somebody_else"),
2165 "{why}"
2166 ),
2167 Ok(()) => panic!("a door answering as ses_somebody_else counted as ses_standin"),
2168 }
2169 assert_eq!(
2170 door_answers_as(Ok("ses_standin".to_string()), "ses_standin"),
2171 Ok(())
2172 );
2173 assert_eq!(
2174 door_answers_as(Err("connection refused".to_string()), "ses_standin"),
2175 Err("connection refused".to_string())
2176 );
2177 }
2178
2179 #[test]
2180 fn an_empty_url_is_refused_as_no_door() {
2181 let (failure, _) = refused(
2182 Some(r#"{"formation_id":"{formation}","session_id":"ses_standin","url":""}"#),
2183 Duration::from_secs(10),
2184 );
2185 assert_eq!(only_reason(&failure), &MemberFailureReason::ReportedNoDoor);
2186 assert!(failure.to_string().contains("it reported no door"));
2187 assert_reaped(&failure);
2188 }
2189
2190 #[test]
2191 fn a_wrong_formation_or_none_is_refused_naming_what_was_reported() {
2192 let other = FormationId::mint();
2193 let (failure, id) = refused(
2194 Some(&format!(
2195 r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:1"}}"#,
2196 other.as_str()
2197 )),
2198 Duration::from_secs(10),
2199 );
2200 assert_eq!(
2201 only_reason(&failure),
2202 &MemberFailureReason::WrongFormation {
2203 reported: Some(other.as_str().to_string()),
2204 expected: id.as_str().to_string(),
2205 }
2206 );
2207 assert!(failure
2208 .to_string()
2209 .contains(&format!("it reported formation {other} instead of {id}")));
2210 assert_reaped(&failure);
2211
2212 let (failure, id) = refused(
2213 Some(r#"{"session_id":"ses_standin","url":"localhost:1"}"#),
2214 Duration::from_secs(10),
2215 );
2216 assert!(failure
2217 .to_string()
2218 .contains(&format!("it reported no formation instead of {id}")));
2219 assert_reaped(&failure);
2220 }
2221
2222 #[test]
2223 fn no_line_at_all_is_refused_at_the_deadline() {
2224 let started = Instant::now();
2225 let (failure, _) = refused(None, Duration::from_millis(600));
2226 assert!(started.elapsed() >= Duration::from_millis(600));
2227 assert_eq!(
2228 only_reason(&failure),
2229 &MemberFailureReason::DidNotReport {
2230 deadline: Duration::from_millis(600)
2231 }
2232 );
2233 assert_reaped(&failure);
2234 }
2235
2236 #[test]
2237 fn a_door_that_starts_answering_before_the_deadline_is_ready() {
2238 let _guard = binary_guard();
2239 let port = door("ses_standin", Duration::from_millis(400));
2240 let dir = tempfile::tempdir().unwrap();
2241 let options = options(Duration::from_secs(20));
2242 let line = format!(
2243 r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
2244 options.formation_id
2245 );
2246 std::env::set_var(
2247 child_launch::MUR_BINARY_ENV,
2248 stand_in(dir.path(), Some(&line)),
2249 );
2250 let result = launch_plan(plan_of("coder"), options);
2251 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2252 let formation = result.expect("the door answered before the deadline");
2253 assert_eq!(formation.peers().len(), 1);
2254 let peer = &formation.peers()[0];
2255 assert_eq!(peer.session_id, "ses_standin");
2256 assert_eq!(peer.url, format!("http://localhost:{port}"));
2257 assert_eq!(formation.callees_of("planner"), ["coder"]);
2258 assert!(formation.callees_of("coder").is_empty());
2259 let pid = peer.pid;
2260 assert!(crate::running::pid_is_alive(pid));
2261 let dropped = Instant::now();
2262 drop(formation);
2263 assert!(
2264 !crate::running::pid_is_alive(pid),
2265 "dropping the owner stops the peer"
2266 );
2267 assert!(
2268 dropped.elapsed() < Duration::from_secs(2),
2269 "the peer ended on its lifeline's EOF, inside the grace: {:?}",
2270 dropped.elapsed()
2271 );
2272 assert!(
2273 !dir.path().join("sigterm").exists(),
2274 "the launcher sent the peer SIGTERM"
2275 );
2276 }
2277
2278 #[test]
2280 fn a_member_deaf_to_its_lifeline_is_killed_after_the_grace() {
2281 let _guard = binary_guard();
2282 let port = door("ses_standin", Duration::ZERO);
2283 let dir = tempfile::tempdir().unwrap();
2284 let mut options = options(Duration::from_secs(20));
2285 options.stop_grace = Duration::from_millis(600);
2286 let line = format!(
2287 r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
2288 options.formation_id
2289 );
2290 let script = dir.path().join("mur");
2291 std::fs::write(dir.path().join("line"), format!("{line}\n")).unwrap();
2292 std::fs::write(
2293 &script,
2294 format!(
2295 "#!/bin/sh\ncat '{}'\nexec sleep 600\n",
2296 dir.path().join("line").display()
2297 ),
2298 )
2299 .unwrap();
2300 use std::os::unix::fs::PermissionsExt;
2301 std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
2302 std::env::set_var(child_launch::MUR_BINARY_ENV, &script);
2303 let result = launch_plan(plan_of("coder"), options);
2304 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2305 let formation = result.expect("the door answers");
2306 let pid = formation.peers()[0].pid;
2307 let dropped = Instant::now();
2308 drop(formation);
2309 assert!(dropped.elapsed() >= Duration::from_millis(600));
2310 assert!(!crate::running::pid_is_alive(pid));
2311 }
2312
2313 #[test]
2318 fn each_member_reads_only_its_own_credentials_and_callees_on_its_channel() {
2319 use crate::formation_credentials::{parse_address_line, FormationBundle};
2320 let _guard = binary_guard();
2321 let dir = tempfile::tempdir().unwrap();
2322 let coder_port = door("ses_coder", Duration::ZERO);
2323 let reviewer_port = door("ses_reviewer", Duration::ZERO);
2324 let options = options(Duration::from_secs(20));
2325 let id = options.formation_id.clone();
2326 let line = |session: &str, port: u16| {
2327 format!(
2328 r#"{{"formation_id":"{id}","session_id":"{session}","url":"localhost:{port}"}}"#
2329 )
2330 };
2331 let out = dir.path().display().to_string();
2332 let script = dir.path().join("mur");
2333 std::fs::write(
2334 &script,
2335 format!(
2336 "#!/bin/sh\n\
2337 case \" $* \" in\n\
2338 *' --capsule coder '*) name=coder; line='{coder}';;\n\
2339 *' --capsule reviewer '*) name=reviewer; line='{reviewer}';;\n\
2340 *) name=planner; line='';;\n\
2341 esac\n\
2342 env > '{out}/'$name.env\n\
2343 printf '%s\\n' \"$*\" > '{out}/'$name.argv\n\
2344 cat /dev/fd/$MURMUR_FORMATION_CHANNEL >> '{out}/'$name.channel &\n\
2345 [ -n \"$line\" ] && printf '%s\\n' \"$line\"\n\
2346 exec sleep 600\n",
2347 coder = line("ses_coder", coder_port),
2348 reviewer = line("ses_reviewer", reviewer_port),
2349 ),
2350 )
2351 .unwrap();
2352 use std::os::unix::fs::PermissionsExt;
2353 std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
2354 std::env::set_var(child_launch::MUR_BINARY_ENV, &script);
2355 let plan = FormationPlan {
2356 peers: vec![member("coder"), member("reviewer")],
2357 entry: member("planner"),
2358 callees: vec![
2361 ("planner".to_string(), vec!["coder".to_string()]),
2362 ("coder".to_string(), vec!["reviewer".to_string()]),
2363 ],
2364 };
2365 let result = launch_plan(plan, options);
2366 let mut formation = result.expect("both stand-ins came up");
2367 formation.start_entry().expect("the entry stand-in starts");
2368 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2369
2370 let read_lines = |name: &str, count: usize| -> Vec<String> {
2371 let path = dir.path().join(format!("{name}.channel"));
2372 let deadline = Instant::now() + Duration::from_secs(10);
2373 loop {
2374 let lines: Vec<String> = std::fs::read_to_string(&path)
2375 .unwrap_or_default()
2376 .lines()
2377 .map(str::to_string)
2378 .collect();
2379 if lines.len() >= count || Instant::now() >= deadline {
2380 return lines;
2381 }
2382 std::thread::sleep(POLL_INTERVAL);
2383 }
2384 };
2385 let coder = read_lines("coder", 2);
2386 let planner = read_lines("planner", 2);
2387 let reviewer = read_lines("reviewer", 1);
2388 std::thread::sleep(Duration::from_millis(300));
2389 assert_eq!(read_lines("reviewer", 1).len(), 1, "reviewer calls nobody");
2390
2391 let bundle = |line: &str| FormationBundle::parse_line(line).expect("a member's bundle");
2392 let names = |bundle: &FormationBundle| -> Vec<String> {
2393 bundle.calls.iter().map(|(name, _)| name.clone()).collect()
2394 };
2395 let (coder_bundle, planner_bundle, reviewer_bundle) =
2396 (bundle(&coder[0]), bundle(&planner[0]), bundle(&reviewer[0]));
2397 assert_eq!(coder_bundle.member, "coder");
2398 assert_eq!(names(&coder_bundle), ["reviewer"]);
2399 assert_eq!(names(&planner_bundle), ["coder"]);
2400 assert!(names(&reviewer_bundle).is_empty());
2401 let launcher = crate::running::ProcessIdentity::of_this_process();
2404 assert!(launcher.is_some());
2405 for bundle in [&coder_bundle, &planner_bundle, &reviewer_bundle] {
2406 assert_eq!(bundle.formation_id, id);
2407 assert_eq!(bundle.verify_key, coder_bundle.verify_key);
2408 assert_eq!(bundle.launcher, launcher, "{}", bundle.member);
2409 }
2410 assert!(!planner.join("\n").contains("reviewer"), "{planner:?}");
2411 let addresses = |line: &str| -> Vec<(String, String)> {
2412 parse_address_line(line)
2413 .unwrap()
2414 .into_iter()
2415 .map(|peer| (peer.name, peer.url))
2416 .collect()
2417 };
2418 assert_eq!(
2419 addresses(&coder[1]),
2420 [(
2421 "reviewer".to_string(),
2422 format!("http://localhost:{reviewer_port}")
2423 )]
2424 );
2425 assert_eq!(
2426 addresses(&planner[1]),
2427 [(
2428 "coder".to_string(),
2429 format!("http://localhost:{coder_port}")
2430 )]
2431 );
2432
2433 for name in ["coder", "reviewer", "planner"] {
2434 let env = std::fs::read_to_string(dir.path().join(format!("{name}.env"))).unwrap();
2435 let argv = std::fs::read_to_string(dir.path().join(format!("{name}.argv"))).unwrap();
2436 assert!(!env.contains("mft1.") && !argv.contains("mft1."), "{name}");
2437 assert!(!env.contains(FORMATION_PEERS_ENV), "{name}: {env}");
2438 assert!(
2439 env.contains(&format!("{FORMATION_CHANNEL_ENV}=")),
2440 "{name}: {env}"
2441 );
2442 }
2443 let pids: Vec<u32> = formation.peers().iter().map(|peer| peer.pid).collect();
2444 drop(formation);
2445 for pid in pids {
2446 assert!(!crate::running::pid_is_alive(pid), "{pid}");
2447 }
2448 }
2449
2450 #[test]
2451 fn a_registered_member_is_killed_by_the_panic_hooks_sweep() {
2452 use std::os::unix::process::CommandExt;
2453 let _guard = binary_guard();
2455 let mut child = std::process::Command::new("sleep")
2456 .arg("600")
2457 .process_group(0)
2458 .spawn()
2459 .unwrap();
2460 let slot = RegistrySlot::register(child.id(), true);
2461 assert!(slot.0.is_some());
2462 kill_registered_members();
2463 let status = child.wait().unwrap();
2464 use std::os::unix::process::ExitStatusExt;
2465 assert_eq!(status.signal(), Some(libc::SIGKILL));
2466 drop(slot);
2467 assert!(MEMBER_REGISTRY
2468 .iter()
2469 .all(|slot| slot.load(Ordering::SeqCst) != i64::from(child.id())));
2470 }
2471
2472 #[test]
2473 fn a_launch_failure_names_each_member_and_every_stopped_one() {
2474 let failure = MemberLaunchFailure {
2475 failed: vec![failure(
2476 &member("broken"),
2477 MemberFailureReason::ExitedBeforeReporting {
2478 status: "status 1".to_string(),
2479 stderr_tail: vec!["error[E-RUN-008]: missing".to_string()],
2480 },
2481 )],
2482 stopped: vec![
2483 StoppedMember {
2484 name: "coder".to_string(),
2485 pid: 7,
2486 },
2487 StoppedMember {
2488 name: "reviewer".to_string(),
2489 pid: 8,
2490 },
2491 ],
2492 interrupted_by: None,
2493 };
2494 assert_eq!(
2495 failure.to_string(),
2496 "the formation could not be launched\n 'broken' (broken@0.1.0): its process exited \
2497 with status 1 before reporting; its last stderr lines were:\n \
2498 error[E-RUN-008]: missing\n stopped: coder (pid 7), reviewer (pid 8)"
2499 );
2500 }
2501
2502 #[test]
2503 fn ready_peer_debug_prints_no_door_token() {
2504 let peer = ReadyPeer {
2505 name: "coder".to_string(),
2506 capsule: "coder".to_string(),
2507 version: "1.2.0".to_string(),
2508 pid: 1,
2509 session_id: "ses_x".to_string(),
2510 url: "http://localhost:1".to_string(),
2511 workdir: PathBuf::from("/home/u/.murmur/formations/frm_x/coder"),
2512 door_token: Some(DoorToken::new("mdt1.secretpayload.mac".to_string())),
2513 };
2514 let debug = format!("{peer:?}");
2515 assert!(!debug.contains("secretpayload"), "{debug}");
2516 assert!(debug.contains("<redacted>"), "{debug}");
2517 }
2518
2519 #[test]
2521 fn a_member_directory_is_made_owner_only_under_its_formation() {
2522 use std::os::unix::fs::PermissionsExt;
2523
2524 let home = tempfile::tempdir().unwrap();
2525 let formations = home.path().join(FORMATIONS_DIR);
2526 let id = FormationId::mint();
2527 let workdir = make_member_workdir(&formations, &id, "worker").unwrap();
2528 assert_eq!(workdir, formations.join(id.as_str()).join("worker"));
2529 for dir in [&formations, &formations.join(id.as_str()), &workdir] {
2530 let mode = std::fs::metadata(dir).unwrap().permissions().mode() & 0o777;
2531 assert_eq!(mode, 0o700, "{}", dir.display());
2532 }
2533 }
2534
2535 #[test]
2538 fn an_existing_member_path_is_refused_and_named() {
2539 let home = tempfile::tempdir().unwrap();
2540 let formations = home.path().join(FORMATIONS_DIR);
2541 let id = FormationId::mint();
2542 make_member_workdir(&formations, &id, "worker").unwrap();
2543 let again = make_member_workdir(&formations, &id, "worker").unwrap_err();
2544 let path = formations.join(id.as_str()).join("worker");
2545 assert!(again.contains(&path.display().to_string()), "{again}");
2546 std::os::unix::fs::symlink(home.path(), formations.join(id.as_str()).join("linked"))
2547 .unwrap();
2548 assert!(make_member_workdir(&formations, &id, "linked").is_err());
2549 }
2550
2551 #[test]
2553 fn a_peer_whose_directory_cannot_be_made_refuses_the_launch() {
2554 let _guard = binary_guard();
2555 let dir = tempfile::tempdir().unwrap();
2556 let options = options(Duration::from_secs(5));
2557 let formations = options.formations_dir.clone().unwrap();
2558 let taken = formations.join(options.formation_id.as_str()).join("coder");
2559 std::fs::create_dir_all(&taken).unwrap();
2560 std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2561 let result = launch_plan(plan_of("coder"), options);
2562 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2563 let Err(failure) = result else {
2564 panic!("the launch was expected to be refused");
2565 };
2566 assert!(failure.stopped.is_empty(), "{failure:?}");
2567 match only_reason(&failure) {
2568 MemberFailureReason::NotStarted { error } => {
2569 assert!(error.contains(&taken.display().to_string()), "{error}")
2570 }
2571 other => panic!("expected NotStarted, got {other:?}"),
2572 }
2573 }
2574
2575 #[test]
2579 fn member_roots_are_the_records_groups_of_member_directories() {
2580 let home = tempfile::tempdir().unwrap();
2581 crate::murmur_home::run_with_home(
2582 "formation_launch::tests::member_roots_inner",
2583 home.path(),
2584 );
2585 }
2586
2587 #[test]
2588 #[ignore = "run by member_roots_are_the_records_groups_of_member_directories"]
2589 fn member_roots_inner() {
2590 if !crate::murmur_home::in_scratch_home() {
2591 return;
2592 }
2593 assert!(formation_member_roots(&FormationId::mint()).is_empty());
2594
2595 let id = FormationId::mint();
2596 let home = PathBuf::from(std::env::var_os("HOME").unwrap());
2597 let formations = home.join(".murmur").join(FORMATIONS_DIR);
2598 for member in ["worker", "reviewer", "idle"] {
2599 assert_eq!(
2600 make_member_workdir(&formations, &id, member).unwrap(),
2601 formations.join(id.as_str()).join(member)
2602 );
2603 }
2604 let group = |member: &str| {
2605 crate::records_dir::records_group(&formations.join(id.as_str()).join(member)).unwrap()
2606 };
2607 for member in ["worker", "reviewer"] {
2608 std::fs::create_dir_all(group(member)).unwrap();
2609 }
2610 assert_eq!(
2611 formation_member_roots_in(&formations, &id),
2612 [group("reviewer"), group("worker")]
2613 );
2614 assert!(group("worker").starts_with(home.join(".murmur/records")));
2615 }
2616
2617 #[test]
2619 fn formation_marker_round_trips_as_three_keys() {
2620 let marker = FormationMarker {
2621 project_dir: PathBuf::from("/srv/project"),
2622 launcher_pid: 4242,
2623 launcher_start: "12345".to_string(),
2624 };
2625 let value = serde_json::to_value(&marker).unwrap();
2626 assert_eq!(
2627 value,
2628 serde_json::json!({
2629 "project_dir": "/srv/project",
2630 "launcher_pid": 4242,
2631 "launcher_start": "12345",
2632 })
2633 );
2634 let dir = tempfile::tempdir().unwrap();
2635 std::fs::write(dir.path().join(FORMATION_MARKER_FILE), value.to_string()).unwrap();
2636 assert_eq!(FormationMarker::read(dir.path()), Some(marker));
2637 std::fs::write(dir.path().join(FORMATION_MARKER_FILE), "{").unwrap();
2638 assert_eq!(FormationMarker::read(dir.path()), None);
2639 assert_eq!(FormationMarker::read(&dir.path().join("missing")), None);
2640 }
2641
2642 #[test]
2646 fn formation_marker_is_written_owner_only_before_the_first_peer_directory() {
2647 use std::os::unix::fs::PermissionsExt;
2648
2649 let _guard = binary_guard();
2650 let dir = tempfile::tempdir().unwrap();
2651 let project = tempfile::tempdir().unwrap();
2652 let mut options = options(Duration::from_secs(1));
2653 options.project_dir = project.path().join(".");
2654 let formation_dir = options
2655 .formations_dir
2656 .clone()
2657 .unwrap()
2658 .join(options.formation_id.as_str());
2659 std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2660 let result = launch_plan(plan_of("coder"), options);
2661 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2662 assert!(result.is_err(), "the stand-in never reports itself");
2663
2664 let path = formation_dir.join(FORMATION_MARKER_FILE);
2665 let metadata = std::fs::metadata(&path).unwrap();
2666 assert_eq!(metadata.permissions().mode() & 0o777, 0o600);
2667 let marker = FormationMarker::read(&formation_dir).unwrap();
2668 let pid = std::process::id();
2669 assert_eq!(
2670 marker,
2671 FormationMarker {
2672 project_dir: std::fs::canonicalize(project.path()).unwrap(),
2673 launcher_pid: pid,
2674 launcher_start: crate::running::process_start_token(pid).unwrap_or_default(),
2675 }
2676 );
2677 let peer_dir = std::fs::metadata(formation_dir.join("coder")).unwrap();
2678 assert!(
2679 metadata.modified().unwrap() <= peer_dir.modified().unwrap(),
2680 "the marker was written after the peer's directory was made"
2681 );
2682 }
2683
2684 #[test]
2686 fn formation_marker_is_never_written_without_peers() {
2687 let _guard = binary_guard();
2688 let dir = tempfile::tempdir().unwrap();
2689 let options = options(Duration::from_secs(5));
2690 let formations = options.formations_dir.clone().unwrap();
2691 std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2692 let plan = FormationPlan {
2693 peers: Vec::new(),
2694 entry: member("planner"),
2695 callees: Vec::new(),
2696 };
2697 let result = launch_plan(plan, options);
2698 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2699 let formation = result.expect("a formation with no peers has nothing to wait for");
2700 drop(formation);
2701 assert!(
2702 std::fs::read_dir(&formations).map_or(true, |mut entries| entries.next().is_none()),
2703 "{}",
2704 formations.display()
2705 );
2706 }
2707
2708 fn marker_of(pid: u32, start: &str) -> FormationMarker {
2709 FormationMarker {
2710 project_dir: PathBuf::from("/srv/project"),
2711 launcher_pid: pid,
2712 launcher_start: start.to_string(),
2713 }
2714 }
2715
2716 fn reaped_pid() -> u32 {
2718 let mut child = std::process::Command::new("true").spawn().unwrap();
2719 let pid = child.id();
2720 child.wait().unwrap();
2721 pid
2722 }
2723
2724 fn running_record(pid: u32, formation: Option<&FormationId>) -> crate::running::RunningRecord {
2725 crate::running::RunningRecord {
2726 session_id: "ses_0199c4e2f1b7712a9d3e4f5061728394".to_string(),
2727 url: "127.0.0.1:1".to_string(),
2728 pid,
2729 process_start: crate::running::process_start_token(pid).unwrap_or_default(),
2730 capsule_name: "worker".to_string(),
2731 capsule_version: "0.1.0".to_string(),
2732 workdir: PathBuf::from("/tmp/worker"),
2733 records_dir: PathBuf::from("/tmp/worker").join("records"),
2734 outlives_launcher: false,
2735 started_at: "2026-01-01T00:00:00Z".to_string(),
2736 door_token: None,
2737 credentials: Default::default(),
2738 formation_id: formation.cloned(),
2739 formation_member: None,
2740 formation_lifeline: false,
2741 formation_launcher: None,
2742 spawned_by: None,
2743 }
2744 }
2745
2746 #[test]
2749 fn prune_formations_liveness_reads_the_launcher_from_the_marker() {
2750 let id = FormationId::mint();
2751 let own = std::process::id();
2752 let token = crate::running::process_start_token(own).unwrap();
2753 assert!(formation_is_live(&id, &marker_of(own, &token), Some(&[])));
2754 assert!(!formation_is_live(
2755 &id,
2756 &marker_of(reaped_pid(), &token),
2757 Some(&[])
2758 ));
2759 }
2760
2761 #[test]
2765 fn prune_formations_liveness_reads_running_records_that_name_the_formation() {
2766 let id = FormationId::mint();
2767 let dead = marker_of(reaped_pid(), "1");
2768 let mut sleeper = std::process::Command::new("sleep")
2769 .arg("30")
2770 .spawn()
2771 .unwrap();
2772 let live = running_record(sleeper.id(), Some(&id));
2773 let other = running_record(sleeper.id(), Some(&FormationId::mint()));
2774 let unnamed = running_record(sleeper.id(), None);
2775 let gone = running_record(reaped_pid(), Some(&id));
2776
2777 assert!(formation_is_live(&id, &dead, Some(&[live])));
2778 assert!(!formation_is_live(
2779 &id,
2780 &dead,
2781 Some(&[other, unnamed, gone])
2782 ));
2783 let _ = sleeper.kill();
2784 let _ = sleeper.wait();
2785 }
2786
2787 #[test]
2790 fn prune_formations_liveness_treats_an_unlistable_running_directory_as_all_live() {
2791 let id = FormationId::mint();
2792 assert!(formation_is_live(&id, &marker_of(reaped_pid(), "1"), None));
2793
2794 let home = tempfile::tempdir().unwrap();
2795 let project = tempfile::tempdir().unwrap();
2796 let formations = home.path().join(FORMATIONS_DIR);
2797 let old = FormationId::parse(&format!("frm_{:012x}{:020x}", 1_000_000u64, 1)).unwrap();
2798 let old_dir = formations.join(old.as_str());
2799 std::fs::create_dir_all(old_dir.join("worker")).unwrap();
2800 let marker = FormationMarker {
2801 project_dir: std::fs::canonicalize(project.path()).unwrap(),
2802 ..marker_of(reaped_pid(), "1")
2803 };
2804 std::fs::write(
2805 old_dir.join(FORMATION_MARKER_FILE),
2806 serde_json::to_string(&marker).unwrap(),
2807 )
2808 .unwrap();
2809 let policy = murmur_artifact::TraceRetainConfig {
2810 max_sessions: Some(1),
2811 max_age_secs: Some(1),
2812 };
2813
2814 let running = home.path().join("running");
2816 std::fs::write(&running, "").unwrap();
2817 let current = FormationId::mint();
2818 assert!(prune_ended_formations_in(
2819 &formations,
2820 &running,
2821 ¤t,
2822 project.path(),
2823 &policy
2824 )
2825 .is_empty());
2826 assert!(old_dir.is_dir());
2827
2828 std::fs::remove_file(&running).unwrap();
2830 let pruned =
2831 prune_ended_formations_in(&formations, &running, ¤t, project.path(), &policy);
2832 assert_eq!(pruned.len(), 1, "{pruned:?}");
2833 assert_eq!(pruned[0].formation_id, old);
2834 assert_eq!(pruned[0].reason, crate::trace::RETENTION_REASON_MAX_AGE);
2835 assert!(!old_dir.exists());
2836 }
2837}