Skip to main content

capsule_runtime/
formation_launch.rs

1//! Launching a formation: every member of an admitted roster as a `mur run` subprocess of its own,
2//! peers first and the entry member last, and every member stopped once the entry member's process
3//! has ended.
4//!
5//! **The launcher supervises; it runs no session.** Each member, the entry member included, is a
6//! `mur run --capsule <capsule> --capsule-version <version>` process this module starts and owns.
7//! The launcher waits on the entry member's process, so the peers are stopped however the entry
8//! member ends — its task finishing, its own `exit(143)`, or a `SIGKILL` of its process.
9//!
10//! **Every member holds a lifeline.** Each member is handed the read end of a pipe of its own
11//! ([`crate::lifeline`]), whose write end only the launcher holds and never writes to. A member
12//! that reads EOF winds down as a first `SIGTERM` would wind it down. The launcher ends a
13//! formation by closing lifelines; when the launcher itself dies — `SIGKILL` and the OOM killer
14//! included — the kernel closes them, so no member outlives its formation.
15//!
16//! **Ready means the door answers as the session that reported it.** A peer is ready when the door
17//! at the URL its readiness line reported serves an agent card naming the session id that line
18//! reported, read through [`crate::running`]'s door probe. A live pid proves nothing, and a door
19//! some other process holds on the reported port names another session. Every member has one
20//! deadline, counted from its spawn.
21//!
22//! **All or nothing.** A peer that fails to come up refuses the whole launch: every member already
23//! started is stopped and reaped before the refusal is returned, and the entry member is never
24//! started.
25//!
26//! **The launcher is the formation's one principal.** It generates the formation's signing key
27//! ([`FormationAuthority`]) after the id and holds it in memory alone. Each member gets a pipe of
28//! its own, whose read end only that member's `mur run` inherits, named by
29//! [`FORMATION_CHANNEL_ENV`]. Its first line — written before the member is spawned — carries the
30//! formation id, the member's name, the verification key and one token per member it may call.
31//! Once every peer's door answers, each peer that calls other peers is written one address line
32//! naming their doors; the entry member's first line and address line are both written before it is
33//! spawned. Nothing here writes a token to stdout, stderr, a file, argv or an environment
34//! variable. Callees never include the entry member: it runs the formation's own task from launch
35//! until the formation ends and serves no peer, and admission never admits an edge into it. The
36//! key and the write ends are dropped once every member is reaped.
37//!
38//! **Teardown.**
39//!
40//! * One owner holds every started member, and every return path stops and reaps them. Its `Drop`
41//!   does the same, so an early return or an unwinding panic leaves nothing running.
42//! * Stopping the members is closing every lifeline, up to [`MEMBER_STOP_GRACE`] for each to end
43//!   in order, `SIGKILL` to the process group of any still running, and a reap. The launcher sends
44//!   no member `SIGTERM`. Members are waited on concurrently.
45//! * Every lifeline is created before the first member is spawned, and the launcher spawns
46//!   nothing but members, so no process but the launcher ever holds a write end.
47//! * Peers lead process groups of their own, so a terminal's `^C` reaches only the launcher and
48//!   the entry member, and the launcher stops the peers in order. The entry member stays in the
49//!   launcher's group and behaves at a terminal as a hand-run `mur run` does.
50//! * Under `panic = "abort"` no destructor runs, so a panic hook — installed once per process —
51//!   `SIGKILL`s every registered member's process group before the abort. It reads a lock-free
52//!   registry of member pids, never a mutex a panicking thread might hold.
53//! * [`catch_launcher_signals`] turns `SIGINT`, `SIGTERM` and `SIGHUP` into a flag every wait here
54//!   polls, so a signalled launcher stops its members — the entry member through its lifeline,
55//!   like every other — before it exits.
56
57use 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
76/// How long a member has, from its spawn, to report itself and have its door answer.
77///
78/// The bound a delegated child gets for its readiness line, for the same reason: it bounds one
79/// `mur run` subprocess becoming usable on a cold or loaded host, where compiling the driver
80/// component is the slowest step.
81pub const MEMBER_READY_TIMEOUT: Duration = Duration::from_secs(180);
82
83/// How long a member has, after its lifeline is closed, to end in order before its process group
84/// is killed.
85///
86/// Longer than the runtime's own teardown deadline, so a member whose session is ending in order
87/// finishes before it is killed and its trace stays readable.
88pub const MEMBER_STOP_GRACE: Duration = Duration::from_secs(25);
89
90/// How often a peer's door is probed while it is not yet ready.
91const READY_PROBE_INTERVAL: Duration = Duration::from_millis(100);
92
93/// How often a wait here looks at a member's process, at the launcher's signal flag, and at
94/// whether another peer has already failed.
95const POLL_INTERVAL: Duration = Duration::from_millis(50);
96
97/// How long a member whose stdout has closed is given to exit on its own before it is ended.
98const EXIT_AFTER_EOF_WAIT: Duration = Duration::from_secs(2);
99
100/// The address every peer binds its door on. Members of one formation share a host.
101pub const PEER_BIND_ADDR: &str = "127.0.0.1";
102
103/// The hidden `mur run` flag carrying the sha256 admission bound a member to.
104pub const CAPSULE_SHA256_FLAG: &str = "--capsule-sha256";
105
106/// The hidden `mur run` flag naming the directory a peer resolves its artifacts, its lock and its
107/// workspace from, apart from its `--workdir`.
108pub const STORE_ROOT_FLAG: &str = "--store-root";
109
110/// The directory under the murmur home that holds every formation's peer directories.
111pub const FORMATIONS_DIR: &str = "formations";
112
113/// The ownership marker in `<murmur home>/formations/<frm_id>/`, a [`FormationMarker`]. Member
114/// names start with a letter and hold no `.`, so it never collides with a member's directory.
115pub const FORMATION_MARKER_FILE: &str = "formation.json";
116
117/// What a formation launch is given besides its roster.
118#[derive(Debug, Clone)]
119pub struct FormationLaunchOptions {
120    /// The id every member is launched with, minted once for this launch.
121    pub formation_id: FormationId,
122    /// The roster's project directory: the entry member's `--workdir`, and every peer's
123    /// `--store-root`, so every member resolves from the stores admission read.
124    pub project_dir: PathBuf,
125    /// `--task`, passed to the entry member only.
126    pub task: Option<String>,
127    /// `--json`, passed to the entry member only; peers always run with it.
128    pub json: bool,
129    /// `--verbose`, passed to the entry member only.
130    pub verbose: bool,
131    /// `--no-env-file`, passed to every member.
132    pub no_env_file: bool,
133    /// `--containment <class>`, passed to every member.
134    pub containment: Option<String>,
135    /// Each member's deadline to become ready, from its spawn. [`MEMBER_READY_TIMEOUT`] in
136    /// production.
137    pub ready_timeout: Duration,
138    /// How long a stopped member has after its lifeline is closed. [`MEMBER_STOP_GRACE`] in
139    /// production.
140    pub stop_grace: Duration,
141    /// Where peers' directories are made in place of `<murmur home>/formations`. `None` in
142    /// production; set by tests that must not write under the developer's home.
143    pub(crate) formations_dir: Option<PathBuf>,
144}
145
146impl FormationLaunchOptions {
147    /// Options with the production deadline and grace, and nothing passed through.
148    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/// One member as the launcher starts it.
165#[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    /// The sha256 admission bound the member to, handed to its `mur run` to check.
171    pub(crate) sha256: String,
172}
173
174/// What a launch starts: the peers in roster order, the entry member, and the members each member
175/// may call.
176#[derive(Debug, Clone)]
177pub(crate) struct FormationPlan {
178    pub(crate) peers: Vec<PlannedMember>,
179    pub(crate) entry: PlannedMember,
180    /// Every member's callees in roster order. Admission never admits an edge into the entry
181    /// member, so it is never among them. A member with none is absent.
182    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    /// The members `member` may call, in roster order.
218    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    /// `member`'s channel first line, signed by `authority` and naming `launcher` as the process
227    /// that started it.
228    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
241/// A peer whose door answered as the session its readiness line named.
242pub struct ReadyPeer {
243    pub name: String,
244    pub capsule: String,
245    pub version: String,
246    pub pid: u32,
247    pub session_id: String,
248    /// `http://host:port`.
249    pub url: String,
250    /// The peer's accessible directory, `<murmur home>/formations/<frm_id>/<member>`: its
251    /// `--workdir`, where its `task.md` is written and its sessions nest.
252    pub workdir: PathBuf,
253    /// The operator token from the peer's readiness line, present when the peer declares
254    /// `network.authentication`. Held for the readiness probe and never printed.
255    door_token: Option<DoorToken>,
256}
257
258impl ReadyPeer {
259    /// The operator token for this peer's door, or `None` for a public door.
260    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/// Why one member did not come up.
284#[derive(Debug, Clone, PartialEq, Eq)]
285pub enum MemberFailureReason {
286    /// Its process could not be started.
287    NotStarted { error: String },
288    /// Its process ended before it printed a readiness line.
289    ExitedBeforeReporting {
290        status: String,
291        stderr_tail: Vec<String>,
292    },
293    /// Its process reported and then ended before its door answered.
294    ExitedBeforeDoorAnswered {
295        status: String,
296        stderr_tail: Vec<String>,
297    },
298    /// No readiness line within the deadline.
299    DidNotReport { deadline: Duration },
300    /// The first line it printed is not a readiness line.
301    UnreadableReport { why: String },
302    /// Its readiness line names no door to probe.
303    ReportedNoDoor,
304    /// Its readiness line names another formation, or none.
305    WrongFormation {
306        reported: Option<String>,
307        expected: String,
308    },
309    /// Its door never answered as the session it reported, within the deadline.
310    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/// One member that did not come up, and why.
378#[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/// A member whose process was still running when the launcher stopped it.
387#[derive(Debug, Clone, PartialEq, Eq)]
388pub struct StoppedMember {
389    pub name: String,
390    pub pid: u32,
391}
392
393/// A launch that did not complete. By the time this is returned, nothing it started is running.
394#[derive(Debug, Clone, PartialEq, Eq)]
395pub struct MemberLaunchFailure {
396    /// Every member known to have failed, in roster order. Empty when the launch was interrupted
397    /// by a signal before any member failed.
398    pub failed: Vec<MemberFailure>,
399    /// Every member that had started and was stopped, in roster order.
400    pub stopped: Vec<StoppedMember>,
401    /// The signal the launcher caught while members were coming up, when that is what ended the
402    /// launch.
403    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
427/// `name (pid N), …`, or `none`.
428pub 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
439/// The conventional name of the three signals a launcher catches.
440pub 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/// How the entry member's process, and with it the formation, ended.
450#[derive(Debug, Clone, PartialEq, Eq)]
451pub enum FormationEnding {
452    /// The entry member's process ended on its own, or by a signal the launcher did not send.
453    EntryExited(ExitStatus),
454    /// The launcher caught this signal and stopped the formation, the entry member included.
455    Signalled(i32),
456}
457
458/// A formation that has ended, every member stopped and reaped.
459#[derive(Debug, Clone)]
460pub struct FormationExit {
461    pub ending: FormationEnding,
462    /// Every member still running when the formation was stopped, in roster order.
463    pub stopped: Vec<StoppedMember>,
464}
465
466impl FormationExit {
467    /// The launcher's exit status: the entry member's code, or 128 plus the signal that ended it,
468    /// or 128 plus the signal the launcher caught.
469    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
477/// `status N` or `signal N`, for a refusal.
478fn 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
492// ── The launcher's signals ────────────────────────────────────────────────────
493
494/// The signal the launcher caught, or 0. Written only by [`record_signal`].
495static CAUGHT_SIGNAL: AtomicI32 = AtomicI32::new(0);
496
497#[cfg(unix)]
498extern "C" fn record_signal(signal: libc::c_int) {
499    // An atomic store is async-signal-safe; nothing else happens in the handler.
500    CAUGHT_SIGNAL.store(signal, Ordering::SeqCst);
501}
502
503/// Catch `SIGINT`, `SIGTERM` and `SIGHUP` for the rest of this process's life, so a formation
504/// launcher stops its members before it exits rather than dying with them running.
505///
506/// Every wait in this module polls the flag the handler sets. Called by the launcher only: a test
507/// harness that launched formations in-process would otherwise stop answering `^C`.
508#[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        // SAFETY: `sigaction` is given a zeroed, then fully initialised, struct whose handler is
513        // an `extern "C"` function that only stores to an atomic, which is async-signal-safe. The
514        // old-action pointer is null, so nothing is written back.
515        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
534/// The signal [`catch_launcher_signals`] caught, if any.
535pub fn caught_signal() -> Option<i32> {
536    match CAUGHT_SIGNAL.load(Ordering::SeqCst) {
537        0 => None,
538        signal => Some(signal),
539    }
540}
541
542// ── The member registry the panic hook reads ──────────────────────────────────
543
544/// How many members one process can have registered at once. A slot that cannot be had leaves
545/// that member to the owner's `Drop` alone; nothing else depends on the registry.
546const REGISTRY_SLOTS: usize = 1024;
547
548/// Every running member's pid: positive for a member leading its own process group, which the
549/// hook kills as a group; negative for the entry member, which shares the launcher's group and is
550/// killed alone; 0 for a free slot.
551static MEMBER_REGISTRY: [AtomicI64; REGISTRY_SLOTS] = [const { AtomicI64::new(0) }; REGISTRY_SLOTS];
552
553/// One registered member. Releasing it — on reap, or on drop — frees the slot.
554struct 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
580/// `SIGKILL` every registered member: a member leading its own group as a group, the entry member
581/// alone. Reads atomics only, so a panic hook can call it whatever lock the panicking thread holds.
582pub(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
592/// Install, once per process, the hook that kills every registered member before a
593/// `panic = "abort"` build aborts. An unwinding build runs the owner's `Drop` instead, which stops
594/// members in order.
595fn 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    // SAFETY: `kill` takes two integers and dereferences nothing. `pid` is positive, so it names
618    // one process, never a group or every process.
619    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    // SAFETY: `killpg` takes two integers and dereferences nothing. `pgid` is greater than 1, so
634    // it names one process group, never the caller's own or every process.
635    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
646// ── One member's process ──────────────────────────────────────────────────────
647
648/// A started member's process, owned here and nowhere else.
649struct MemberProcess {
650    name: String,
651    pid: u32,
652    child: Child,
653    own_group: bool,
654    /// The launcher's side of the member's lifeline: the write end, open until the member is
655    /// stopped.
656    #[cfg(unix)]
657    lifeline: Option<crate::lifeline::MemberLifeline>,
658    /// The exit status, once reaped. A reaped pid may already belong to another process, so
659    /// nothing signals or probes it after this is set.
660    status: Option<ExitStatus>,
661    slot: RegistrySlot,
662    grace: Duration,
663    /// The write end of the member's formation channel, kept open while the member runs so an
664    /// address line can follow its first line, and closed once it is reaped.
665    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    /// Write one line to the member's channel. A member that has already gone has closed its end;
693    /// the line then reaches nobody, which is what it is for.
694    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    /// Whether the process has ended, reaped or not.
701    fn has_exited(&self) -> bool {
702        self.status.is_some() || !crate::running::pid_is_alive(self.pid)
703    }
704
705    /// The exit status, reaping the process if it has ended. `None` while it runs.
706    ///
707    /// A member leading its own group has the group killed first, while its unreaped process
708    /// still holds the group id, so nothing it started outlives it and no recycled id is signalled.
709    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            // `wait` failing leaves nothing further to ask; the process is not ours to probe.
728            self.status = Some(ExitStatus::default());
729        }
730        self.slot.release();
731        self.channel = None;
732    }
733
734    /// Close the member's lifeline: it reads EOF and winds down. Idempotent.
735    fn close_lifeline(&mut self) {
736        #[cfg(unix)]
737        if let Some(lifeline) = &mut self.lifeline {
738            lifeline.close();
739        }
740    }
741
742    /// Whether the process is running and not yet reaped.
743    fn is_running(&self) -> bool {
744        self.status.is_none() && !self.has_exited()
745    }
746
747    /// Close its lifeline, up to the grace for it to end, `SIGKILL` to its group, and a reap.
748    /// Idempotent. `Some` when the process was still running when the stop began.
749    fn stop(&mut self) -> Option<StoppedMember> {
750        let was_running = self.is_running();
751        self.stop_from(was_running)
752    }
753
754    /// [`Self::stop`], with whether the process was running when the stop began read by the
755    /// caller: a member can end on its lifeline's EOF the moment that lifeline closes.
756    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
781/// Every started member, in roster order: the one owner, and what every exit path stops.
782struct MemberSet(Vec<MemberProcess>);
783
784impl MemberSet {
785    /// Close every member's lifeline, then wait on, kill if need be, and reap every member
786    /// concurrently. Idempotent.
787    fn stop_all(&mut self) -> Vec<StoppedMember> {
788        // Read before any lifeline closes: a member that ends on its EOF while another lifeline is
789        // still being closed was running when the stop began, and is reported as stopped.
790        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
815// ── Starting members ──────────────────────────────────────────────────────────
816
817/// The arguments every member's `mur run` starts with, running in `workdir`.
818fn 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
832/// The pass-through flags every member gets.
833fn 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
843/// A peer's arguments after the binary: it runs in `workdir`, a directory of its own, and resolves
844/// from the project's stores. Its readiness line carries its tokens, which this launcher reads on
845/// the peer's stdout pipe for the door probe.
846pub(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
864/// The entry member's arguments after the binary. It runs in the project directory, where its
865/// `--task` is written. Never [`child_launch::READINESS_TOKENS_FLAG`]: the entry's output is the
866/// operator's.
867pub(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
889/// `<murmur home>/formations`, resolved without creating anything.
890fn formations_dir() -> Result<PathBuf, String> {
891    Ok(crate::state_store::murmur_home_dir()?.join(FORMATIONS_DIR))
892}
893
894/// Every records group a formation's peers record under: the records group of each directory
895/// under `<murmur home>/formations/<frm_id>/`, sorted, for each one that exists.
896///
897/// Reads the directory and creates nothing. A formation with no directory, or a `HOME` that does
898/// not resolve, has no roots. A formation directory that exists and cannot be listed is returned
899/// as a root itself, so the caller's read of it fails and is reported as an unreadable root.
900pub 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
907/// [`formation_member_roots`] under `formations_dir` in place of `<murmur home>/formations`. Each
908/// member's records group is still resolved under this host's `~/.murmur/records`.
909pub 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/// Who launched a formation, from which project: the contents of [`FORMATION_MARKER_FILE`].
930///
931/// The launcher writes it once, before the first peer's directory is made, and nothing rewrites
932/// it. Retention reads it to decide whether a formation directory is this project's to remove and
933/// whether the launcher that made it is still running; a directory without one is never removed.
934#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
935pub struct FormationMarker {
936    /// The roster's project directory, canonicalized when the launcher could, as given otherwise.
937    pub project_dir: PathBuf,
938    /// The launcher's pid.
939    pub launcher_pid: u32,
940    /// The launcher's start token, read as [`crate::running::process_start_token`] reads it, or
941    /// `""` when it could not be read.
942    pub launcher_start: String,
943}
944
945impl FormationMarker {
946    /// The marker for a formation this process launches from `project_dir`.
947    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    /// The marker in `formation_dir`, or `None` when it is missing or does not parse.
957    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
963/// `project_dir` canonicalized, or as given when it cannot be: the one form a marker records and
964/// the one ownership is compared in.
965pub(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
969/// Write this launch's [`FormationMarker`] into `<formations dir>/<frm_id>/`, holding both
970/// directories owner-only first. The marker itself is written at `0600`.
971fn 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
986/// Remove the ended formation directories of `project_dir` that the entry member's `policy` does
987/// not keep, and return what went.
988///
989/// Called by the launcher once every member of `current` has been stopped and reaped. `current`
990/// and every formation minted after it are never removed; see [`crate::retention::prune_formations`]
991/// for what else never is. A `HOME` that does not resolve removes nothing.
992pub 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
1005/// [`prune_ended_formations`] over `formations_dir` and `running_dir` in place of the murmur
1006/// home's.
1007pub 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    // One scan for the whole pass, and only a scan: nothing here unlinks a record.
1015    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
1028/// Whether any part of formation `id` may still be running: its launcher is not gone, or a running
1029/// record names `id` and its process is not gone.
1030///
1031/// `records` is `None` when the running directory could not be listed, and then every formation
1032/// is live. [`crate::running::ProcessState::Unverified`] is live: a reading that failed says
1033/// nothing about the process.
1034fn 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
1051/// Make `member`'s directory under `formations_dir`, owner-only, and return it.
1052///
1053/// `formations_dir` and the formation's directory are held at `0700` whether or not this call made
1054/// them. The member's own directory is made with a single non-recursive create at `0700`, so a
1055/// path that already exists — a directory, a file or a symlink — is refused rather than reused:
1056/// nothing another process placed there becomes a member's preopen.
1057fn 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    // The umask may have narrowed the create's mode further; it can never have widened it.
1082    Ok(workdir)
1083}
1084
1085/// The directory peers' directories are made under for this launch: the override, or
1086/// `<murmur home>/formations` with the home held owner-only first.
1087fn 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
1097/// What became of one peer's start.
1098enum PeerOutcome {
1099    Ready(ReadyPeer),
1100    Failed(MemberFailureReason),
1101    /// Given up on because another peer failed or the launcher was signalled.
1102    Abandoned,
1103}
1104
1105/// A member's formation channel: the write end the launcher keeps, already holding the member's
1106/// first line, and the read end the member's process inherits.
1107struct Channel {
1108    writer: std::io::PipeWriter,
1109    reader: std::io::PipeReader,
1110}
1111
1112impl Channel {
1113    /// A fresh pipe with `first_line` already written into it. Both ends are close-on-exec, so no
1114    /// process inherits either unless [`ProcessLaunch::inherit_fd`] names it.
1115    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    /// The descriptor number the member reads its channel from.
1132    fn fd(&self) -> i32 {
1133        use std::os::fd::AsRawFd;
1134        self.reader.as_raw_fd()
1135    }
1136
1137    /// The environment every member is started with: its formation id, its channel, and
1138    /// [`crate::host_warnings::HOST_WARNINGS_REPORTED_ENV`], because the launcher prints the
1139    /// host-level warnings once before any member starts.
1140    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
1150/// Start one peer and wait until its door answers, it fails, or `abort` is raised.
1151///
1152/// The process, when one was started, is returned whatever the outcome, so the caller owns
1153/// stopping it.
1154fn 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    // The member holds its own copy of the read end now, or never will.
1194    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    // The readiness line.
1234    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                // Its stdout closes as it exits, so the exit itself is at most a moment behind. A
1251                // process that closed stdout and stayed up is as good as gone, and is ended.
1252                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    // The door.
1281    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
1333/// Whether a door probe's answer counts as the reported session being ready: `Ok` only when the
1334/// card names `reported`. A card naming another session, or a failed probe, is the `Err` that
1335/// becomes the refusal's `last_error` if the deadline passes.
1336fn 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
1344/// What a peer's readiness line says about its door.
1345struct Report {
1346    session_id: String,
1347    /// `http://host:port`.
1348    url: String,
1349    door_token: Option<DoorToken>,
1350}
1351
1352/// Read a readiness line, refusing one that names another formation or no door.
1353fn 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
1388// ── The launch ────────────────────────────────────────────────────────────────
1389
1390/// Start every peer of `roster` at once and wait for each door to answer.
1391///
1392/// On success the peers are running and ready, and the entry member is not yet started: the caller
1393/// announces the formation, then calls [`RunningFormation::start_entry`]. On failure nothing this
1394/// started is still running.
1395pub 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    // One per peer and one for the entry member, all before the first spawn: where the
1439    // close-on-exec flag is set after `pipe` returns, a member forked in between would inherit
1440    // another member's write end and keep that member's formation alive.
1441    #[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    // Read once, so every member, the entry member included, records the same launcher. A
1466    // launcher that cannot read its own start time names none, and `mur stop` then cannot reach
1467    // it through its members.
1468    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    // Before any peer's directory exists, so no formation directory a launch made is without one.
1475    // A launch with no peers makes no formation directory, and so no marker.
1476    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    // Every peer's door answers, so every address is known: each peer that calls other peers is
1564    // told where they are.
1565    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
1588/// The door of each of `callees` among the ready peers, in roster order.
1589fn 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/// `count` new lifelines, or the first error creating one.
1601#[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
1623/// A formation whose peers are ready: the one owner of every member it started, and of the
1624/// formation's signing key.
1625///
1626/// Dropping it stops and reaps every member, then drops the key: nothing can mint for this
1627/// formation again.
1628pub struct RunningFormation {
1629    formation_id: FormationId,
1630    entry: PlannedMember,
1631    plan: FormationPlan,
1632    peers: Vec<ReadyPeer>,
1633    /// Declared before `authority`, so the members are stopped and reaped before the key is
1634    /// dropped.
1635    members: MemberSet,
1636    /// The entry member's position in `members`, once started.
1637    entry_index: Option<usize>,
1638    /// The entry member's lifeline, created with the peers' and held here until the entry
1639    /// member is started with it.
1640    #[cfg(unix)]
1641    entry_lifeline: Option<crate::lifeline::MemberLifeline>,
1642    binary: PathBuf,
1643    options: FormationLaunchOptions,
1644    authority: FormationAuthority,
1645    /// This process, as every member's first line names it.
1646    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    /// The entry member's roster name.
1665    pub fn entry_name(&self) -> &str {
1666        &self.entry.name
1667    }
1668
1669    /// Every peer, ready, in roster order.
1670    pub fn peers(&self) -> &[ReadyPeer] {
1671        &self.peers
1672    }
1673
1674    /// The members `member` is handed a credential and an address for, in roster order: its
1675    /// roster callees.
1676    pub fn callees_of(&self, member: &str) -> &[String] {
1677        self.plan.callees_of(member)
1678    }
1679
1680    /// The entry member's pid, once started.
1681    pub fn entry_pid(&self) -> Option<u32> {
1682        self.entry_index.map(|index| self.members.0[index].pid)
1683    }
1684
1685    /// Start the entry member, last, with its standard output and error this process's own.
1686    ///
1687    /// On failure every peer has been stopped and reaped.
1688    pub fn start_entry(&mut self) -> Result<u32, MemberLaunchFailure> {
1689        let args = entry_args(&self.entry, &self.options);
1690        // Its whole channel is written before it starts: every address it may need is known.
1691        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    /// Wait for the entry member's process to end, or for the launcher to be signalled, then stop
1779    /// and reap every member: every lifeline closed, the entry member's included.
1780    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    /// How long a launch that is expected to succeed may take before the test gives up on it: a
1809    /// guard against a launch that never completes on a loaded host, not a claim about how fast
1810    /// one does.
1811    const LIVENESS_BOUND: Duration = Duration::from_secs(120);
1812
1813    /// The host's 1-, 5- and 15-minute load averages, for failure messages that may be load.
1814    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    /// The production bounds, stated: the readiness deadline a delegated child also gets, and a
1824    /// grace that outlasts the runtime's own teardown deadline.
1825    #[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    /// Every member, peer or entry, is started told its launcher already printed the host-level
1842    /// warnings.
1843    #[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        // Peers' directories go under a scratch directory, never the developer's home.
1861        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    /// Serializes the tests that point `MURMUR_MUR_BINARY` at a stand-in; the variable is
1929    /// process-wide.
1930    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    /// A stand-in `mur` that prints `line` (when given), records any `SIGTERM` it is sent in
1937    /// `dir/sigterm`, and runs until its lifeline reads EOF, whatever it is asked to run.
1938    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        // Through `/dev/fd`: a POSIX shell's `<&` takes a single-digit descriptor only, and a
1948        // lifeline is numbered wherever `pipe` put it.
1949        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    /// Launch one peer from a stand-in printing `line`, returning the refusal and the stand-in's
1969    /// pid as the refusal named it.
1970    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    /// A port with nothing listening on it: claimed, and never bound.
2009    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            // One deadline covers the readiness line and the door, so a stand-in that is not
2033            // scheduled in time is refused before its door is ever probed.
2034            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    /// A one-route door on `127.0.0.1`, served forever: each connection's request is read, then
2051    /// `GET /.well-known/agent-card.json` is answered with a card naming the session `answer`
2052    /// returns for that request, or with `{}` when it returns `None`.
2053    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    /// A door answering with a card naming `session_id` from `answer_after` on, and with a card
2084    /// naming no session before then.
2085    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    /// How many card requests [`door_naming_another_session_first`] answers as another session.
2091    const WRONG_ANSWERS: usize = 3;
2092
2093    /// A door that answers as `ses_somebody_else` to its first [`WRONG_ANSWERS`] card requests and
2094    /// as `ses_standin` to every one after, with the count of requests it has read. The count is
2095    /// raised before the answer is written, so it covers every answer a client has seen.
2096    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    /// The launcher probes past every answer naming another session and is ready only on the one
2110    /// naming the reported session. Accepting any answer would stop at the door's first request.
2111    #[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    /// A member that does not end on EOF is killed with its process group once the grace has run.
2279    #[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    /// Every member reads exactly its own bundle on the descriptor `MURMUR_FORMATION_CHANNEL`
2314    /// names: a token per callee and nothing about anyone else, then — once every peer is ready —
2315    /// its callees' doors. The entry member gets both lines before it starts, and an edge into it
2316    /// is served to nobody. No token is in any member's argv or environment.
2317    #[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            // `reviewer` calls no one, so it is absent; no plan has a callee that is the entry
2359            // member, because admission never admits an edge into it.
2360            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        // Every member, the entry member included, is told the one launcher that started it:
2402        // this test process.
2403        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        // The sweep kills every member this process registered, so no other launch may be live.
2454        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    /// A peer's directory is `<home>/formations/<frm_id>/<member>`, every level owner-only.
2520    #[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    /// Nothing already at a member's path becomes its preopen: a directory, a file or a symlink
2536    /// there refuses the member, naming the path.
2537    #[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    /// A peer whose directory cannot be made is not started, and the launch is refused naming it.
2552    #[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    /// A member's directory is `formations/<frm_id>/<member>`, and the roots `mur trace show
2576    /// frm_<id>` searches are the records group of each member directory that has one, sorted,
2577    /// and none for a formation with no directory.
2578    #[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    /// The marker is three keys, written and read back unchanged.
2618    #[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    /// A launch with peers writes its marker at `0600` into its formation's directory before the
2643    /// first peer's directory is made, naming the canonical project directory and this process as
2644    /// its launcher — and writes it even when the launch is then refused.
2645    #[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    /// A plan with no peers makes no formation directory, so there is nowhere for a marker.
2685    #[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    /// A pid this test started and has already reaped: no process holds it.
2717    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    /// A formation is live while its launcher is — this process with its own start token — and
2747    /// not once its launcher's pid is held by nothing.
2748    #[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    /// A formation whose launcher is gone is live while a running record names it and that
2762    /// record's process is alive; a record naming another formation, or a gone process, is not
2763    /// enough.
2764    #[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    /// A running directory that cannot be listed says nothing about what is running, so every
2788    /// formation is live and nothing is removed.
2789    #[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        // A regular file where the running directory should be cannot be listed.
2815        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            &current,
2822            project.path(),
2823            &policy
2824        )
2825        .is_empty());
2826        assert!(old_dir.is_dir());
2827
2828        // A running directory that does not exist is an empty machine, and the ended one goes.
2829        std::fs::remove_file(&running).unwrap();
2830        let pruned =
2831            prune_ended_formations_in(&formations, &running, &current, 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}