1use std::io::Write;
63use std::path::{Path, PathBuf};
64use std::process::{Child, ExitStatus};
65use std::sync::atomic::{AtomicBool, AtomicI32, AtomicI64, Ordering};
66use std::sync::{mpsc, Once};
67use std::time::{Duration, Instant};
68
69use crate::child_launch::{self, AfterReadiness, ProcessEnv, ProcessLaunch, StartedProcess};
70use crate::delegation::SPAWNER_ENV;
71use crate::door_auth::DoorToken;
72use crate::formation::{FormationId, FormationPeer, FORMATION_PEERS_ENV};
73use crate::formation_credentials::{
74 render_address_line, FormationAuthority, FORMATION_CHANNEL_ENV,
75};
76use crate::lifeline::SPAWNER_LIFELINE_ENV;
77use crate::roster::AdmittedRoster;
78use crate::running::ProcessIdentity;
79use crate::session_status::{SessionDeadline, INHERITED_DEADLINE_FLAG};
80use serde::{Deserialize, Serialize};
81
82pub const MEMBER_READY_TIMEOUT: Duration = Duration::from_secs(180);
88
89pub const MEMBER_STOP_GRACE: Duration = Duration::from_secs(25);
95
96const READY_PROBE_INTERVAL: Duration = Duration::from_millis(100);
98
99const POLL_INTERVAL: Duration = Duration::from_millis(50);
102
103const EXIT_AFTER_EOF_WAIT: Duration = Duration::from_secs(2);
105
106pub const PEER_BIND_ADDR: &str = "127.0.0.1";
108
109pub const CAPSULE_SHA256_FLAG: &str = "--capsule-sha256";
111
112pub const STORE_ROOT_FLAG: &str = "--store-root";
115
116pub const FORMATIONS_DIR: &str = "formations";
118
119pub const FORMATION_MARKER_FILE: &str = "formation.json";
122
123#[derive(Debug, Clone)]
125pub struct FormationLaunchOptions {
126 pub formation_id: FormationId,
128 pub project_dir: PathBuf,
131 pub task: Option<String>,
133 pub json: bool,
135 pub verbose: bool,
137 pub no_env_file: bool,
139 pub containment: Option<String>,
141 pub unconfined: bool,
146 pub ready_timeout: Duration,
149 pub stop_grace: Duration,
152 pub deadline: Option<SessionDeadline>,
156 pub(crate) formations_dir: Option<PathBuf>,
159}
160
161impl FormationLaunchOptions {
162 pub fn new(formation_id: FormationId, project_dir: PathBuf) -> Self {
164 Self {
165 formation_id,
166 project_dir,
167 task: None,
168 json: false,
169 verbose: false,
170 no_env_file: false,
171 containment: None,
172 unconfined: false,
173 ready_timeout: MEMBER_READY_TIMEOUT,
174 stop_grace: MEMBER_STOP_GRACE,
175 deadline: None,
176 formations_dir: None,
177 }
178 }
179}
180
181#[derive(Debug, Clone, PartialEq, Eq)]
183pub(crate) struct PlannedMember {
184 pub(crate) name: String,
185 pub(crate) capsule: String,
186 pub(crate) version: String,
187 pub(crate) sha256: String,
189}
190
191#[derive(Debug, Clone)]
194pub(crate) struct FormationPlan {
195 pub(crate) peers: Vec<PlannedMember>,
196 pub(crate) entry: PlannedMember,
197 pub(crate) callees: Vec<(String, Vec<String>)>,
200}
201
202impl FormationPlan {
203 pub(crate) fn from_roster(roster: &AdmittedRoster) -> Self {
204 let planned = |member: &crate::roster::AdmittedMember| PlannedMember {
205 name: member.name.clone(),
206 capsule: member.capsule.clone(),
207 version: member.version.clone(),
208 sha256: member.sha256.clone(),
209 };
210 let entry = roster.entry();
211 Self {
212 peers: roster
213 .members()
214 .iter()
215 .filter(|member| !member.entry)
216 .map(planned)
217 .collect(),
218 entry: planned(entry),
219 callees: roster
220 .members()
221 .iter()
222 .map(|member| {
223 let callees: Vec<String> = roster
224 .callees(&member.name)
225 .map(|callee| callee.name.clone())
226 .collect();
227 (member.name.clone(), callees)
228 })
229 .filter(|(_, callees)| !callees.is_empty())
230 .collect(),
231 }
232 }
233
234 pub(crate) fn callees_of(&self, member: &str) -> &[String] {
236 self.callees
237 .iter()
238 .find(|(name, _)| name == member)
239 .map(|(_, callees)| callees.as_slice())
240 .unwrap_or_default()
241 }
242
243 fn first_line(
246 &self,
247 authority: &FormationAuthority,
248 launcher: Option<&ProcessIdentity>,
249 member: &str,
250 ) -> String {
251 let callees: Vec<&str> = self.callees_of(member).iter().map(String::as_str).collect();
252 let mut bundle = authority.member_bundle(member, &callees);
253 bundle.launcher = launcher.cloned();
254 bundle.entry = member == self.entry.name;
255 bundle.render_line()
256 }
257}
258
259pub struct ReadyPeer {
261 pub name: String,
262 pub capsule: String,
263 pub version: String,
264 pub pid: u32,
265 pub session_id: String,
266 pub url: String,
268 pub workdir: PathBuf,
271 door_token: Option<DoorToken>,
274}
275
276impl ReadyPeer {
277 pub fn door_token(&self) -> Option<&DoorToken> {
279 self.door_token.as_ref()
280 }
281}
282
283impl std::fmt::Debug for ReadyPeer {
284 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
285 f.debug_struct("ReadyPeer")
286 .field("name", &self.name)
287 .field("capsule", &self.capsule)
288 .field("version", &self.version)
289 .field("pid", &self.pid)
290 .field("session_id", &self.session_id)
291 .field("url", &self.url)
292 .field("workdir", &self.workdir)
293 .field(
294 "door_token",
295 &self.door_token.as_ref().map(|_| "<redacted>"),
296 )
297 .finish()
298 }
299}
300
301#[derive(Debug, Clone, PartialEq, Eq)]
303pub enum MemberFailureReason {
304 NotStarted { error: String },
306 ExitedBeforeReporting {
308 status: String,
309 stderr_tail: Vec<String>,
310 },
311 ExitedBeforeDoorAnswered {
313 status: String,
314 stderr_tail: Vec<String>,
315 },
316 DidNotReport { deadline: Duration },
318 UnreadableReport { why: String },
320 ReportedNoDoor,
322 WrongFormation {
324 reported: Option<String>,
325 expected: String,
326 },
327 DoorDidNotAnswer {
329 url: String,
330 session_id: String,
331 deadline: Duration,
332 last_error: String,
333 },
334}
335
336impl std::fmt::Display for MemberFailureReason {
337 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
338 let tail = |f: &mut std::fmt::Formatter<'_>, lines: &[String]| -> std::fmt::Result {
339 if lines.is_empty() {
340 return f.write_str("; it wrote nothing to stderr");
341 }
342 f.write_str("; its last stderr lines were:")?;
343 for line in lines {
344 write!(f, "\n {line}")?;
345 }
346 Ok(())
347 };
348 match self {
349 Self::NotStarted { error } => write!(f, "its process could not be started: {error}"),
350 Self::ExitedBeforeReporting {
351 status,
352 stderr_tail,
353 } => {
354 write!(f, "its process exited with {status} before reporting")?;
355 tail(f, stderr_tail)
356 }
357 Self::ExitedBeforeDoorAnswered {
358 status,
359 stderr_tail,
360 } => {
361 write!(
362 f,
363 "its process reported and then exited with {status} before its door answered"
364 )?;
365 tail(f, stderr_tail)
366 }
367 Self::DidNotReport { deadline } => {
368 write!(f, "it did not report within {}s", deadline.as_secs_f32())
369 }
370 Self::UnreadableReport { why } => {
371 write!(f, "its first line is not a readiness line: {why}")
372 }
373 Self::ReportedNoDoor => f.write_str("it reported no door"),
374 Self::WrongFormation { reported, expected } => match reported {
375 Some(reported) => {
376 write!(f, "it reported formation {reported} instead of {expected}")
377 }
378 None => write!(f, "it reported no formation instead of {expected}"),
379 },
380 Self::DoorDidNotAnswer {
381 url,
382 session_id,
383 deadline,
384 last_error,
385 } => write!(
386 f,
387 "its door at {url} did not answer as session {session_id} within {}s; the last \
388 probe said: {last_error}",
389 deadline.as_secs_f32()
390 ),
391 }
392 }
393}
394
395#[derive(Debug, Clone, PartialEq, Eq)]
397pub struct MemberFailure {
398 pub name: String,
399 pub capsule: String,
400 pub version: String,
401 pub reason: MemberFailureReason,
402}
403
404#[derive(Debug, Clone, PartialEq, Eq)]
406pub struct StoppedMember {
407 pub name: String,
408 pub pid: u32,
409}
410
411#[derive(Debug, Clone, PartialEq, Eq)]
413pub struct MemberLaunchFailure {
414 pub failed: Vec<MemberFailure>,
417 pub stopped: Vec<StoppedMember>,
419 pub interrupted_by: Option<i32>,
422 pub deadline_reached: bool,
425}
426
427impl std::fmt::Display for MemberLaunchFailure {
428 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
429 match self.interrupted_by {
430 Some(signal) if self.failed.is_empty() => write!(
431 f,
432 "the formation launch was interrupted by {} before every member was ready",
433 signal_name(signal)
434 )?,
435 None if self.deadline_reached && self.failed.is_empty() => {
436 f.write_str("the formation reached its deadline before its entry member started")?
437 }
438 _ => f.write_str("the formation could not be launched")?,
439 }
440 for failure in &self.failed {
441 write!(
442 f,
443 "\n '{}' ({}@{}): {}",
444 failure.name, failure.capsule, failure.version, failure.reason
445 )?;
446 }
447 write!(f, "\n stopped: {}", render_stopped(&self.stopped))
448 }
449}
450
451pub fn render_stopped(stopped: &[StoppedMember]) -> String {
453 if stopped.is_empty() {
454 return "none".to_string();
455 }
456 stopped
457 .iter()
458 .map(|member| format!("{} (pid {})", member.name, member.pid))
459 .collect::<Vec<_>>()
460 .join(", ")
461}
462
463pub fn signal_name(signal: i32) -> String {
465 match signal {
466 libc::SIGINT => "SIGINT".to_string(),
467 libc::SIGTERM => "SIGTERM".to_string(),
468 libc::SIGHUP => "SIGHUP".to_string(),
469 other => format!("signal {other}"),
470 }
471}
472
473#[derive(Debug, Clone, PartialEq, Eq)]
475pub enum FormationEnding {
476 EntryExited(ExitStatus),
478 Signalled(i32),
480}
481
482#[derive(Debug, Clone)]
484pub struct FormationExit {
485 pub ending: FormationEnding,
486 pub stopped: Vec<StoppedMember>,
488}
489
490impl FormationExit {
491 pub fn exit_code(&self) -> i32 {
494 match &self.ending {
495 FormationEnding::EntryExited(status) => crate::shell::exit_code_of(status),
496 FormationEnding::Signalled(signal) => 128 + signal,
497 }
498 }
499}
500
501fn describe_status(status: ExitStatus) -> String {
503 if let Some(code) = status.code() {
504 return format!("status {code}");
505 }
506 #[cfg(unix)]
507 {
508 use std::os::unix::process::ExitStatusExt;
509 if let Some(signal) = status.signal() {
510 return format!("signal {signal}");
511 }
512 }
513 status.to_string()
514}
515
516static CAUGHT_SIGNAL: AtomicI32 = AtomicI32::new(0);
520
521#[cfg(unix)]
522extern "C" fn record_signal(signal: libc::c_int) {
523 CAUGHT_SIGNAL.store(signal, Ordering::SeqCst);
525}
526
527#[cfg(unix)]
533#[allow(unsafe_code)]
534pub fn catch_launcher_signals() -> std::io::Result<()> {
535 for signal in [libc::SIGINT, libc::SIGTERM, libc::SIGHUP] {
536 let installed = unsafe {
540 let mut action: libc::sigaction = std::mem::zeroed();
541 action.sa_sigaction = record_signal as extern "C" fn(libc::c_int) as libc::sighandler_t;
542 action.sa_flags = libc::SA_RESTART;
543 libc::sigemptyset(&mut action.sa_mask);
544 libc::sigaction(signal, &action, std::ptr::null_mut())
545 };
546 if installed != 0 {
547 return Err(std::io::Error::last_os_error());
548 }
549 }
550 Ok(())
551}
552
553#[cfg(not(unix))]
554pub fn catch_launcher_signals() -> std::io::Result<()> {
555 Ok(())
556}
557
558pub fn caught_signal() -> Option<i32> {
560 match CAUGHT_SIGNAL.load(Ordering::SeqCst) {
561 0 => None,
562 signal => Some(signal),
563 }
564}
565
566const REGISTRY_SLOTS: usize = 1024;
571
572static MEMBER_REGISTRY: [AtomicI64; REGISTRY_SLOTS] = [const { AtomicI64::new(0) }; REGISTRY_SLOTS];
576
577struct RegistrySlot(Option<usize>);
579
580impl RegistrySlot {
581 fn register(pid: u32, own_group: bool) -> Self {
582 let pid = i64::from(pid);
583 let value = if own_group { pid } else { -pid };
584 let slot = MEMBER_REGISTRY.iter().position(|slot| {
585 slot.compare_exchange(0, value, Ordering::SeqCst, Ordering::SeqCst)
586 .is_ok()
587 });
588 Self(slot)
589 }
590
591 fn release(&mut self) {
592 if let Some(index) = self.0.take() {
593 MEMBER_REGISTRY[index].store(0, Ordering::SeqCst);
594 }
595 }
596}
597
598impl Drop for RegistrySlot {
599 fn drop(&mut self) {
600 self.release();
601 }
602}
603
604pub(crate) fn kill_registered_members() {
607 for slot in &MEMBER_REGISTRY {
608 match slot.load(Ordering::SeqCst) {
609 0 => {}
610 pid if pid > 0 => {
611 signal_group(pid, libc::SIGKILL);
612 }
613 pid => signal_pid(-pid, libc::SIGKILL),
614 }
615 }
616}
617
618fn install_panic_hook() {
622 static INSTALLED: Once = Once::new();
623 INSTALLED.call_once(|| {
624 let previous = std::panic::take_hook();
625 std::panic::set_hook(Box::new(move |info| {
626 previous(info);
627 if cfg!(panic = "abort") {
628 kill_registered_members();
629 }
630 }));
631 });
632}
633
634#[cfg(unix)]
635#[allow(unsafe_code)]
636fn signal_pid(pid: i64, signal: libc::c_int) {
637 let Ok(pid) = libc::pid_t::try_from(pid) else {
638 return;
639 };
640 if pid <= 0 {
641 return;
642 }
643 unsafe {
646 libc::kill(pid, signal);
647 }
648}
649
650#[cfg(unix)]
653#[allow(unsafe_code)]
654pub(crate) fn signal_group(pgid: i64, signal: libc::c_int) -> bool {
655 let Ok(pgid) = libc::pid_t::try_from(pgid) else {
656 return false;
657 };
658 if pgid <= 1 {
659 return false;
660 }
661 unsafe { libc::killpg(pgid, signal) == 0 }
664}
665
666#[cfg(not(unix))]
667fn signal_pid(_pid: i64, _signal: i32) {}
668
669#[cfg(not(unix))]
670pub(crate) fn signal_group(_pgid: i64, _signal: i32) -> bool {
671 false
672}
673
674struct MemberProcess {
678 name: String,
679 pid: u32,
680 child: Child,
681 own_group: bool,
682 #[cfg(unix)]
685 lifeline: Option<crate::lifeline::MemberLifeline>,
686 status: Option<ExitStatus>,
689 slot: RegistrySlot,
690 grace: Duration,
691 channel: Option<std::io::PipeWriter>,
694}
695
696impl MemberProcess {
697 fn new(
698 name: &str,
699 child: Child,
700 own_group: bool,
701 grace: Duration,
702 channel: std::io::PipeWriter,
703 #[cfg(unix)] lifeline: Option<crate::lifeline::MemberLifeline>,
704 ) -> Self {
705 let pid = child.id();
706 Self {
707 name: name.to_string(),
708 pid,
709 child,
710 own_group,
711 #[cfg(unix)]
712 lifeline,
713 status: None,
714 slot: RegistrySlot::register(pid, own_group),
715 grace,
716 channel: Some(channel),
717 }
718 }
719
720 fn send_line(&mut self, line: &str) {
723 if let Some(channel) = self.channel.as_mut() {
724 let _ = channel.write_all(format!("{line}\n").as_bytes());
725 }
726 }
727
728 fn has_exited(&self) -> bool {
730 self.status.is_some() || !crate::running::pid_is_alive(self.pid)
731 }
732
733 fn exit_status_now(&mut self) -> Option<ExitStatus> {
738 if self.status.is_none() && self.has_exited() {
739 self.reap();
740 }
741 self.status
742 }
743
744 fn reap(&mut self) {
745 if self.status.is_some() {
746 return;
747 }
748 if self.own_group {
749 signal_group(i64::from(self.pid), libc::SIGKILL);
750 } else if !self.has_exited() {
751 signal_pid(i64::from(self.pid), libc::SIGKILL);
752 }
753 self.status = self.child.wait().ok();
754 if self.status.is_none() {
755 self.status = Some(ExitStatus::default());
757 }
758 self.slot.release();
759 self.channel = None;
760 }
761
762 fn close_lifeline(&mut self) {
764 #[cfg(unix)]
765 if let Some(lifeline) = &mut self.lifeline {
766 lifeline.close();
767 }
768 }
769
770 fn is_running(&self) -> bool {
772 self.status.is_none() && !self.has_exited()
773 }
774
775 fn stop(&mut self) -> Option<StoppedMember> {
778 let was_running = self.is_running();
779 self.stop_from(was_running)
780 }
781
782 fn stop_from(&mut self, was_running: bool) -> Option<StoppedMember> {
785 if self.status.is_some() {
786 return None;
787 }
788 self.close_lifeline();
789 if was_running {
790 let deadline = Instant::now() + self.grace;
791 while !self.has_exited() && Instant::now() < deadline {
792 std::thread::sleep(POLL_INTERVAL);
793 }
794 }
795 self.reap();
796 was_running.then(|| StoppedMember {
797 name: self.name.clone(),
798 pid: self.pid,
799 })
800 }
801}
802
803impl Drop for MemberProcess {
804 fn drop(&mut self) {
805 let _ = self.stop();
806 }
807}
808
809struct MemberSet(Vec<MemberProcess>);
811
812impl MemberSet {
813 fn stop_all(&mut self) -> Vec<StoppedMember> {
816 let running: Vec<bool> = self.0.iter().map(MemberProcess::is_running).collect();
819 for member in &mut self.0 {
820 member.close_lifeline();
821 }
822 std::thread::scope(|scope| {
823 let stopping: Vec<_> = self
824 .0
825 .iter_mut()
826 .zip(running)
827 .map(|(member, was_running)| scope.spawn(move || member.stop_from(was_running)))
828 .collect();
829 stopping
830 .into_iter()
831 .filter_map(|handle| handle.join().ok().flatten())
832 .collect()
833 })
834 }
835}
836
837impl Drop for MemberSet {
838 fn drop(&mut self) {
839 let _ = self.stop_all();
840 }
841}
842
843fn member_args(member: &PlannedMember, workdir: &Path) -> Vec<String> {
847 vec![
848 "run".to_string(),
849 "--capsule".to_string(),
850 member.capsule.clone(),
851 "--capsule-version".to_string(),
852 member.version.clone(),
853 CAPSULE_SHA256_FLAG.to_string(),
854 member.sha256.clone(),
855 "--workdir".to_string(),
856 workdir.display().to_string(),
857 ]
858}
859
860fn pass_through(args: &mut Vec<String>, options: &FormationLaunchOptions) {
863 if options.no_env_file {
864 args.push("--no-env-file".to_string());
865 }
866 if let Some(class) = &options.containment {
867 args.push("--containment".to_string());
868 args.push(class.clone());
869 }
870 if options.unconfined {
871 args.push("--unconfined".to_string());
872 } else {
873 args.push(child_launch::CONTAINED_LAUNCH_FLAG.to_string());
874 }
875 if let Some(deadline) = &options.deadline {
876 args.push(INHERITED_DEADLINE_FLAG.to_string());
877 args.push(deadline.at_unix_millis().to_string());
878 }
879}
880
881pub(crate) fn peer_args(
885 member: &PlannedMember,
886 options: &FormationLaunchOptions,
887 workdir: &Path,
888) -> Vec<String> {
889 let mut args = member_args(member, workdir);
890 args.extend([
891 STORE_ROOT_FLAG.to_string(),
892 options.project_dir.display().to_string(),
893 "--json".to_string(),
894 child_launch::READINESS_TOKENS_FLAG.to_string(),
895 "--bind".to_string(),
896 PEER_BIND_ADDR.to_string(),
897 ]);
898 pass_through(&mut args, options);
899 args
900}
901
902pub(crate) fn entry_args(member: &PlannedMember, options: &FormationLaunchOptions) -> Vec<String> {
911 let mut args = member_args(member, &options.project_dir);
912 let (task_acceptance, after_task) = match options.task {
913 Some(_) => ("single", "exit"),
914 None => ("queue", "sleep"),
915 };
916 args.extend([
917 "--lifecycle-task-acceptance".to_string(),
918 task_acceptance.to_string(),
919 "--lifecycle-after-task".to_string(),
920 after_task.to_string(),
921 ]);
922 if let Some(task) = &options.task {
923 args.push("--task".to_string());
924 args.push(task.clone());
925 }
926 if options.json {
927 args.push("--json".to_string());
928 }
929 if options.verbose {
930 args.push("--verbose".to_string());
931 }
932 pass_through(&mut args, options);
933 args
934}
935
936fn formations_dir() -> Result<PathBuf, String> {
938 Ok(crate::state_store::murmur_home_dir()?.join(FORMATIONS_DIR))
939}
940
941pub fn formation_member_roots(formation_id: &FormationId) -> Vec<PathBuf> {
948 match formations_dir() {
949 Ok(dir) => formation_member_roots_in(&dir, formation_id),
950 Err(_) => Vec::new(),
951 }
952}
953
954pub fn formation_member_roots_in(
957 formations_dir: &Path,
958 formation_id: &FormationId,
959) -> Vec<PathBuf> {
960 let dir = formations_dir.join(formation_id.as_str());
961 let entries = match std::fs::read_dir(&dir) {
962 Ok(entries) => entries,
963 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Vec::new(),
964 Err(_) => return vec![dir],
965 };
966 let mut roots: Vec<PathBuf> = entries
967 .filter_map(Result::ok)
968 .filter(|entry| entry.file_type().is_ok_and(|kind| kind.is_dir()))
969 .filter_map(|entry| crate::records_dir::records_group(&entry.path()).ok())
970 .filter(|root| root.exists())
971 .collect();
972 roots.sort();
973 roots
974}
975
976#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
982pub struct FormationMarker {
983 pub project_dir: PathBuf,
985 pub launcher_pid: u32,
987 pub launcher_start: String,
990}
991
992impl FormationMarker {
993 pub fn for_this_launcher(project_dir: &Path) -> Self {
995 let pid = std::process::id();
996 Self {
997 project_dir: canonical_project_dir(project_dir),
998 launcher_pid: pid,
999 launcher_start: crate::running::process_start_token(pid).unwrap_or_default(),
1000 }
1001 }
1002
1003 pub fn read(formation_dir: &Path) -> Option<Self> {
1005 let raw = std::fs::read_to_string(formation_dir.join(FORMATION_MARKER_FILE)).ok()?;
1006 serde_json::from_str(&raw).ok()
1007 }
1008}
1009
1010pub(crate) fn canonical_project_dir(project_dir: &Path) -> PathBuf {
1013 std::fs::canonicalize(project_dir).unwrap_or_else(|_| project_dir.to_path_buf())
1014}
1015
1016fn write_formation_marker(options: &FormationLaunchOptions) -> Result<(), String> {
1019 let formations_dir = launch_formations_dir(options)?;
1020 let formation_dir = formations_dir.join(options.formation_id.as_str());
1021 for dir in [formations_dir.as_path(), formation_dir.as_path()] {
1022 crate::state_store::ensure_private_dir(dir, crate::murmur_home::MURMUR_HOME_DIR_MODE)
1023 .map_err(|reason| format!("{}: {reason}", dir.display()))?;
1024 }
1025 let marker = FormationMarker::for_this_launcher(&options.project_dir);
1026 let json = serde_json::to_string(&marker).map_err(|error| error.to_string())?;
1027 crate::murmur_home::write_private_file(
1028 &formation_dir.join(FORMATION_MARKER_FILE),
1029 format!("{json}\n").as_bytes(),
1030 )
1031}
1032
1033pub fn prune_ended_formations(
1040 current: &FormationId,
1041 project_dir: &Path,
1042 policy: &murmur_artifact::TraceRetainConfig,
1043) -> Vec<crate::retention::PrunedFormation> {
1044 let (Ok(formations_dir), Ok(running_dir)) =
1045 (formations_dir(), crate::running::running_dir_location())
1046 else {
1047 return Vec::new();
1048 };
1049 prune_ended_formations_in(&formations_dir, &running_dir, current, project_dir, policy)
1050}
1051
1052pub fn prune_ended_formations_in(
1055 formations_dir: &Path,
1056 running_dir: &Path,
1057 current: &FormationId,
1058 project_dir: &Path,
1059 policy: &murmur_artifact::TraceRetainConfig,
1060) -> Vec<crate::retention::PrunedFormation> {
1061 let records = crate::running::scan(running_dir)
1063 .ok()
1064 .map(|scan| scan.records);
1065 crate::retention::prune_formations(
1066 formations_dir,
1067 current,
1068 project_dir,
1069 policy,
1070 crate::retention::now_ms(),
1071 |id, marker| formation_is_live(id, marker, records.as_deref()),
1072 )
1073}
1074
1075fn formation_is_live(
1082 id: &FormationId,
1083 marker: &FormationMarker,
1084 records: Option<&[crate::running::RunningRecord]>,
1085) -> bool {
1086 let Some(records) = records else {
1087 return true;
1088 };
1089 crate::running::process_state_of(marker.launcher_pid, &marker.launcher_start).is_running()
1090 || records.iter().any(|record| {
1091 record.formation_id.as_ref() == Some(id)
1092 && crate::running::process_state(record).is_running()
1093 })
1094}
1095
1096fn make_member_workdir(
1103 formations_dir: &Path,
1104 formation_id: &FormationId,
1105 member: &str,
1106) -> Result<PathBuf, String> {
1107 use std::os::unix::fs::DirBuilderExt;
1108
1109 let formation_dir = formations_dir.join(formation_id.as_str());
1110 let workdir = formation_dir.join(member);
1111 let failed = |path: &Path, reason: String| {
1112 format!(
1113 "its directory {} could not be made: {}: {reason}",
1114 workdir.display(),
1115 path.display()
1116 )
1117 };
1118 for dir in [formations_dir, formation_dir.as_path()] {
1119 crate::state_store::ensure_private_dir(dir, crate::murmur_home::MURMUR_HOME_DIR_MODE)
1120 .map_err(|reason| failed(dir, reason))?;
1121 }
1122 std::fs::DirBuilder::new()
1123 .mode(crate::murmur_home::MURMUR_HOME_DIR_MODE)
1124 .create(&workdir)
1125 .map_err(|error| failed(&workdir, error.to_string()))?;
1126 Ok(workdir)
1128}
1129
1130fn launch_formations_dir(options: &FormationLaunchOptions) -> Result<PathBuf, String> {
1133 match &options.formations_dir {
1134 Some(dir) => Ok(dir.clone()),
1135 None => {
1136 crate::murmur_home::ensure_murmur_home()?;
1137 formations_dir()
1138 }
1139 }
1140}
1141
1142enum PeerOutcome {
1144 Ready(ReadyPeer),
1145 Failed(MemberFailureReason),
1146 Abandoned,
1148 DeadlineReached,
1150}
1151
1152struct Channel {
1155 writer: std::io::PipeWriter,
1156 reader: std::io::PipeReader,
1157}
1158
1159impl Channel {
1160 fn open(first_line: &str) -> Result<Self, MemberFailureReason> {
1163 let (reader, mut writer) = std::io::pipe().map_err(|error| {
1164 not_started(&format!(
1165 "its formation channel could not be created: {error}"
1166 ))
1167 })?;
1168 writer
1169 .write_all(format!("{first_line}\n").as_bytes())
1170 .map_err(|error| {
1171 not_started(&format!(
1172 "its formation channel could not be written: {error}"
1173 ))
1174 })?;
1175 Ok(Self { writer, reader })
1176 }
1177
1178 fn fd(&self) -> i32 {
1180 use std::os::fd::AsRawFd;
1181 self.reader.as_raw_fd()
1182 }
1183
1184 fn member_env(&self, formation_id: &FormationId) -> Vec<(String, String)> {
1188 let (id_name, id_value) = formation_id.env_pair();
1189 vec![
1190 (id_name.to_string(), id_value),
1191 (FORMATION_CHANNEL_ENV.to_string(), self.fd().to_string()),
1192 crate::host_warnings::reported_env_pair(),
1193 ]
1194 }
1195}
1196
1197fn start_peer(
1202 binary: &Path,
1203 member: &PlannedMember,
1204 first_line: &str,
1205 options: &FormationLaunchOptions,
1206 abort: &AtomicBool,
1207 #[cfg(unix)] mut lifeline: crate::lifeline::MemberLifeline,
1208) -> (Option<MemberProcess>, PeerOutcome) {
1209 let workdir = match launch_formations_dir(options)
1210 .and_then(|dir| make_member_workdir(&dir, &options.formation_id, &member.name))
1211 {
1212 Ok(workdir) => workdir,
1213 Err(error) => return (None, PeerOutcome::Failed(not_started(&error))),
1214 };
1215 let args = peer_args(member, options, &workdir);
1216 let prefix = format!("[{}] ", member.name);
1217 let channel = match Channel::open(first_line) {
1218 Ok(channel) => channel,
1219 Err(reason) => return (None, PeerOutcome::Failed(reason)),
1220 };
1221 let set = channel.member_env(&options.formation_id);
1222 let spawned_at = Instant::now();
1223 let deadline = options
1227 .deadline
1228 .map_or(spawned_at + options.ready_timeout, |formation| {
1229 (spawned_at + options.ready_timeout).min(formation.instant())
1230 });
1231 let deadline_passed = || {
1232 options
1233 .deadline
1234 .is_some_and(|formation| formation.left(Instant::now()).is_zero())
1235 };
1236 let started = child_launch::start_process(&ProcessLaunch {
1237 binary,
1238 args: &args,
1239 cwd: None,
1240 env: ProcessEnv::Inherited {
1241 set: &set,
1242 remove: &[FORMATION_PEERS_ENV, SPAWNER_ENV, SPAWNER_LIFELINE_ENV],
1243 },
1244 stdin_line: None,
1245 stderr_prefix: Some(&prefix),
1246 inherit_output: false,
1247 own_process_group: true,
1248 inherit_fd: Some(channel.fd()),
1249 #[cfg(unix)]
1250 lifeline: Some(crate::child_launch::Lifeline::Member(&lifeline)),
1251 });
1252 let Channel { writer, reader } = channel;
1254 drop(reader);
1255 let StartedProcess {
1256 mut child,
1257 stderr_tail,
1258 ..
1259 } = match started {
1260 Ok(started) => started,
1261 Err(error) => {
1262 return (
1263 None,
1264 PeerOutcome::Failed(MemberFailureReason::NotStarted {
1265 error: format!("{}: {error}", binary.display()),
1266 }),
1267 )
1268 }
1269 };
1270 #[cfg(unix)]
1271 lifeline.spawned();
1272 let stdout = child.stdout.take();
1273 let mut process = MemberProcess::new(
1274 &member.name,
1275 child,
1276 true,
1277 options.stop_grace,
1278 writer,
1279 #[cfg(unix)]
1280 Some(lifeline),
1281 );
1282 let Some(stdout) = stdout else {
1283 return (
1284 Some(process),
1285 PeerOutcome::Failed(MemberFailureReason::NotStarted {
1286 error: "the process exposed no standard output".to_string(),
1287 }),
1288 );
1289 };
1290 let lines = child_launch::relay_readiness_line(stdout, AfterReadiness::PrefixedStderr(prefix));
1291
1292 let line = loop {
1294 if abort.load(Ordering::SeqCst) {
1295 return (Some(process), PeerOutcome::Abandoned);
1296 }
1297 let now = Instant::now();
1298 if now >= deadline && deadline_passed() {
1299 return (Some(process), PeerOutcome::DeadlineReached);
1300 }
1301 if now >= deadline {
1302 return (
1303 Some(process),
1304 PeerOutcome::Failed(MemberFailureReason::DidNotReport {
1305 deadline: options.ready_timeout,
1306 }),
1307 );
1308 }
1309 match lines.recv_timeout(POLL_INTERVAL.min(deadline - now)) {
1310 Ok(Some(line)) => break line,
1311 Ok(None) | Err(mpsc::RecvTimeoutError::Disconnected) => {
1312 let closed_at = Instant::now();
1315 let status = loop {
1316 if let Some(status) = process.exit_status_now() {
1317 break status;
1318 }
1319 if closed_at.elapsed() >= EXIT_AFTER_EOF_WAIT {
1320 process.reap();
1321 break process.status.unwrap_or_default();
1322 }
1323 std::thread::sleep(POLL_INTERVAL);
1324 };
1325 if deadline_passed() {
1326 return (Some(process), PeerOutcome::DeadlineReached);
1327 }
1328 return (
1329 Some(process),
1330 PeerOutcome::Failed(MemberFailureReason::ExitedBeforeReporting {
1331 status: describe_status(status),
1332 stderr_tail: stderr_tail.lines_at_end(),
1333 }),
1334 );
1335 }
1336 Err(mpsc::RecvTimeoutError::Timeout) => {}
1337 }
1338 };
1339
1340 let report = match read_report(&line, &options.formation_id) {
1341 Ok(report) => report,
1342 Err(reason) => return (Some(process), PeerOutcome::Failed(reason)),
1343 };
1344
1345 let mut last_error;
1347 loop {
1348 if abort.load(Ordering::SeqCst) {
1349 return (Some(process), PeerOutcome::Abandoned);
1350 }
1351 if let Some(status) = process.exit_status_now() {
1352 if deadline_passed() {
1353 return (Some(process), PeerOutcome::DeadlineReached);
1354 }
1355 return (
1356 Some(process),
1357 PeerOutcome::Failed(MemberFailureReason::ExitedBeforeDoorAnswered {
1358 status: describe_status(status),
1359 stderr_tail: stderr_tail.lines_at_end(),
1360 }),
1361 );
1362 }
1363 match door_answers_as(
1364 crate::running::probe_door_session_id(&report.url, report.door_token.as_ref()),
1365 &report.session_id,
1366 ) {
1367 Ok(()) => {
1368 let pid = process.pid;
1369 return (
1370 Some(process),
1371 PeerOutcome::Ready(ReadyPeer {
1372 name: member.name.clone(),
1373 capsule: member.capsule.clone(),
1374 version: member.version.clone(),
1375 pid,
1376 session_id: report.session_id,
1377 url: report.url,
1378 workdir,
1379 door_token: report.door_token,
1380 }),
1381 );
1382 }
1383 Err(why) => last_error = why,
1384 }
1385 let now = Instant::now();
1386 if now >= deadline && deadline_passed() {
1387 return (Some(process), PeerOutcome::DeadlineReached);
1388 }
1389 if now >= deadline {
1390 return (
1391 Some(process),
1392 PeerOutcome::Failed(MemberFailureReason::DoorDidNotAnswer {
1393 url: report.url,
1394 session_id: report.session_id,
1395 deadline: options.ready_timeout,
1396 last_error,
1397 }),
1398 );
1399 }
1400 std::thread::sleep(READY_PROBE_INTERVAL.min(deadline - now));
1401 }
1402}
1403
1404fn door_answers_as(probed: Result<String, String>, reported: &str) -> Result<(), String> {
1408 match probed {
1409 Ok(session_id) if session_id == reported => Ok(()),
1410 Ok(other) => Err(format!("the door answers as session {other}")),
1411 Err(error) => Err(error),
1412 }
1413}
1414
1415struct Report {
1417 session_id: String,
1418 url: String,
1420 door_token: Option<DoorToken>,
1421}
1422
1423fn read_report(line: &str, expected: &FormationId) -> Result<Report, MemberFailureReason> {
1425 let report: serde_json::Value =
1426 serde_json::from_str(line).map_err(|error| MemberFailureReason::UnreadableReport {
1427 why: format!("it does not parse as JSON ({error})"),
1428 })?;
1429 let reported = report
1430 .get("formation_id")
1431 .and_then(serde_json::Value::as_str);
1432 if reported != Some(expected.as_str()) {
1433 return Err(MemberFailureReason::WrongFormation {
1434 reported: reported.map(str::to_string),
1435 expected: expected.as_str().to_string(),
1436 });
1437 }
1438 let session_id = report
1439 .get("session_id")
1440 .and_then(serde_json::Value::as_str)
1441 .filter(|session_id| !session_id.is_empty())
1442 .ok_or_else(|| MemberFailureReason::UnreadableReport {
1443 why: "it names no session_id".to_string(),
1444 })?;
1445 let url = report
1446 .get("url")
1447 .and_then(serde_json::Value::as_str)
1448 .unwrap_or_default();
1449 if url.is_empty() {
1450 return Err(MemberFailureReason::ReportedNoDoor);
1451 }
1452 Ok(Report {
1453 session_id: session_id.to_string(),
1454 url: format!("http://{}", url.trim_start_matches("http://")),
1455 door_token: child_launch::readiness_door_token(&report),
1456 })
1457}
1458
1459pub fn launch_formation(
1467 roster: &AdmittedRoster,
1468 options: FormationLaunchOptions,
1469) -> Result<RunningFormation, MemberLaunchFailure> {
1470 launch_plan(FormationPlan::from_roster(roster), options)
1471}
1472
1473pub(crate) fn launch_plan(
1474 plan: FormationPlan,
1475 options: FormationLaunchOptions,
1476) -> Result<RunningFormation, MemberLaunchFailure> {
1477 install_panic_hook();
1478 let binary = match child_launch::mur_binary() {
1479 Ok(binary) => binary,
1480 Err(error) => {
1481 return Err(MemberLaunchFailure {
1482 failed: plan
1483 .peers
1484 .iter()
1485 .chain([&plan.entry])
1486 .map(|member| failure(member, not_started(&error.to_string())))
1487 .collect(),
1488 stopped: Vec::new(),
1489 interrupted_by: None,
1490 deadline_reached: false,
1491 })
1492 }
1493 };
1494
1495 let authority = match FormationAuthority::generate(&options.formation_id) {
1496 Ok(authority) => authority,
1497 Err(error) => {
1498 return Err(MemberLaunchFailure {
1499 failed: plan
1500 .peers
1501 .iter()
1502 .chain([&plan.entry])
1503 .map(|member| failure(member, not_started(&error)))
1504 .collect(),
1505 stopped: Vec::new(),
1506 interrupted_by: None,
1507 deadline_reached: false,
1508 })
1509 }
1510 };
1511 #[cfg(unix)]
1515 let (peer_lifelines, entry_lifeline) = match member_lifelines(plan.peers.len() + 1) {
1516 Ok(mut lifelines) => {
1517 let entry = lifelines.pop();
1518 (lifelines, entry)
1519 }
1520 Err(error) => {
1521 return Err(MemberLaunchFailure {
1522 failed: plan
1523 .peers
1524 .iter()
1525 .chain([&plan.entry])
1526 .map(|member| {
1527 failure(
1528 member,
1529 not_started(&format!("its lifeline could not be created: {error}")),
1530 )
1531 })
1532 .collect(),
1533 stopped: Vec::new(),
1534 interrupted_by: None,
1535 deadline_reached: false,
1536 })
1537 }
1538 };
1539 let launcher = ProcessIdentity::of_this_process();
1543 let first_lines: Vec<String> = plan
1544 .peers
1545 .iter()
1546 .map(|member| plan.first_line(&authority, launcher.as_ref(), &member.name))
1547 .collect();
1548 if !plan.peers.is_empty() {
1551 if let Err(reason) = write_formation_marker(&options) {
1552 crate::runtime_err!(
1553 "[mur run] warning: formation {} has no ownership marker ({reason}); its \
1554 directory will never be removed automatically",
1555 options.formation_id
1556 );
1557 }
1558 }
1559
1560 let abort = AtomicBool::new(false);
1561 let mut outcomes: Vec<Option<(Option<MemberProcess>, PeerOutcome)>> =
1562 plan.peers.iter().map(|_| None).collect();
1563 let mut interrupted_by = caught_signal();
1564 if interrupted_by.is_some() {
1565 abort.store(true, Ordering::SeqCst);
1566 }
1567 std::thread::scope(|scope| {
1568 let (tx, rx) = mpsc::channel();
1569 #[cfg(unix)]
1570 let mut peer_lifelines = peer_lifelines.into_iter();
1571 for (index, member) in plan.peers.iter().enumerate() {
1572 let tx = tx.clone();
1573 let (binary, options, abort) = (&binary, &options, &abort);
1574 let first_line = first_lines[index].as_str();
1575 #[cfg(unix)]
1576 let lifeline = peer_lifelines
1577 .next()
1578 .expect("member_lifelines made one lifeline per peer");
1579 scope.spawn(move || {
1580 let started = start_peer(
1581 binary,
1582 member,
1583 first_line,
1584 options,
1585 abort,
1586 #[cfg(unix)]
1587 lifeline,
1588 );
1589 let _ = tx.send((index, started));
1590 });
1591 }
1592 drop(tx);
1593 loop {
1594 match rx.recv_timeout(POLL_INTERVAL) {
1595 Ok((index, outcome)) => {
1596 if matches!(
1597 outcome.1,
1598 PeerOutcome::Failed(_) | PeerOutcome::DeadlineReached
1599 ) {
1600 abort.store(true, Ordering::SeqCst);
1601 }
1602 outcomes[index] = Some(outcome);
1603 }
1604 Err(mpsc::RecvTimeoutError::Timeout) => {}
1605 Err(mpsc::RecvTimeoutError::Disconnected) => break,
1606 }
1607 if interrupted_by.is_none() {
1608 if let Some(signal) = caught_signal() {
1609 interrupted_by = Some(signal);
1610 abort.store(true, Ordering::SeqCst);
1611 }
1612 }
1613 }
1614 });
1615
1616 let mut members = MemberSet(Vec::new());
1617 let mut peers = Vec::new();
1618 let mut failed = Vec::new();
1619 let mut deadline_reached = false;
1620 for (member, outcome) in plan.peers.iter().zip(outcomes) {
1621 let Some((process, outcome)) = outcome else {
1622 continue;
1623 };
1624 members.0.extend(process);
1625 match outcome {
1626 PeerOutcome::Ready(peer) => peers.push(peer),
1627 PeerOutcome::Failed(reason) => failed.push(failure(member, reason)),
1628 PeerOutcome::Abandoned => {}
1629 PeerOutcome::DeadlineReached => deadline_reached = true,
1630 }
1631 }
1632
1633 if !failed.is_empty()
1634 || interrupted_by.is_some()
1635 || deadline_reached
1636 || peers.len() != plan.peers.len()
1637 {
1638 let stopped = members.stop_all();
1639 return Err(MemberLaunchFailure {
1640 failed,
1641 stopped,
1642 interrupted_by,
1643 deadline_reached,
1644 });
1645 }
1646
1647 for member in &mut members.0 {
1650 let addresses = addresses_of(&peers, plan.callees_of(&member.name));
1651 if !addresses.is_empty() {
1652 member.send_line(&render_address_line(&addresses));
1653 }
1654 }
1655
1656 Ok(RunningFormation {
1657 formation_id: options.formation_id.clone(),
1658 entry: plan.entry.clone(),
1659 plan,
1660 peers,
1661 members,
1662 entry_index: None,
1663 #[cfg(unix)]
1664 entry_lifeline,
1665 binary,
1666 options,
1667 authority,
1668 launcher,
1669 })
1670}
1671
1672fn addresses_of(ready: &[ReadyPeer], callees: &[String]) -> Vec<FormationPeer> {
1674 ready
1675 .iter()
1676 .filter(|peer| callees.contains(&peer.name))
1677 .map(|peer| FormationPeer {
1678 name: peer.name.clone(),
1679 url: peer.url.clone(),
1680 })
1681 .collect()
1682}
1683
1684#[cfg(unix)]
1686fn member_lifelines(count: usize) -> std::io::Result<Vec<crate::lifeline::MemberLifeline>> {
1687 (0..count)
1688 .map(|_| crate::lifeline::MemberLifeline::new())
1689 .collect()
1690}
1691
1692fn failure(member: &PlannedMember, reason: MemberFailureReason) -> MemberFailure {
1693 MemberFailure {
1694 name: member.name.clone(),
1695 capsule: member.capsule.clone(),
1696 version: member.version.clone(),
1697 reason,
1698 }
1699}
1700
1701fn not_started(error: &str) -> MemberFailureReason {
1702 MemberFailureReason::NotStarted {
1703 error: error.to_string(),
1704 }
1705}
1706
1707pub struct RunningFormation {
1713 formation_id: FormationId,
1714 entry: PlannedMember,
1715 plan: FormationPlan,
1716 peers: Vec<ReadyPeer>,
1717 members: MemberSet,
1720 entry_index: Option<usize>,
1722 #[cfg(unix)]
1725 entry_lifeline: Option<crate::lifeline::MemberLifeline>,
1726 binary: PathBuf,
1727 options: FormationLaunchOptions,
1728 authority: FormationAuthority,
1729 launcher: Option<ProcessIdentity>,
1731}
1732
1733impl std::fmt::Debug for RunningFormation {
1734 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1735 f.debug_struct("RunningFormation")
1736 .field("formation_id", &self.formation_id)
1737 .field("entry", &self.entry.name)
1738 .field("peers", &self.peers)
1739 .finish()
1740 }
1741}
1742
1743impl RunningFormation {
1744 pub fn formation_id(&self) -> &FormationId {
1745 &self.formation_id
1746 }
1747
1748 pub fn entry_name(&self) -> &str {
1750 &self.entry.name
1751 }
1752
1753 pub fn peers(&self) -> &[ReadyPeer] {
1755 &self.peers
1756 }
1757
1758 pub fn callees_of(&self, member: &str) -> &[String] {
1761 self.plan.callees_of(member)
1762 }
1763
1764 pub fn entry_pid(&self) -> Option<u32> {
1766 self.entry_index.map(|index| self.members.0[index].pid)
1767 }
1768
1769 pub fn start_entry(&mut self) -> Result<u32, MemberLaunchFailure> {
1773 if self
1775 .options
1776 .deadline
1777 .is_some_and(|deadline| deadline.left(Instant::now()).is_zero())
1778 {
1779 let stopped = self.members.stop_all();
1780 return Err(MemberLaunchFailure {
1781 failed: Vec::new(),
1782 stopped,
1783 interrupted_by: None,
1784 deadline_reached: true,
1785 });
1786 }
1787 let args = entry_args(&self.entry, &self.options);
1788 let channel = Channel::open(&self.plan.first_line(
1790 &self.authority,
1791 self.launcher.as_ref(),
1792 &self.entry.name,
1793 ))
1794 .and_then(|mut channel| {
1795 let addresses = addresses_of(&self.peers, self.plan.callees_of(&self.entry.name));
1796 if !addresses.is_empty() {
1797 channel
1798 .writer
1799 .write_all(format!("{}\n", render_address_line(&addresses)).as_bytes())
1800 .map_err(|error| {
1801 not_started(&format!(
1802 "its formation channel could not be written: {error}"
1803 ))
1804 })?;
1805 }
1806 Ok(channel)
1807 });
1808 let channel = match channel {
1809 Ok(channel) => channel,
1810 Err(reason) => {
1811 let stopped = self.members.stop_all();
1812 return Err(MemberLaunchFailure {
1813 failed: vec![failure(&self.entry, reason)],
1814 stopped,
1815 interrupted_by: None,
1816 deadline_reached: false,
1817 });
1818 }
1819 };
1820 let set = channel.member_env(&self.formation_id);
1821 let started = child_launch::start_process(&ProcessLaunch {
1822 binary: &self.binary,
1823 args: &args,
1824 cwd: None,
1825 env: ProcessEnv::Inherited {
1826 set: &set,
1827 remove: &[SPAWNER_ENV, FORMATION_PEERS_ENV, SPAWNER_LIFELINE_ENV],
1828 },
1829 stdin_line: None,
1830 stderr_prefix: None,
1831 inherit_output: true,
1832 own_process_group: false,
1833 inherit_fd: Some(channel.fd()),
1834 #[cfg(unix)]
1835 lifeline: self
1836 .entry_lifeline
1837 .as_ref()
1838 .map(crate::child_launch::Lifeline::Member),
1839 });
1840 let Channel { writer, reader } = channel;
1841 drop(reader);
1842 match started {
1843 Ok(started) => {
1844 #[cfg(unix)]
1845 let lifeline = self.entry_lifeline.take().map(|mut lifeline| {
1846 lifeline.spawned();
1847 lifeline
1848 });
1849 let process = MemberProcess::new(
1850 &self.entry.name,
1851 started.child,
1852 false,
1853 self.options.stop_grace,
1854 writer,
1855 #[cfg(unix)]
1856 lifeline,
1857 );
1858 let pid = process.pid;
1859 self.members.0.push(process);
1860 self.entry_index = Some(self.members.0.len() - 1);
1861 Ok(pid)
1862 }
1863 Err(error) => {
1864 let stopped = self.members.stop_all();
1865 Err(MemberLaunchFailure {
1866 failed: vec![failure(
1867 &self.entry,
1868 not_started(&format!("{}: {error}", self.binary.display())),
1869 )],
1870 stopped,
1871 interrupted_by: None,
1872 deadline_reached: false,
1873 })
1874 }
1875 }
1876 }
1877
1878 pub fn wait(mut self) -> FormationExit {
1881 let ending = match self.entry_index {
1882 None => caught_signal()
1883 .map(FormationEnding::Signalled)
1884 .unwrap_or(FormationEnding::EntryExited(ExitStatus::default())),
1885 Some(index) => loop {
1886 if let Some(status) = self.members.0[index].exit_status_now() {
1887 break FormationEnding::EntryExited(status);
1888 }
1889 if let Some(signal) = caught_signal() {
1890 break FormationEnding::Signalled(signal);
1891 }
1892 std::thread::sleep(POLL_INTERVAL);
1893 },
1894 };
1895 let stopped = self.members.stop_all();
1896 FormationExit { ending, stopped }
1897 }
1898}
1899
1900#[cfg(test)]
1901mod tests {
1902 use super::*;
1903 use std::io::{Read, Write};
1904 use std::net::TcpListener;
1905 use std::sync::atomic::AtomicUsize;
1906 use std::sync::Arc;
1907
1908 const LIVENESS_BOUND: Duration = Duration::from_secs(120);
1912
1913 fn load_average() -> String {
1915 std::fs::read_to_string("/proc/loadavg")
1916 .unwrap_or_default()
1917 .split(' ')
1918 .take(3)
1919 .collect::<Vec<_>>()
1920 .join(" ")
1921 }
1922
1923 #[test]
1926 fn the_production_bounds_are_180s_and_a_grace_past_the_runtime_teardown() {
1927 assert_eq!(MEMBER_READY_TIMEOUT, Duration::from_secs(180));
1928 assert_eq!(MEMBER_STOP_GRACE, Duration::from_secs(25));
1929 assert!(MEMBER_STOP_GRACE > crate::runtime::TERMINATE_TEARDOWN_DEADLINE);
1930 }
1931
1932 fn member(name: &str) -> PlannedMember {
1933 PlannedMember {
1934 name: name.to_string(),
1935 capsule: name.to_string(),
1936 version: "0.1.0".to_string(),
1937 sha256: "ab".repeat(32),
1938 }
1939 }
1940
1941 #[test]
1944 fn every_member_is_told_host_warnings_were_reported() {
1945 let channel = Channel::open("first").expect("a channel opens");
1946 let env = channel.member_env(&FormationId::mint());
1947 let values: Vec<&str> = env
1948 .iter()
1949 .filter(|(key, _)| key == crate::host_warnings::HOST_WARNINGS_REPORTED_ENV)
1950 .map(|(_, value)| value.as_str())
1951 .collect();
1952 assert_eq!(values, vec!["1"], "{env:?}");
1953 }
1954
1955 fn options(deadline: Duration) -> FormationLaunchOptions {
1956 let mut options =
1957 FormationLaunchOptions::new(FormationId::mint(), PathBuf::from("/tmp/project"));
1958 options.ready_timeout = deadline;
1959 options.stop_grace = Duration::from_secs(2);
1960 options.formations_dir = Some(tempfile::tempdir().unwrap().keep());
1962 options
1963 }
1964
1965 #[test]
1966 fn member_argv_is_exactly_the_documented_shape() {
1967 let mut options = options(MEMBER_READY_TIMEOUT);
1968 let coder = member("coder");
1969 let workdir = PathBuf::from("/home/u/.murmur/formations/frm_x/coder");
1970 assert_eq!(
1971 peer_args(&coder, &options, &workdir),
1972 [
1973 "run",
1974 "--capsule",
1975 "coder",
1976 "--capsule-version",
1977 "0.1.0",
1978 "--capsule-sha256",
1979 &"ab".repeat(32),
1980 "--workdir",
1981 "/home/u/.murmur/formations/frm_x/coder",
1982 "--store-root",
1983 "/tmp/project",
1984 "--json",
1985 "--readiness-tokens",
1986 "--bind",
1987 "127.0.0.1",
1988 child_launch::CONTAINED_LAUNCH_FLAG,
1989 ]
1990 );
1991 options.task = Some("probe".to_string());
1992 options.json = true;
1993 options.verbose = true;
1994 options.no_env_file = true;
1995 options.containment = Some("scoped".to_string());
1996 let entry = entry_args(&member("planner"), &options);
1997 assert_eq!(&entry[7..9], ["--workdir", "/tmp/project"]);
1998 assert_eq!(
1999 &entry[9..],
2000 [
2001 "--lifecycle-task-acceptance",
2002 "single",
2003 "--lifecycle-after-task",
2004 "exit",
2005 "--task",
2006 "probe",
2007 "--json",
2008 "--verbose",
2009 "--no-env-file",
2010 "--containment",
2011 "scoped",
2012 child_launch::CONTAINED_LAUNCH_FLAG,
2013 ]
2014 );
2015 assert!(
2016 !entry.iter().any(|arg| arg == "--bind"
2017 || arg == STORE_ROOT_FLAG
2018 || arg == child_launch::READINESS_TOKENS_FLAG),
2019 "{entry:?}"
2020 );
2021 let peer = peer_args(&coder, &options, &workdir);
2022 assert!(peer.ends_with(&[
2023 "--no-env-file".to_string(),
2024 "--containment".to_string(),
2025 "scoped".to_string(),
2026 child_launch::CONTAINED_LAUNCH_FLAG.to_string(),
2027 ]));
2028 assert!(!peer.iter().any(|arg| arg == "--task" || arg == "--verbose"));
2029 assert!(
2030 !peer
2031 .iter()
2032 .chain(&entry)
2033 .any(|arg| arg == INHERITED_DEADLINE_FLAG),
2034 "no deadline, no flag: {peer:?} {entry:?}"
2035 );
2036
2037 options.task = None;
2040 let untasked = entry_args(&member("planner"), &options);
2041 assert_eq!(
2042 &untasked[9..],
2043 [
2044 "--lifecycle-task-acceptance",
2045 "queue",
2046 "--lifecycle-after-task",
2047 "sleep",
2048 "--json",
2049 "--verbose",
2050 "--no-env-file",
2051 "--containment",
2052 "scoped",
2053 child_launch::CONTAINED_LAUNCH_FLAG,
2054 ]
2055 );
2056 assert_eq!(peer_args(&coder, &options, &workdir), peer);
2057 options.task = Some("probe".to_string());
2058
2059 let deadline = SessionDeadline::after(
2061 Instant::now(),
2062 std::time::SystemTime::UNIX_EPOCH + Duration::from_millis(1_792_497_600_250),
2063 60,
2064 );
2065 options.deadline = Some(deadline);
2066 let flag = [
2067 INHERITED_DEADLINE_FLAG.to_string(),
2068 "1792497660250".to_string(),
2069 ];
2070 let peer = peer_args(&coder, &options, &workdir);
2071 assert!(peer.ends_with(&flag), "{peer:?}");
2072 let entry = entry_args(&member("planner"), &options);
2073 assert!(entry.ends_with(&flag), "{entry:?}");
2074 assert_eq!(
2075 entry
2076 .iter()
2077 .filter(|arg| *arg == INHERITED_DEADLINE_FLAG)
2078 .count(),
2079 1
2080 );
2081 }
2082
2083 #[test]
2086 fn every_member_carries_the_rosters_stance() {
2087 let mut options = options(MEMBER_READY_TIMEOUT);
2088 let coder = member("coder");
2089 let workdir = PathBuf::from("/home/u/.murmur/formations/frm_x/coder");
2090 for args in [
2091 peer_args(&coder, &options, &workdir),
2092 entry_args(&member("planner"), &options),
2093 ] {
2094 assert!(
2095 args.iter()
2096 .any(|arg| arg == child_launch::CONTAINED_LAUNCH_FLAG),
2097 "{args:?}"
2098 );
2099 assert!(!args.iter().any(|arg| arg == "--unconfined"), "{args:?}");
2100 }
2101 options.unconfined = true;
2102 for args in [
2103 peer_args(&coder, &options, &workdir),
2104 entry_args(&member("planner"), &options),
2105 ] {
2106 assert_eq!(
2107 args.iter().filter(|arg| *arg == "--unconfined").count(),
2108 1,
2109 "{args:?}"
2110 );
2111 assert!(
2112 !args
2113 .iter()
2114 .any(|arg| arg == child_launch::CONTAINED_LAUNCH_FLAG),
2115 "{args:?}"
2116 );
2117 }
2118 }
2119
2120 fn binary_guard() -> std::sync::MutexGuard<'static, ()> {
2123 crate::formation::FORMATION_ENV_LOCK
2124 .lock()
2125 .unwrap_or_else(|poisoned| poisoned.into_inner())
2126 }
2127
2128 fn stand_in(dir: &Path, line: Option<&str>) -> PathBuf {
2131 let script = dir.join("mur");
2132 let print = match line {
2133 Some(line) => {
2134 std::fs::write(dir.join("line"), format!("{line}\n")).unwrap();
2135 format!("cat '{}'\n", dir.join("line").display())
2136 }
2137 None => String::new(),
2138 };
2139 let body = format!(
2142 "#!/bin/sh\ntrap 'echo TERM >> \"{dir}/sigterm\"' TERM\n{print}\
2143 cat \"/dev/fd/$MURMUR_FORMATION_LIFELINE\" >/dev/null\n",
2144 dir = dir.display()
2145 );
2146 std::fs::write(&script, body).unwrap();
2147 use std::os::unix::fs::PermissionsExt;
2148 std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
2149 script
2150 }
2151
2152 fn plan_of(peer: &str) -> FormationPlan {
2153 FormationPlan {
2154 peers: vec![member(peer)],
2155 entry: member("planner"),
2156 callees: vec![("planner".to_string(), vec![peer.to_string()])],
2157 }
2158 }
2159
2160 fn refused(line: Option<&str>, deadline: Duration) -> (MemberLaunchFailure, FormationId) {
2163 let _guard = binary_guard();
2164 let dir = tempfile::tempdir().unwrap();
2165 let options = options(deadline);
2166 let id = options.formation_id.clone();
2167 let line = line.map(|line| line.replace("{formation}", id.as_str()));
2168 std::env::set_var(
2169 child_launch::MUR_BINARY_ENV,
2170 stand_in(dir.path(), line.as_deref()),
2171 );
2172 let result = launch_plan(plan_of("coder"), options);
2173 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2174 match result {
2175 Ok(_) => panic!("the launch was expected to be refused"),
2176 Err(failure) => (failure, id),
2177 }
2178 }
2179
2180 fn assert_reaped(failure: &MemberLaunchFailure) {
2181 assert_eq!(failure.stopped.len(), 1, "{failure:?}");
2182 let pid = failure.stopped[0].pid;
2183 assert_eq!(failure.stopped[0].name, "coder");
2184 assert!(
2185 !crate::running::pid_is_alive(pid),
2186 "the stand-in {pid} is still running"
2187 );
2188 assert!(
2189 !Path::new(&format!("/proc/{pid}")).exists() || cfg!(not(target_os = "linux")),
2190 "the stand-in {pid} was not reaped"
2191 );
2192 }
2193
2194 fn only_reason(failure: &MemberLaunchFailure) -> &MemberFailureReason {
2195 assert_eq!(failure.failed.len(), 1, "{failure:?}");
2196 assert_eq!(failure.failed[0].name, "coder");
2197 &failure.failed[0].reason
2198 }
2199
2200 fn silent_port() -> u16 {
2202 crate::pinned_port::claim()
2203 }
2204
2205 #[test]
2206 fn a_door_nothing_answers_on_is_refused_at_the_deadline() {
2207 let deadline = Duration::from_secs(10);
2208 let port = silent_port();
2209 let started = Instant::now();
2210 let (failure, _) = refused(
2211 Some(&format!(
2212 r#"{{"formation_id":"{{formation}}","session_id":"ses_standin","url":"localhost:{port}"}}"#
2213 )),
2214 deadline,
2215 );
2216 assert!(started.elapsed() >= deadline);
2217 match only_reason(&failure) {
2218 MemberFailureReason::DoorDidNotAnswer {
2219 url, session_id, ..
2220 } => {
2221 assert_eq!(url, &format!("http://localhost:{port}"));
2222 assert_eq!(session_id, "ses_standin");
2223 }
2224 MemberFailureReason::DidNotReport { .. } => panic!(
2227 "the stand-in's readiness line did not arrive within {deadline:?} (load average {}); \
2228 host load is the likely cause, since the stand-in prints it at once: {failure:?}",
2229 load_average()
2230 ),
2231 other => panic!("{other:?}"),
2232 }
2233 let message = failure.to_string();
2234 assert!(
2235 message.contains("did not answer as session ses_standin"),
2236 "{message}"
2237 );
2238 assert!(message.contains("'coder' (coder@0.1.0)"), "{message}");
2239 assert_reaped(&failure);
2240 }
2241
2242 #[test]
2246 fn a_formation_deadline_before_the_entry_member_ends_the_launch_at_the_deadline() {
2247 let _guard = binary_guard();
2248 let dir = tempfile::tempdir().unwrap();
2249 let mut options = options(Duration::from_secs(60));
2250 let started = Instant::now();
2251 options.deadline = Some(SessionDeadline::after(
2252 started,
2253 std::time::SystemTime::now(),
2254 2,
2255 ));
2256 std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2257 let result = launch_plan(plan_of("coder"), options);
2258 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2259 let Err(failure) = result else {
2260 panic!("the launch was expected to end at the deadline");
2261 };
2262 let took = started.elapsed();
2263 assert!(took >= Duration::from_secs(2), "{took:?}");
2264 assert!(took < Duration::from_secs(20), "{took:?}: {failure:?}");
2265 assert!(failure.deadline_reached, "{failure:?}");
2266 assert!(failure.failed.is_empty(), "{failure:?}");
2267 assert_eq!(failure.interrupted_by, None);
2268 assert_reaped(&failure);
2269 assert_eq!(
2270 failure.to_string(),
2271 format!(
2272 "the formation reached its deadline before its entry member started\n \
2273 stopped: coder (pid {})",
2274 failure.stopped[0].pid
2275 )
2276 );
2277 }
2278
2279 #[test]
2281 fn start_entry_after_the_formation_deadline_starts_nothing() {
2282 let _guard = binary_guard();
2283 let dir = tempfile::tempdir().unwrap();
2284 let mut options = options(Duration::from_secs(60));
2285 let started = Instant::now();
2286 options.deadline = Some(SessionDeadline::after(
2287 started,
2288 std::time::SystemTime::now(),
2289 1,
2290 ));
2291 std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2292 let plan = FormationPlan {
2293 peers: Vec::new(),
2294 entry: member("planner"),
2295 callees: Vec::new(),
2296 };
2297 let launched = launch_plan(plan, options);
2298 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2299 let mut formation = launched.expect("a formation with no peers launches at once");
2300 std::thread::sleep(Duration::from_millis(1_100));
2301 let failure = formation
2302 .start_entry()
2303 .expect_err("the deadline has fallen");
2304 assert!(failure.deadline_reached, "{failure:?}");
2305 assert!(
2306 failure.failed.is_empty() && failure.stopped.is_empty(),
2307 "{failure:?}"
2308 );
2309 assert_eq!(formation.entry_pid(), None);
2310 }
2311
2312 fn card_door(mut answer: impl FnMut() -> Option<&'static str> + Send + 'static) -> u16 {
2316 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
2317 let port = listener.local_addr().unwrap().port();
2318 std::thread::spawn(move || {
2319 for stream in listener.incoming().flatten() {
2320 let mut stream = stream;
2321 let mut buffer = [0u8; 4096];
2322 let _ = stream.read(&mut buffer);
2323 let body = match answer() {
2324 Some(session_id) => serde_json::json!({
2325 "name": "coder",
2326 "capabilities": {"extensions": [{
2327 "uri": crate::identity::CAPSULE_EXTENSION_URI,
2328 "params": {"sessionId": session_id},
2329 }]},
2330 })
2331 .to_string(),
2332 None => "{}".to_string(),
2333 };
2334 let _ = write!(
2335 stream,
2336 "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\n\
2337 connection: close\r\n\r\n{body}",
2338 body.len()
2339 );
2340 }
2341 });
2342 port
2343 }
2344
2345 fn door(session_id: &'static str, answer_after: Duration) -> u16 {
2348 let opened = Instant::now();
2349 card_door(move || (opened.elapsed() >= answer_after).then_some(session_id))
2350 }
2351
2352 const WRONG_ANSWERS: usize = 3;
2354
2355 fn door_naming_another_session_first() -> (u16, Arc<AtomicUsize>) {
2359 let served = Arc::new(AtomicUsize::new(0));
2360 let counted = Arc::clone(&served);
2361 let port = card_door(move || {
2362 if counted.fetch_add(1, Ordering::SeqCst) < WRONG_ANSWERS {
2363 Some("ses_somebody_else")
2364 } else {
2365 Some("ses_standin")
2366 }
2367 });
2368 (port, served)
2369 }
2370
2371 #[test]
2374 fn a_door_naming_another_session_is_not_ready() {
2375 let _guard = binary_guard();
2376 let (port, served) = door_naming_another_session_first();
2377 let dir = tempfile::tempdir().unwrap();
2378 let options = options(LIVENESS_BOUND);
2379 let line = format!(
2380 r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
2381 options.formation_id
2382 );
2383 std::env::set_var(
2384 child_launch::MUR_BINARY_ENV,
2385 stand_in(dir.path(), Some(&line)),
2386 );
2387 let result = launch_plan(plan_of("coder"), options);
2388 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2389 let formation = match result {
2390 Ok(formation) => formation,
2391 Err(failure) => match only_reason(&failure) {
2392 MemberFailureReason::DidNotReport { .. } => panic!(
2393 "the stand-in's readiness line did not arrive within {LIVENESS_BOUND:?} \
2394 (load average {}); host load is the likely cause: {failure:?}",
2395 load_average()
2396 ),
2397 _ => panic!(
2398 "the launch was refused although the door answered as ses_standin from its \
2399 request {} on: {failure:?}",
2400 WRONG_ANSWERS + 1
2401 ),
2402 },
2403 };
2404 let peer = &formation.peers()[0];
2405 assert_eq!(peer.session_id, "ses_standin");
2406 let served = served.load(Ordering::SeqCst);
2407 assert!(
2408 served > WRONG_ANSWERS,
2409 "an answer naming another session was counted as ready: the door served {served} \
2410 request(s), and its first {WRONG_ANSWERS} named ses_somebody_else"
2411 );
2412 let pid = peer.pid;
2413 drop(formation);
2414 assert!(
2415 !crate::running::pid_is_alive(pid),
2416 "dropping the owner stops the peer"
2417 );
2418 }
2419
2420 #[test]
2421 fn a_door_counts_as_ready_only_when_it_answers_as_the_reported_session() {
2422 let port = door("ses_somebody_else", Duration::ZERO);
2423 let probed = crate::running::probe_door_session_id(&format!("localhost:{port}"), None);
2424 match door_answers_as(probed, "ses_standin") {
2425 Err(why) => assert!(
2426 why.contains("the door answers as session ses_somebody_else"),
2427 "{why}"
2428 ),
2429 Ok(()) => panic!("a door answering as ses_somebody_else counted as ses_standin"),
2430 }
2431 assert_eq!(
2432 door_answers_as(Ok("ses_standin".to_string()), "ses_standin"),
2433 Ok(())
2434 );
2435 assert_eq!(
2436 door_answers_as(Err("connection refused".to_string()), "ses_standin"),
2437 Err("connection refused".to_string())
2438 );
2439 }
2440
2441 #[test]
2442 fn an_empty_url_is_refused_as_no_door() {
2443 let (failure, _) = refused(
2444 Some(r#"{"formation_id":"{formation}","session_id":"ses_standin","url":""}"#),
2445 Duration::from_secs(10),
2446 );
2447 assert_eq!(only_reason(&failure), &MemberFailureReason::ReportedNoDoor);
2448 assert!(failure.to_string().contains("it reported no door"));
2449 assert_reaped(&failure);
2450 }
2451
2452 #[test]
2453 fn a_wrong_formation_or_none_is_refused_naming_what_was_reported() {
2454 let other = FormationId::mint();
2455 let (failure, id) = refused(
2456 Some(&format!(
2457 r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:1"}}"#,
2458 other.as_str()
2459 )),
2460 Duration::from_secs(10),
2461 );
2462 assert_eq!(
2463 only_reason(&failure),
2464 &MemberFailureReason::WrongFormation {
2465 reported: Some(other.as_str().to_string()),
2466 expected: id.as_str().to_string(),
2467 }
2468 );
2469 assert!(failure
2470 .to_string()
2471 .contains(&format!("it reported formation {other} instead of {id}")));
2472 assert_reaped(&failure);
2473
2474 let (failure, id) = refused(
2475 Some(r#"{"session_id":"ses_standin","url":"localhost:1"}"#),
2476 Duration::from_secs(10),
2477 );
2478 assert!(failure
2479 .to_string()
2480 .contains(&format!("it reported no formation instead of {id}")));
2481 assert_reaped(&failure);
2482 }
2483
2484 #[test]
2485 fn no_line_at_all_is_refused_at_the_deadline() {
2486 let started = Instant::now();
2487 let (failure, _) = refused(None, Duration::from_millis(600));
2488 assert!(started.elapsed() >= Duration::from_millis(600));
2489 assert_eq!(
2490 only_reason(&failure),
2491 &MemberFailureReason::DidNotReport {
2492 deadline: Duration::from_millis(600)
2493 }
2494 );
2495 assert_reaped(&failure);
2496 }
2497
2498 #[test]
2499 fn a_door_that_starts_answering_before_the_deadline_is_ready() {
2500 let _guard = binary_guard();
2501 let port = door("ses_standin", Duration::from_millis(400));
2502 let dir = tempfile::tempdir().unwrap();
2503 let options = options(Duration::from_secs(20));
2504 let line = format!(
2505 r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
2506 options.formation_id
2507 );
2508 std::env::set_var(
2509 child_launch::MUR_BINARY_ENV,
2510 stand_in(dir.path(), Some(&line)),
2511 );
2512 let result = launch_plan(plan_of("coder"), options);
2513 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2514 let formation = result.expect("the door answered before the deadline");
2515 assert_eq!(formation.peers().len(), 1);
2516 let peer = &formation.peers()[0];
2517 assert_eq!(peer.session_id, "ses_standin");
2518 assert_eq!(peer.url, format!("http://localhost:{port}"));
2519 assert_eq!(formation.callees_of("planner"), ["coder"]);
2520 assert!(formation.callees_of("coder").is_empty());
2521 let pid = peer.pid;
2522 assert!(crate::running::pid_is_alive(pid));
2523 let dropped = Instant::now();
2524 drop(formation);
2525 assert!(
2526 !crate::running::pid_is_alive(pid),
2527 "dropping the owner stops the peer"
2528 );
2529 assert!(
2530 dropped.elapsed() < Duration::from_secs(2),
2531 "the peer ended on its lifeline's EOF, inside the grace: {:?}",
2532 dropped.elapsed()
2533 );
2534 assert!(
2535 !dir.path().join("sigterm").exists(),
2536 "the launcher sent the peer SIGTERM"
2537 );
2538 }
2539
2540 #[test]
2542 fn a_member_deaf_to_its_lifeline_is_killed_after_the_grace() {
2543 let _guard = binary_guard();
2544 let port = door("ses_standin", Duration::ZERO);
2545 let dir = tempfile::tempdir().unwrap();
2546 let mut options = options(Duration::from_secs(20));
2547 options.stop_grace = Duration::from_millis(600);
2548 let line = format!(
2549 r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
2550 options.formation_id
2551 );
2552 let script = dir.path().join("mur");
2553 std::fs::write(dir.path().join("line"), format!("{line}\n")).unwrap();
2554 std::fs::write(
2555 &script,
2556 format!(
2557 "#!/bin/sh\ncat '{}'\nexec sleep 600\n",
2558 dir.path().join("line").display()
2559 ),
2560 )
2561 .unwrap();
2562 use std::os::unix::fs::PermissionsExt;
2563 std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
2564 std::env::set_var(child_launch::MUR_BINARY_ENV, &script);
2565 let result = launch_plan(plan_of("coder"), options);
2566 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2567 let formation = result.expect("the door answers");
2568 let pid = formation.peers()[0].pid;
2569 let dropped = Instant::now();
2570 drop(formation);
2571 assert!(dropped.elapsed() >= Duration::from_millis(600));
2572 assert!(!crate::running::pid_is_alive(pid));
2573 }
2574
2575 #[test]
2580 fn each_member_reads_only_its_own_credentials_and_callees_on_its_channel() {
2581 use crate::formation_credentials::{parse_address_line, FormationBundle};
2582 let _guard = binary_guard();
2583 let dir = tempfile::tempdir().unwrap();
2584 let coder_port = door("ses_coder", Duration::ZERO);
2585 let reviewer_port = door("ses_reviewer", Duration::ZERO);
2586 let options = options(Duration::from_secs(20));
2587 let id = options.formation_id.clone();
2588 let line = |session: &str, port: u16| {
2589 format!(
2590 r#"{{"formation_id":"{id}","session_id":"{session}","url":"localhost:{port}"}}"#
2591 )
2592 };
2593 let out = dir.path().display().to_string();
2594 let script = dir.path().join("mur");
2595 std::fs::write(
2596 &script,
2597 format!(
2598 "#!/bin/sh\n\
2599 case \" $* \" in\n\
2600 *' --capsule coder '*) name=coder; line='{coder}';;\n\
2601 *' --capsule reviewer '*) name=reviewer; line='{reviewer}';;\n\
2602 *) name=planner; line='';;\n\
2603 esac\n\
2604 env > '{out}/'$name.env\n\
2605 printf '%s\\n' \"$*\" > '{out}/'$name.argv\n\
2606 cat /dev/fd/$MURMUR_FORMATION_CHANNEL >> '{out}/'$name.channel &\n\
2607 [ -n \"$line\" ] && printf '%s\\n' \"$line\"\n\
2608 exec sleep 600\n",
2609 coder = line("ses_coder", coder_port),
2610 reviewer = line("ses_reviewer", reviewer_port),
2611 ),
2612 )
2613 .unwrap();
2614 use std::os::unix::fs::PermissionsExt;
2615 std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
2616 std::env::set_var(child_launch::MUR_BINARY_ENV, &script);
2617 let plan = FormationPlan {
2618 peers: vec![member("coder"), member("reviewer")],
2619 entry: member("planner"),
2620 callees: vec![
2623 ("planner".to_string(), vec!["coder".to_string()]),
2624 ("coder".to_string(), vec!["reviewer".to_string()]),
2625 ],
2626 };
2627 let result = launch_plan(plan, options);
2628 let mut formation = result.expect("both stand-ins came up");
2629 formation.start_entry().expect("the entry stand-in starts");
2630 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2631
2632 let read_lines = |name: &str, count: usize| -> Vec<String> {
2633 let path = dir.path().join(format!("{name}.channel"));
2634 let deadline = Instant::now() + Duration::from_secs(10);
2635 loop {
2636 let lines: Vec<String> = std::fs::read_to_string(&path)
2637 .unwrap_or_default()
2638 .lines()
2639 .map(str::to_string)
2640 .collect();
2641 if lines.len() >= count || Instant::now() >= deadline {
2642 return lines;
2643 }
2644 std::thread::sleep(POLL_INTERVAL);
2645 }
2646 };
2647 let coder = read_lines("coder", 2);
2648 let planner = read_lines("planner", 2);
2649 let reviewer = read_lines("reviewer", 1);
2650 std::thread::sleep(Duration::from_millis(300));
2651 assert_eq!(read_lines("reviewer", 1).len(), 1, "reviewer calls nobody");
2652
2653 let bundle = |line: &str| FormationBundle::parse_line(line).expect("a member's bundle");
2654 let names = |bundle: &FormationBundle| -> Vec<String> {
2655 bundle.calls.iter().map(|(name, _)| name.clone()).collect()
2656 };
2657 let (coder_bundle, planner_bundle, reviewer_bundle) =
2658 (bundle(&coder[0]), bundle(&planner[0]), bundle(&reviewer[0]));
2659 assert_eq!(coder_bundle.member, "coder");
2660 assert_eq!(names(&coder_bundle), ["reviewer"]);
2661 assert_eq!(names(&planner_bundle), ["coder"]);
2662 assert!(names(&reviewer_bundle).is_empty());
2663 let launcher = crate::running::ProcessIdentity::of_this_process();
2666 assert!(launcher.is_some());
2667 for bundle in [&coder_bundle, &planner_bundle, &reviewer_bundle] {
2668 assert_eq!(bundle.formation_id, id);
2669 assert_eq!(bundle.verify_key, coder_bundle.verify_key);
2670 assert_eq!(bundle.launcher, launcher, "{}", bundle.member);
2671 }
2672 assert!(planner_bundle.entry);
2674 assert!(!coder_bundle.entry && !reviewer_bundle.entry);
2675 assert!(planner[0].contains("\"entry\":true"), "{planner:?}");
2676 assert!(
2677 !coder[0].contains("\"entry\"") && !reviewer[0].contains("\"entry\""),
2678 "{coder:?} {reviewer:?}"
2679 );
2680 assert!(!planner.join("\n").contains("reviewer"), "{planner:?}");
2681 let addresses = |line: &str| -> Vec<(String, String)> {
2682 parse_address_line(line)
2683 .unwrap()
2684 .into_iter()
2685 .map(|peer| (peer.name, peer.url))
2686 .collect()
2687 };
2688 assert_eq!(
2689 addresses(&coder[1]),
2690 [(
2691 "reviewer".to_string(),
2692 format!("http://localhost:{reviewer_port}")
2693 )]
2694 );
2695 assert_eq!(
2696 addresses(&planner[1]),
2697 [(
2698 "coder".to_string(),
2699 format!("http://localhost:{coder_port}")
2700 )]
2701 );
2702
2703 for name in ["coder", "reviewer", "planner"] {
2704 let env = std::fs::read_to_string(dir.path().join(format!("{name}.env"))).unwrap();
2705 let argv = std::fs::read_to_string(dir.path().join(format!("{name}.argv"))).unwrap();
2706 assert!(!env.contains("mft1.") && !argv.contains("mft1."), "{name}");
2707 assert!(!env.contains(FORMATION_PEERS_ENV), "{name}: {env}");
2708 assert!(
2709 env.contains(&format!("{FORMATION_CHANNEL_ENV}=")),
2710 "{name}: {env}"
2711 );
2712 }
2713 let pids: Vec<u32> = formation.peers().iter().map(|peer| peer.pid).collect();
2714 drop(formation);
2715 for pid in pids {
2716 assert!(!crate::running::pid_is_alive(pid), "{pid}");
2717 }
2718 }
2719
2720 #[test]
2721 fn a_registered_member_is_killed_by_the_panic_hooks_sweep() {
2722 use std::os::unix::process::CommandExt;
2723 let _guard = binary_guard();
2725 let mut child = std::process::Command::new("sleep")
2726 .arg("600")
2727 .process_group(0)
2728 .spawn()
2729 .unwrap();
2730 let slot = RegistrySlot::register(child.id(), true);
2731 assert!(slot.0.is_some());
2732 kill_registered_members();
2733 let status = child.wait().unwrap();
2734 use std::os::unix::process::ExitStatusExt;
2735 assert_eq!(status.signal(), Some(libc::SIGKILL));
2736 drop(slot);
2737 assert!(MEMBER_REGISTRY
2738 .iter()
2739 .all(|slot| slot.load(Ordering::SeqCst) != i64::from(child.id())));
2740 }
2741
2742 #[test]
2743 fn a_launch_failure_names_each_member_and_every_stopped_one() {
2744 let failure = MemberLaunchFailure {
2745 failed: vec![failure(
2746 &member("broken"),
2747 MemberFailureReason::ExitedBeforeReporting {
2748 status: "status 1".to_string(),
2749 stderr_tail: vec!["error[E-RUN-008]: missing".to_string()],
2750 },
2751 )],
2752 stopped: vec![
2753 StoppedMember {
2754 name: "coder".to_string(),
2755 pid: 7,
2756 },
2757 StoppedMember {
2758 name: "reviewer".to_string(),
2759 pid: 8,
2760 },
2761 ],
2762 interrupted_by: None,
2763 deadline_reached: false,
2764 };
2765 assert_eq!(
2766 failure.to_string(),
2767 "the formation could not be launched\n 'broken' (broken@0.1.0): its process exited \
2768 with status 1 before reporting; its last stderr lines were:\n \
2769 error[E-RUN-008]: missing\n stopped: coder (pid 7), reviewer (pid 8)"
2770 );
2771 }
2772
2773 #[test]
2774 fn ready_peer_debug_prints_no_door_token() {
2775 let peer = ReadyPeer {
2776 name: "coder".to_string(),
2777 capsule: "coder".to_string(),
2778 version: "1.2.0".to_string(),
2779 pid: 1,
2780 session_id: "ses_x".to_string(),
2781 url: "http://localhost:1".to_string(),
2782 workdir: PathBuf::from("/home/u/.murmur/formations/frm_x/coder"),
2783 door_token: Some(DoorToken::new("mdt1.secretpayload.mac".to_string())),
2784 };
2785 let debug = format!("{peer:?}");
2786 assert!(!debug.contains("secretpayload"), "{debug}");
2787 assert!(debug.contains("<redacted>"), "{debug}");
2788 }
2789
2790 #[test]
2792 fn a_member_directory_is_made_owner_only_under_its_formation() {
2793 use std::os::unix::fs::PermissionsExt;
2794
2795 let home = tempfile::tempdir().unwrap();
2796 let formations = home.path().join(FORMATIONS_DIR);
2797 let id = FormationId::mint();
2798 let workdir = make_member_workdir(&formations, &id, "worker").unwrap();
2799 assert_eq!(workdir, formations.join(id.as_str()).join("worker"));
2800 for dir in [&formations, &formations.join(id.as_str()), &workdir] {
2801 let mode = std::fs::metadata(dir).unwrap().permissions().mode() & 0o777;
2802 assert_eq!(mode, 0o700, "{}", dir.display());
2803 }
2804 }
2805
2806 #[test]
2809 fn an_existing_member_path_is_refused_and_named() {
2810 let home = tempfile::tempdir().unwrap();
2811 let formations = home.path().join(FORMATIONS_DIR);
2812 let id = FormationId::mint();
2813 make_member_workdir(&formations, &id, "worker").unwrap();
2814 let again = make_member_workdir(&formations, &id, "worker").unwrap_err();
2815 let path = formations.join(id.as_str()).join("worker");
2816 assert!(again.contains(&path.display().to_string()), "{again}");
2817 std::os::unix::fs::symlink(home.path(), formations.join(id.as_str()).join("linked"))
2818 .unwrap();
2819 assert!(make_member_workdir(&formations, &id, "linked").is_err());
2820 }
2821
2822 #[test]
2824 fn a_peer_whose_directory_cannot_be_made_refuses_the_launch() {
2825 let _guard = binary_guard();
2826 let dir = tempfile::tempdir().unwrap();
2827 let options = options(Duration::from_secs(5));
2828 let formations = options.formations_dir.clone().unwrap();
2829 let taken = formations.join(options.formation_id.as_str()).join("coder");
2830 std::fs::create_dir_all(&taken).unwrap();
2831 std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2832 let result = launch_plan(plan_of("coder"), options);
2833 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2834 let Err(failure) = result else {
2835 panic!("the launch was expected to be refused");
2836 };
2837 assert!(failure.stopped.is_empty(), "{failure:?}");
2838 match only_reason(&failure) {
2839 MemberFailureReason::NotStarted { error } => {
2840 assert!(error.contains(&taken.display().to_string()), "{error}")
2841 }
2842 other => panic!("expected NotStarted, got {other:?}"),
2843 }
2844 }
2845
2846 #[test]
2850 fn member_roots_are_the_records_groups_of_member_directories() {
2851 let home = tempfile::tempdir().unwrap();
2852 crate::murmur_home::run_with_home(
2853 "formation_launch::tests::member_roots_inner",
2854 home.path(),
2855 );
2856 }
2857
2858 #[test]
2859 #[ignore = "run by member_roots_are_the_records_groups_of_member_directories"]
2860 fn member_roots_inner() {
2861 if !crate::murmur_home::in_scratch_home() {
2862 return;
2863 }
2864 assert!(formation_member_roots(&FormationId::mint()).is_empty());
2865
2866 let id = FormationId::mint();
2867 let home = PathBuf::from(std::env::var_os("HOME").unwrap());
2868 let formations = home.join(".murmur").join(FORMATIONS_DIR);
2869 for member in ["worker", "reviewer", "idle"] {
2870 assert_eq!(
2871 make_member_workdir(&formations, &id, member).unwrap(),
2872 formations.join(id.as_str()).join(member)
2873 );
2874 }
2875 let group = |member: &str| {
2876 crate::records_dir::records_group(&formations.join(id.as_str()).join(member)).unwrap()
2877 };
2878 for member in ["worker", "reviewer"] {
2879 std::fs::create_dir_all(group(member)).unwrap();
2880 }
2881 assert_eq!(
2882 formation_member_roots_in(&formations, &id),
2883 [group("reviewer"), group("worker")]
2884 );
2885 assert!(group("worker").starts_with(home.join(".murmur/records")));
2886 }
2887
2888 #[test]
2890 fn formation_marker_round_trips_as_three_keys() {
2891 let marker = FormationMarker {
2892 project_dir: PathBuf::from("/srv/project"),
2893 launcher_pid: 4242,
2894 launcher_start: "12345".to_string(),
2895 };
2896 let value = serde_json::to_value(&marker).unwrap();
2897 assert_eq!(
2898 value,
2899 serde_json::json!({
2900 "project_dir": "/srv/project",
2901 "launcher_pid": 4242,
2902 "launcher_start": "12345",
2903 })
2904 );
2905 let dir = tempfile::tempdir().unwrap();
2906 std::fs::write(dir.path().join(FORMATION_MARKER_FILE), value.to_string()).unwrap();
2907 assert_eq!(FormationMarker::read(dir.path()), Some(marker));
2908 std::fs::write(dir.path().join(FORMATION_MARKER_FILE), "{").unwrap();
2909 assert_eq!(FormationMarker::read(dir.path()), None);
2910 assert_eq!(FormationMarker::read(&dir.path().join("missing")), None);
2911 }
2912
2913 #[test]
2917 fn formation_marker_is_written_owner_only_before_the_first_peer_directory() {
2918 use std::os::unix::fs::PermissionsExt;
2919
2920 let _guard = binary_guard();
2921 let dir = tempfile::tempdir().unwrap();
2922 let project = tempfile::tempdir().unwrap();
2923 let mut options = options(Duration::from_secs(1));
2924 options.project_dir = project.path().join(".");
2925 let formation_dir = options
2926 .formations_dir
2927 .clone()
2928 .unwrap()
2929 .join(options.formation_id.as_str());
2930 std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2931 let result = launch_plan(plan_of("coder"), options);
2932 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2933 assert!(result.is_err(), "the stand-in never reports itself");
2934
2935 let path = formation_dir.join(FORMATION_MARKER_FILE);
2936 let metadata = std::fs::metadata(&path).unwrap();
2937 assert_eq!(metadata.permissions().mode() & 0o777, 0o600);
2938 let marker = FormationMarker::read(&formation_dir).unwrap();
2939 let pid = std::process::id();
2940 assert_eq!(
2941 marker,
2942 FormationMarker {
2943 project_dir: std::fs::canonicalize(project.path()).unwrap(),
2944 launcher_pid: pid,
2945 launcher_start: crate::running::process_start_token(pid).unwrap_or_default(),
2946 }
2947 );
2948 let peer_dir = std::fs::metadata(formation_dir.join("coder")).unwrap();
2949 assert!(
2950 metadata.modified().unwrap() <= peer_dir.modified().unwrap(),
2951 "the marker was written after the peer's directory was made"
2952 );
2953 }
2954
2955 #[test]
2957 fn formation_marker_is_never_written_without_peers() {
2958 let _guard = binary_guard();
2959 let dir = tempfile::tempdir().unwrap();
2960 let options = options(Duration::from_secs(5));
2961 let formations = options.formations_dir.clone().unwrap();
2962 std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
2963 let plan = FormationPlan {
2964 peers: Vec::new(),
2965 entry: member("planner"),
2966 callees: Vec::new(),
2967 };
2968 let result = launch_plan(plan, options);
2969 std::env::remove_var(child_launch::MUR_BINARY_ENV);
2970 let formation = result.expect("a formation with no peers has nothing to wait for");
2971 drop(formation);
2972 assert!(
2973 std::fs::read_dir(&formations).map_or(true, |mut entries| entries.next().is_none()),
2974 "{}",
2975 formations.display()
2976 );
2977 }
2978
2979 fn marker_of(pid: u32, start: &str) -> FormationMarker {
2980 FormationMarker {
2981 project_dir: PathBuf::from("/srv/project"),
2982 launcher_pid: pid,
2983 launcher_start: start.to_string(),
2984 }
2985 }
2986
2987 fn reaped_pid() -> u32 {
2989 let mut child = std::process::Command::new("true").spawn().unwrap();
2990 let pid = child.id();
2991 child.wait().unwrap();
2992 pid
2993 }
2994
2995 fn running_record(pid: u32, formation: Option<&FormationId>) -> crate::running::RunningRecord {
2996 crate::running::RunningRecord {
2997 session_id: "ses_0199c4e2f1b7712a9d3e4f5061728394".to_string(),
2998 url: Some("127.0.0.1:1".to_string()),
2999 pid,
3000 process_start: crate::running::process_start_token(pid).unwrap_or_default(),
3001 capsule_name: "worker".to_string(),
3002 capsule_version: "0.1.0".to_string(),
3003 workdir: PathBuf::from("/tmp/worker"),
3004 records_dir: PathBuf::from("/tmp/worker").join("records"),
3005 outlives_launcher: false,
3006 started_at: "2026-01-01T00:00:00Z".to_string(),
3007 door_token: None,
3008 credentials: Default::default(),
3009 formation_id: formation.cloned(),
3010 formation_member: None,
3011 formation_entry: false,
3012 formation_lifeline: false,
3013 formation_launcher: None,
3014 spawned_by: None,
3015 unconfined: false,
3016 }
3017 }
3018
3019 #[test]
3022 fn prune_formations_liveness_reads_the_launcher_from_the_marker() {
3023 let id = FormationId::mint();
3024 let own = std::process::id();
3025 let token = crate::running::process_start_token(own).unwrap();
3026 assert!(formation_is_live(&id, &marker_of(own, &token), Some(&[])));
3027 assert!(!formation_is_live(
3028 &id,
3029 &marker_of(reaped_pid(), &token),
3030 Some(&[])
3031 ));
3032 }
3033
3034 #[test]
3038 fn prune_formations_liveness_reads_running_records_that_name_the_formation() {
3039 let id = FormationId::mint();
3040 let dead = marker_of(reaped_pid(), "1");
3041 let mut sleeper = std::process::Command::new("sleep")
3042 .arg("30")
3043 .spawn()
3044 .unwrap();
3045 let live = running_record(sleeper.id(), Some(&id));
3046 let other = running_record(sleeper.id(), Some(&FormationId::mint()));
3047 let unnamed = running_record(sleeper.id(), None);
3048 let gone = running_record(reaped_pid(), Some(&id));
3049
3050 assert!(formation_is_live(&id, &dead, Some(&[live])));
3051 assert!(!formation_is_live(
3052 &id,
3053 &dead,
3054 Some(&[other, unnamed, gone])
3055 ));
3056 let _ = sleeper.kill();
3057 let _ = sleeper.wait();
3058 }
3059
3060 #[test]
3063 fn prune_formations_liveness_treats_an_unlistable_running_directory_as_all_live() {
3064 let id = FormationId::mint();
3065 assert!(formation_is_live(&id, &marker_of(reaped_pid(), "1"), None));
3066
3067 let home = tempfile::tempdir().unwrap();
3068 let project = tempfile::tempdir().unwrap();
3069 let formations = home.path().join(FORMATIONS_DIR);
3070 let old = FormationId::parse(&format!("frm_{:012x}{:020x}", 1_000_000u64, 1)).unwrap();
3071 let old_dir = formations.join(old.as_str());
3072 std::fs::create_dir_all(old_dir.join("worker")).unwrap();
3073 let marker = FormationMarker {
3074 project_dir: std::fs::canonicalize(project.path()).unwrap(),
3075 ..marker_of(reaped_pid(), "1")
3076 };
3077 std::fs::write(
3078 old_dir.join(FORMATION_MARKER_FILE),
3079 serde_json::to_string(&marker).unwrap(),
3080 )
3081 .unwrap();
3082 let policy = murmur_artifact::TraceRetainConfig {
3083 max_sessions: Some(1),
3084 max_age_secs: Some(1),
3085 };
3086
3087 let running = home.path().join("running");
3089 std::fs::write(&running, "").unwrap();
3090 let current = FormationId::mint();
3091 assert!(prune_ended_formations_in(
3092 &formations,
3093 &running,
3094 ¤t,
3095 project.path(),
3096 &policy
3097 )
3098 .is_empty());
3099 assert!(old_dir.is_dir());
3100
3101 std::fs::remove_file(&running).unwrap();
3103 let pruned =
3104 prune_ended_formations_in(&formations, &running, ¤t, project.path(), &policy);
3105 assert_eq!(pruned.len(), 1, "{pruned:?}");
3106 assert_eq!(pruned[0].formation_id, old);
3107 assert_eq!(pruned[0].reason, crate::trace::RETENTION_REASON_MAX_AGE);
3108 assert!(!old_dir.exists());
3109 }
3110}