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