use std::io::Write;
use std::path::{Path, PathBuf};
use std::process::{Child, ExitStatus};
use std::sync::atomic::{AtomicBool, AtomicI32, AtomicI64, Ordering};
use std::sync::{mpsc, Once};
use std::time::{Duration, Instant};
use crate::child_launch::{self, AfterReadiness, ProcessEnv, ProcessLaunch, StartedProcess};
use crate::delegation::SPAWNER_ENV;
use crate::door_auth::DoorToken;
use crate::formation::{FormationId, FormationPeer, FORMATION_PEERS_ENV};
use crate::formation_credentials::{
render_address_line, FormationAuthority, FORMATION_CHANNEL_ENV,
};
use crate::lifeline::SPAWNER_LIFELINE_ENV;
use crate::roster::AdmittedRoster;
use crate::running::ProcessIdentity;
use serde::{Deserialize, Serialize};
pub const MEMBER_READY_TIMEOUT: Duration = Duration::from_secs(180);
pub const MEMBER_STOP_GRACE: Duration = Duration::from_secs(25);
const READY_PROBE_INTERVAL: Duration = Duration::from_millis(100);
const POLL_INTERVAL: Duration = Duration::from_millis(50);
const EXIT_AFTER_EOF_WAIT: Duration = Duration::from_secs(2);
pub const PEER_BIND_ADDR: &str = "127.0.0.1";
pub const CAPSULE_SHA256_FLAG: &str = "--capsule-sha256";
pub const STORE_ROOT_FLAG: &str = "--store-root";
pub const FORMATIONS_DIR: &str = "formations";
pub const FORMATION_MARKER_FILE: &str = "formation.json";
#[derive(Debug, Clone)]
pub struct FormationLaunchOptions {
pub formation_id: FormationId,
pub project_dir: PathBuf,
pub task: Option<String>,
pub json: bool,
pub verbose: bool,
pub no_env_file: bool,
pub containment: Option<String>,
pub ready_timeout: Duration,
pub stop_grace: Duration,
pub(crate) formations_dir: Option<PathBuf>,
}
impl FormationLaunchOptions {
pub fn new(formation_id: FormationId, project_dir: PathBuf) -> Self {
Self {
formation_id,
project_dir,
task: None,
json: false,
verbose: false,
no_env_file: false,
containment: None,
ready_timeout: MEMBER_READY_TIMEOUT,
stop_grace: MEMBER_STOP_GRACE,
formations_dir: None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct PlannedMember {
pub(crate) name: String,
pub(crate) capsule: String,
pub(crate) version: String,
pub(crate) sha256: String,
}
#[derive(Debug, Clone)]
pub(crate) struct FormationPlan {
pub(crate) peers: Vec<PlannedMember>,
pub(crate) entry: PlannedMember,
pub(crate) callees: Vec<(String, Vec<String>)>,
}
impl FormationPlan {
pub(crate) fn from_roster(roster: &AdmittedRoster) -> Self {
let planned = |member: &crate::roster::AdmittedMember| PlannedMember {
name: member.name.clone(),
capsule: member.capsule.clone(),
version: member.version.clone(),
sha256: member.sha256.clone(),
};
let entry = roster.entry();
Self {
peers: roster
.members()
.iter()
.filter(|member| !member.entry)
.map(planned)
.collect(),
entry: planned(entry),
callees: roster
.members()
.iter()
.map(|member| {
let callees: Vec<String> = roster
.callees(&member.name)
.map(|callee| callee.name.clone())
.collect();
(member.name.clone(), callees)
})
.filter(|(_, callees)| !callees.is_empty())
.collect(),
}
}
pub(crate) fn callees_of(&self, member: &str) -> &[String] {
self.callees
.iter()
.find(|(name, _)| name == member)
.map(|(_, callees)| callees.as_slice())
.unwrap_or_default()
}
fn first_line(
&self,
authority: &FormationAuthority,
launcher: Option<&ProcessIdentity>,
member: &str,
) -> String {
let callees: Vec<&str> = self.callees_of(member).iter().map(String::as_str).collect();
let mut bundle = authority.member_bundle(member, &callees);
bundle.launcher = launcher.cloned();
bundle.render_line()
}
}
pub struct ReadyPeer {
pub name: String,
pub capsule: String,
pub version: String,
pub pid: u32,
pub session_id: String,
pub url: String,
pub workdir: PathBuf,
door_token: Option<DoorToken>,
}
impl ReadyPeer {
pub fn door_token(&self) -> Option<&DoorToken> {
self.door_token.as_ref()
}
}
impl std::fmt::Debug for ReadyPeer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ReadyPeer")
.field("name", &self.name)
.field("capsule", &self.capsule)
.field("version", &self.version)
.field("pid", &self.pid)
.field("session_id", &self.session_id)
.field("url", &self.url)
.field("workdir", &self.workdir)
.field(
"door_token",
&self.door_token.as_ref().map(|_| "<redacted>"),
)
.finish()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MemberFailureReason {
NotStarted { error: String },
ExitedBeforeReporting {
status: String,
stderr_tail: Vec<String>,
},
ExitedBeforeDoorAnswered {
status: String,
stderr_tail: Vec<String>,
},
DidNotReport { deadline: Duration },
UnreadableReport { why: String },
ReportedNoDoor,
WrongFormation {
reported: Option<String>,
expected: String,
},
DoorDidNotAnswer {
url: String,
session_id: String,
deadline: Duration,
last_error: String,
},
}
impl std::fmt::Display for MemberFailureReason {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let tail = |f: &mut std::fmt::Formatter<'_>, lines: &[String]| -> std::fmt::Result {
if lines.is_empty() {
return f.write_str("; it wrote nothing to stderr");
}
f.write_str("; its last stderr lines were:")?;
for line in lines {
write!(f, "\n {line}")?;
}
Ok(())
};
match self {
Self::NotStarted { error } => write!(f, "its process could not be started: {error}"),
Self::ExitedBeforeReporting {
status,
stderr_tail,
} => {
write!(f, "its process exited with {status} before reporting")?;
tail(f, stderr_tail)
}
Self::ExitedBeforeDoorAnswered {
status,
stderr_tail,
} => {
write!(
f,
"its process reported and then exited with {status} before its door answered"
)?;
tail(f, stderr_tail)
}
Self::DidNotReport { deadline } => {
write!(f, "it did not report within {}s", deadline.as_secs_f32())
}
Self::UnreadableReport { why } => {
write!(f, "its first line is not a readiness line: {why}")
}
Self::ReportedNoDoor => f.write_str("it reported no door"),
Self::WrongFormation { reported, expected } => match reported {
Some(reported) => {
write!(f, "it reported formation {reported} instead of {expected}")
}
None => write!(f, "it reported no formation instead of {expected}"),
},
Self::DoorDidNotAnswer {
url,
session_id,
deadline,
last_error,
} => write!(
f,
"its door at {url} did not answer as session {session_id} within {}s; the last \
probe said: {last_error}",
deadline.as_secs_f32()
),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemberFailure {
pub name: String,
pub capsule: String,
pub version: String,
pub reason: MemberFailureReason,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StoppedMember {
pub name: String,
pub pid: u32,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemberLaunchFailure {
pub failed: Vec<MemberFailure>,
pub stopped: Vec<StoppedMember>,
pub interrupted_by: Option<i32>,
}
impl std::fmt::Display for MemberLaunchFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self.interrupted_by {
Some(signal) if self.failed.is_empty() => write!(
f,
"the formation launch was interrupted by {} before every member was ready",
signal_name(signal)
)?,
_ => f.write_str("the formation could not be launched")?,
}
for failure in &self.failed {
write!(
f,
"\n '{}' ({}@{}): {}",
failure.name, failure.capsule, failure.version, failure.reason
)?;
}
write!(f, "\n stopped: {}", render_stopped(&self.stopped))
}
}
pub fn render_stopped(stopped: &[StoppedMember]) -> String {
if stopped.is_empty() {
return "none".to_string();
}
stopped
.iter()
.map(|member| format!("{} (pid {})", member.name, member.pid))
.collect::<Vec<_>>()
.join(", ")
}
pub fn signal_name(signal: i32) -> String {
match signal {
libc::SIGINT => "SIGINT".to_string(),
libc::SIGTERM => "SIGTERM".to_string(),
libc::SIGHUP => "SIGHUP".to_string(),
other => format!("signal {other}"),
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FormationEnding {
EntryExited(ExitStatus),
Signalled(i32),
}
#[derive(Debug, Clone)]
pub struct FormationExit {
pub ending: FormationEnding,
pub stopped: Vec<StoppedMember>,
}
impl FormationExit {
pub fn exit_code(&self) -> i32 {
match &self.ending {
FormationEnding::EntryExited(status) => crate::shell::exit_code_of(status),
FormationEnding::Signalled(signal) => 128 + signal,
}
}
}
fn describe_status(status: ExitStatus) -> String {
if let Some(code) = status.code() {
return format!("status {code}");
}
#[cfg(unix)]
{
use std::os::unix::process::ExitStatusExt;
if let Some(signal) = status.signal() {
return format!("signal {signal}");
}
}
status.to_string()
}
static CAUGHT_SIGNAL: AtomicI32 = AtomicI32::new(0);
#[cfg(unix)]
extern "C" fn record_signal(signal: libc::c_int) {
CAUGHT_SIGNAL.store(signal, Ordering::SeqCst);
}
#[cfg(unix)]
#[allow(unsafe_code)]
pub fn catch_launcher_signals() -> std::io::Result<()> {
for signal in [libc::SIGINT, libc::SIGTERM, libc::SIGHUP] {
let installed = unsafe {
let mut action: libc::sigaction = std::mem::zeroed();
action.sa_sigaction = record_signal as extern "C" fn(libc::c_int) as libc::sighandler_t;
action.sa_flags = libc::SA_RESTART;
libc::sigemptyset(&mut action.sa_mask);
libc::sigaction(signal, &action, std::ptr::null_mut())
};
if installed != 0 {
return Err(std::io::Error::last_os_error());
}
}
Ok(())
}
#[cfg(not(unix))]
pub fn catch_launcher_signals() -> std::io::Result<()> {
Ok(())
}
pub fn caught_signal() -> Option<i32> {
match CAUGHT_SIGNAL.load(Ordering::SeqCst) {
0 => None,
signal => Some(signal),
}
}
const REGISTRY_SLOTS: usize = 1024;
static MEMBER_REGISTRY: [AtomicI64; REGISTRY_SLOTS] = [const { AtomicI64::new(0) }; REGISTRY_SLOTS];
struct RegistrySlot(Option<usize>);
impl RegistrySlot {
fn register(pid: u32, own_group: bool) -> Self {
let pid = i64::from(pid);
let value = if own_group { pid } else { -pid };
let slot = MEMBER_REGISTRY.iter().position(|slot| {
slot.compare_exchange(0, value, Ordering::SeqCst, Ordering::SeqCst)
.is_ok()
});
Self(slot)
}
fn release(&mut self) {
if let Some(index) = self.0.take() {
MEMBER_REGISTRY[index].store(0, Ordering::SeqCst);
}
}
}
impl Drop for RegistrySlot {
fn drop(&mut self) {
self.release();
}
}
pub(crate) fn kill_registered_members() {
for slot in &MEMBER_REGISTRY {
match slot.load(Ordering::SeqCst) {
0 => {}
pid if pid > 0 => signal_group(pid, libc::SIGKILL),
pid => signal_pid(-pid, libc::SIGKILL),
}
}
}
fn install_panic_hook() {
static INSTALLED: Once = Once::new();
INSTALLED.call_once(|| {
let previous = std::panic::take_hook();
std::panic::set_hook(Box::new(move |info| {
previous(info);
if cfg!(panic = "abort") {
kill_registered_members();
}
}));
});
}
#[cfg(unix)]
#[allow(unsafe_code)]
fn signal_pid(pid: i64, signal: libc::c_int) {
let Ok(pid) = libc::pid_t::try_from(pid) else {
return;
};
if pid <= 0 {
return;
}
unsafe {
libc::kill(pid, signal);
}
}
#[cfg(unix)]
#[allow(unsafe_code)]
fn signal_group(pgid: i64, signal: libc::c_int) {
let Ok(pgid) = libc::pid_t::try_from(pgid) else {
return;
};
if pgid <= 1 {
return;
}
unsafe {
libc::killpg(pgid, signal);
}
}
#[cfg(not(unix))]
fn signal_pid(_pid: i64, _signal: i32) {}
#[cfg(not(unix))]
fn signal_group(_pgid: i64, _signal: i32) {}
struct MemberProcess {
name: String,
pid: u32,
child: Child,
own_group: bool,
#[cfg(unix)]
lifeline: Option<crate::lifeline::MemberLifeline>,
status: Option<ExitStatus>,
slot: RegistrySlot,
grace: Duration,
channel: Option<std::io::PipeWriter>,
}
impl MemberProcess {
fn new(
name: &str,
child: Child,
own_group: bool,
grace: Duration,
channel: std::io::PipeWriter,
#[cfg(unix)] lifeline: Option<crate::lifeline::MemberLifeline>,
) -> Self {
let pid = child.id();
Self {
name: name.to_string(),
pid,
child,
own_group,
#[cfg(unix)]
lifeline,
status: None,
slot: RegistrySlot::register(pid, own_group),
grace,
channel: Some(channel),
}
}
fn send_line(&mut self, line: &str) {
if let Some(channel) = self.channel.as_mut() {
let _ = channel.write_all(format!("{line}\n").as_bytes());
}
}
fn has_exited(&self) -> bool {
self.status.is_some() || !crate::running::pid_is_alive(self.pid)
}
fn exit_status_now(&mut self) -> Option<ExitStatus> {
if self.status.is_none() && self.has_exited() {
self.reap();
}
self.status
}
fn reap(&mut self) {
if self.status.is_some() {
return;
}
if self.own_group {
signal_group(i64::from(self.pid), libc::SIGKILL);
} else if !self.has_exited() {
signal_pid(i64::from(self.pid), libc::SIGKILL);
}
self.status = self.child.wait().ok();
if self.status.is_none() {
self.status = Some(ExitStatus::default());
}
self.slot.release();
self.channel = None;
}
fn close_lifeline(&mut self) {
#[cfg(unix)]
if let Some(lifeline) = &mut self.lifeline {
lifeline.close();
}
}
fn is_running(&self) -> bool {
self.status.is_none() && !self.has_exited()
}
fn stop(&mut self) -> Option<StoppedMember> {
let was_running = self.is_running();
self.stop_from(was_running)
}
fn stop_from(&mut self, was_running: bool) -> Option<StoppedMember> {
if self.status.is_some() {
return None;
}
self.close_lifeline();
if was_running {
let deadline = Instant::now() + self.grace;
while !self.has_exited() && Instant::now() < deadline {
std::thread::sleep(POLL_INTERVAL);
}
}
self.reap();
was_running.then(|| StoppedMember {
name: self.name.clone(),
pid: self.pid,
})
}
}
impl Drop for MemberProcess {
fn drop(&mut self) {
let _ = self.stop();
}
}
struct MemberSet(Vec<MemberProcess>);
impl MemberSet {
fn stop_all(&mut self) -> Vec<StoppedMember> {
let running: Vec<bool> = self.0.iter().map(MemberProcess::is_running).collect();
for member in &mut self.0 {
member.close_lifeline();
}
std::thread::scope(|scope| {
let stopping: Vec<_> = self
.0
.iter_mut()
.zip(running)
.map(|(member, was_running)| scope.spawn(move || member.stop_from(was_running)))
.collect();
stopping
.into_iter()
.filter_map(|handle| handle.join().ok().flatten())
.collect()
})
}
}
impl Drop for MemberSet {
fn drop(&mut self) {
let _ = self.stop_all();
}
}
fn member_args(member: &PlannedMember, workdir: &Path) -> Vec<String> {
vec![
"run".to_string(),
"--capsule".to_string(),
member.capsule.clone(),
"--capsule-version".to_string(),
member.version.clone(),
CAPSULE_SHA256_FLAG.to_string(),
member.sha256.clone(),
"--workdir".to_string(),
workdir.display().to_string(),
]
}
fn pass_through(args: &mut Vec<String>, options: &FormationLaunchOptions) {
if options.no_env_file {
args.push("--no-env-file".to_string());
}
if let Some(class) = &options.containment {
args.push("--containment".to_string());
args.push(class.clone());
}
}
pub(crate) fn peer_args(
member: &PlannedMember,
options: &FormationLaunchOptions,
workdir: &Path,
) -> Vec<String> {
let mut args = member_args(member, workdir);
args.extend([
STORE_ROOT_FLAG.to_string(),
options.project_dir.display().to_string(),
"--json".to_string(),
child_launch::READINESS_TOKENS_FLAG.to_string(),
"--bind".to_string(),
PEER_BIND_ADDR.to_string(),
]);
pass_through(&mut args, options);
args
}
pub(crate) fn entry_args(member: &PlannedMember, options: &FormationLaunchOptions) -> Vec<String> {
let mut args = member_args(member, &options.project_dir);
args.extend([
"--lifecycle-task-acceptance".to_string(),
"single".to_string(),
"--lifecycle-after-task".to_string(),
"exit".to_string(),
]);
if let Some(task) = &options.task {
args.push("--task".to_string());
args.push(task.clone());
}
if options.json {
args.push("--json".to_string());
}
if options.verbose {
args.push("--verbose".to_string());
}
pass_through(&mut args, options);
args
}
fn formations_dir() -> Result<PathBuf, String> {
Ok(crate::state_store::murmur_home_dir()?.join(FORMATIONS_DIR))
}
pub fn formation_member_roots(formation_id: &FormationId) -> Vec<PathBuf> {
match formations_dir() {
Ok(dir) => formation_member_roots_in(&dir, formation_id),
Err(_) => Vec::new(),
}
}
pub fn formation_member_roots_in(
formations_dir: &Path,
formation_id: &FormationId,
) -> Vec<PathBuf> {
let dir = formations_dir.join(formation_id.as_str());
let entries = match std::fs::read_dir(&dir) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Vec::new(),
Err(_) => return vec![dir],
};
let mut roots: Vec<PathBuf> = entries
.filter_map(Result::ok)
.filter(|entry| entry.file_type().is_ok_and(|kind| kind.is_dir()))
.map(|entry| entry.path().join(".murmur"))
.filter(|root| root.exists())
.collect();
roots.sort();
roots
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FormationMarker {
pub project_dir: PathBuf,
pub launcher_pid: u32,
pub launcher_start: String,
}
impl FormationMarker {
pub fn for_this_launcher(project_dir: &Path) -> Self {
let pid = std::process::id();
Self {
project_dir: canonical_project_dir(project_dir),
launcher_pid: pid,
launcher_start: crate::running::process_start_token(pid).unwrap_or_default(),
}
}
pub fn read(formation_dir: &Path) -> Option<Self> {
let raw = std::fs::read_to_string(formation_dir.join(FORMATION_MARKER_FILE)).ok()?;
serde_json::from_str(&raw).ok()
}
}
pub(crate) fn canonical_project_dir(project_dir: &Path) -> PathBuf {
std::fs::canonicalize(project_dir).unwrap_or_else(|_| project_dir.to_path_buf())
}
fn write_formation_marker(options: &FormationLaunchOptions) -> Result<(), String> {
let formations_dir = launch_formations_dir(options)?;
let formation_dir = formations_dir.join(options.formation_id.as_str());
for dir in [formations_dir.as_path(), formation_dir.as_path()] {
crate::state_store::ensure_private_dir(dir, crate::murmur_home::MURMUR_HOME_DIR_MODE)
.map_err(|reason| format!("{}: {reason}", dir.display()))?;
}
let marker = FormationMarker::for_this_launcher(&options.project_dir);
let json = serde_json::to_string(&marker).map_err(|error| error.to_string())?;
crate::murmur_home::write_private_file(
&formation_dir.join(FORMATION_MARKER_FILE),
format!("{json}\n").as_bytes(),
)
}
pub fn prune_ended_formations(
current: &FormationId,
project_dir: &Path,
policy: &murmur_artifact::TraceRetainConfig,
) -> Vec<crate::retention::PrunedFormation> {
let (Ok(formations_dir), Ok(running_dir)) =
(formations_dir(), crate::running::running_dir_location())
else {
return Vec::new();
};
prune_ended_formations_in(&formations_dir, &running_dir, current, project_dir, policy)
}
pub fn prune_ended_formations_in(
formations_dir: &Path,
running_dir: &Path,
current: &FormationId,
project_dir: &Path,
policy: &murmur_artifact::TraceRetainConfig,
) -> Vec<crate::retention::PrunedFormation> {
let records = crate::running::scan(running_dir)
.ok()
.map(|scan| scan.records);
crate::retention::prune_formations(
formations_dir,
current,
project_dir,
policy,
crate::retention::now_ms(),
|id, marker| formation_is_live(id, marker, records.as_deref()),
)
}
fn formation_is_live(
id: &FormationId,
marker: &FormationMarker,
records: Option<&[crate::running::RunningRecord]>,
) -> bool {
let not_gone = |state| !matches!(state, crate::running::ProcessState::Gone(_));
let Some(records) = records else {
return true;
};
not_gone(crate::running::process_state_of(
marker.launcher_pid,
&marker.launcher_start,
)) || records.iter().any(|record| {
record.formation_id.as_ref() == Some(id) && not_gone(crate::running::process_state(record))
})
}
fn make_member_workdir(
formations_dir: &Path,
formation_id: &FormationId,
member: &str,
) -> Result<PathBuf, String> {
use std::os::unix::fs::DirBuilderExt;
let formation_dir = formations_dir.join(formation_id.as_str());
let workdir = formation_dir.join(member);
let failed = |path: &Path, reason: String| {
format!(
"its directory {} could not be made: {}: {reason}",
workdir.display(),
path.display()
)
};
for dir in [formations_dir, formation_dir.as_path()] {
crate::state_store::ensure_private_dir(dir, crate::murmur_home::MURMUR_HOME_DIR_MODE)
.map_err(|reason| failed(dir, reason))?;
}
std::fs::DirBuilder::new()
.mode(crate::murmur_home::MURMUR_HOME_DIR_MODE)
.create(&workdir)
.map_err(|error| failed(&workdir, error.to_string()))?;
Ok(workdir)
}
fn launch_formations_dir(options: &FormationLaunchOptions) -> Result<PathBuf, String> {
match &options.formations_dir {
Some(dir) => Ok(dir.clone()),
None => {
crate::murmur_home::ensure_murmur_home()?;
formations_dir()
}
}
}
enum PeerOutcome {
Ready(ReadyPeer),
Failed(MemberFailureReason),
Abandoned,
}
struct Channel {
writer: std::io::PipeWriter,
reader: std::io::PipeReader,
}
impl Channel {
fn open(first_line: &str) -> Result<Self, MemberFailureReason> {
let (reader, mut writer) = std::io::pipe().map_err(|error| {
not_started(&format!(
"its formation channel could not be created: {error}"
))
})?;
writer
.write_all(format!("{first_line}\n").as_bytes())
.map_err(|error| {
not_started(&format!(
"its formation channel could not be written: {error}"
))
})?;
Ok(Self { writer, reader })
}
fn fd(&self) -> i32 {
use std::os::fd::AsRawFd;
self.reader.as_raw_fd()
}
fn member_env(&self, formation_id: &FormationId) -> Vec<(String, String)> {
let (id_name, id_value) = formation_id.env_pair();
vec![
(id_name.to_string(), id_value),
(FORMATION_CHANNEL_ENV.to_string(), self.fd().to_string()),
crate::host_warnings::reported_env_pair(),
]
}
}
fn start_peer(
binary: &Path,
member: &PlannedMember,
first_line: &str,
options: &FormationLaunchOptions,
abort: &AtomicBool,
#[cfg(unix)] mut lifeline: crate::lifeline::MemberLifeline,
) -> (Option<MemberProcess>, PeerOutcome) {
let workdir = match launch_formations_dir(options)
.and_then(|dir| make_member_workdir(&dir, &options.formation_id, &member.name))
{
Ok(workdir) => workdir,
Err(error) => return (None, PeerOutcome::Failed(not_started(&error))),
};
let args = peer_args(member, options, &workdir);
let prefix = format!("[{}] ", member.name);
let channel = match Channel::open(first_line) {
Ok(channel) => channel,
Err(reason) => return (None, PeerOutcome::Failed(reason)),
};
let set = channel.member_env(&options.formation_id);
let spawned_at = Instant::now();
let deadline = spawned_at + options.ready_timeout;
let started = child_launch::start_process(&ProcessLaunch {
binary,
args: &args,
cwd: None,
env: ProcessEnv::Inherited {
set: &set,
remove: &[FORMATION_PEERS_ENV, SPAWNER_ENV, SPAWNER_LIFELINE_ENV],
},
stdin_line: None,
stderr_prefix: Some(&prefix),
inherit_output: false,
own_process_group: true,
inherit_fd: Some(channel.fd()),
#[cfg(unix)]
lifeline: Some(crate::child_launch::Lifeline::Member(&lifeline)),
});
let Channel { writer, reader } = channel;
drop(reader);
let StartedProcess {
mut child,
stderr_tail,
..
} = match started {
Ok(started) => started,
Err(error) => {
return (
None,
PeerOutcome::Failed(MemberFailureReason::NotStarted {
error: format!("{}: {error}", binary.display()),
}),
)
}
};
#[cfg(unix)]
lifeline.spawned();
let stdout = child.stdout.take();
let mut process = MemberProcess::new(
&member.name,
child,
true,
options.stop_grace,
writer,
#[cfg(unix)]
Some(lifeline),
);
let Some(stdout) = stdout else {
return (
Some(process),
PeerOutcome::Failed(MemberFailureReason::NotStarted {
error: "the process exposed no standard output".to_string(),
}),
);
};
let lines = child_launch::relay_readiness_line(stdout, AfterReadiness::PrefixedStderr(prefix));
let line = loop {
if abort.load(Ordering::SeqCst) {
return (Some(process), PeerOutcome::Abandoned);
}
let now = Instant::now();
if now >= deadline {
return (
Some(process),
PeerOutcome::Failed(MemberFailureReason::DidNotReport {
deadline: options.ready_timeout,
}),
);
}
match lines.recv_timeout(POLL_INTERVAL.min(deadline - now)) {
Ok(Some(line)) => break line,
Ok(None) | Err(mpsc::RecvTimeoutError::Disconnected) => {
let closed_at = Instant::now();
let status = loop {
if let Some(status) = process.exit_status_now() {
break status;
}
if closed_at.elapsed() >= EXIT_AFTER_EOF_WAIT {
process.reap();
break process.status.unwrap_or_default();
}
std::thread::sleep(POLL_INTERVAL);
};
return (
Some(process),
PeerOutcome::Failed(MemberFailureReason::ExitedBeforeReporting {
status: describe_status(status),
stderr_tail: stderr_tail.lines_at_end(),
}),
);
}
Err(mpsc::RecvTimeoutError::Timeout) => {}
}
};
let report = match read_report(&line, &options.formation_id) {
Ok(report) => report,
Err(reason) => return (Some(process), PeerOutcome::Failed(reason)),
};
let mut last_error;
loop {
if abort.load(Ordering::SeqCst) {
return (Some(process), PeerOutcome::Abandoned);
}
if let Some(status) = process.exit_status_now() {
return (
Some(process),
PeerOutcome::Failed(MemberFailureReason::ExitedBeforeDoorAnswered {
status: describe_status(status),
stderr_tail: stderr_tail.lines_at_end(),
}),
);
}
match door_answers_as(
crate::running::probe_door_session_id(&report.url, report.door_token.as_ref()),
&report.session_id,
) {
Ok(()) => {
let pid = process.pid;
return (
Some(process),
PeerOutcome::Ready(ReadyPeer {
name: member.name.clone(),
capsule: member.capsule.clone(),
version: member.version.clone(),
pid,
session_id: report.session_id,
url: report.url,
workdir,
door_token: report.door_token,
}),
);
}
Err(why) => last_error = why,
}
let now = Instant::now();
if now >= deadline {
return (
Some(process),
PeerOutcome::Failed(MemberFailureReason::DoorDidNotAnswer {
url: report.url,
session_id: report.session_id,
deadline: options.ready_timeout,
last_error,
}),
);
}
std::thread::sleep(READY_PROBE_INTERVAL.min(deadline - now));
}
}
fn door_answers_as(probed: Result<String, String>, reported: &str) -> Result<(), String> {
match probed {
Ok(session_id) if session_id == reported => Ok(()),
Ok(other) => Err(format!("the door answers as session {other}")),
Err(error) => Err(error),
}
}
struct Report {
session_id: String,
url: String,
door_token: Option<DoorToken>,
}
fn read_report(line: &str, expected: &FormationId) -> Result<Report, MemberFailureReason> {
let report: serde_json::Value =
serde_json::from_str(line).map_err(|error| MemberFailureReason::UnreadableReport {
why: format!("it does not parse as JSON ({error})"),
})?;
let reported = report
.get("formation_id")
.and_then(serde_json::Value::as_str);
if reported != Some(expected.as_str()) {
return Err(MemberFailureReason::WrongFormation {
reported: reported.map(str::to_string),
expected: expected.as_str().to_string(),
});
}
let session_id = report
.get("session_id")
.and_then(serde_json::Value::as_str)
.filter(|session_id| !session_id.is_empty())
.ok_or_else(|| MemberFailureReason::UnreadableReport {
why: "it names no session_id".to_string(),
})?;
let url = report
.get("url")
.and_then(serde_json::Value::as_str)
.unwrap_or_default();
if url.is_empty() {
return Err(MemberFailureReason::ReportedNoDoor);
}
Ok(Report {
session_id: session_id.to_string(),
url: format!("http://{}", url.trim_start_matches("http://")),
door_token: child_launch::readiness_door_token(&report),
})
}
pub fn launch_formation(
roster: &AdmittedRoster,
options: FormationLaunchOptions,
) -> Result<RunningFormation, MemberLaunchFailure> {
launch_plan(FormationPlan::from_roster(roster), options)
}
pub(crate) fn launch_plan(
plan: FormationPlan,
options: FormationLaunchOptions,
) -> Result<RunningFormation, MemberLaunchFailure> {
install_panic_hook();
let binary = match child_launch::mur_binary() {
Ok(binary) => binary,
Err(error) => {
return Err(MemberLaunchFailure {
failed: plan
.peers
.iter()
.chain([&plan.entry])
.map(|member| failure(member, not_started(&error.to_string())))
.collect(),
stopped: Vec::new(),
interrupted_by: None,
})
}
};
let authority = match FormationAuthority::generate(&options.formation_id) {
Ok(authority) => authority,
Err(error) => {
return Err(MemberLaunchFailure {
failed: plan
.peers
.iter()
.chain([&plan.entry])
.map(|member| failure(member, not_started(&error)))
.collect(),
stopped: Vec::new(),
interrupted_by: None,
})
}
};
#[cfg(unix)]
let (peer_lifelines, entry_lifeline) = match member_lifelines(plan.peers.len() + 1) {
Ok(mut lifelines) => {
let entry = lifelines.pop();
(lifelines, entry)
}
Err(error) => {
return Err(MemberLaunchFailure {
failed: plan
.peers
.iter()
.chain([&plan.entry])
.map(|member| {
failure(
member,
not_started(&format!("its lifeline could not be created: {error}")),
)
})
.collect(),
stopped: Vec::new(),
interrupted_by: None,
})
}
};
let launcher = ProcessIdentity::of_this_process();
let first_lines: Vec<String> = plan
.peers
.iter()
.map(|member| plan.first_line(&authority, launcher.as_ref(), &member.name))
.collect();
if !plan.peers.is_empty() {
if let Err(reason) = write_formation_marker(&options) {
crate::runtime_err!(
"[mur run] warning: formation {} has no ownership marker ({reason}); its \
directory will never be removed automatically",
options.formation_id
);
}
}
let abort = AtomicBool::new(false);
let mut outcomes: Vec<Option<(Option<MemberProcess>, PeerOutcome)>> =
plan.peers.iter().map(|_| None).collect();
let mut interrupted_by = caught_signal();
if interrupted_by.is_some() {
abort.store(true, Ordering::SeqCst);
}
std::thread::scope(|scope| {
let (tx, rx) = mpsc::channel();
#[cfg(unix)]
let mut peer_lifelines = peer_lifelines.into_iter();
for (index, member) in plan.peers.iter().enumerate() {
let tx = tx.clone();
let (binary, options, abort) = (&binary, &options, &abort);
let first_line = first_lines[index].as_str();
#[cfg(unix)]
let lifeline = peer_lifelines
.next()
.expect("member_lifelines made one lifeline per peer");
scope.spawn(move || {
let started = start_peer(
binary,
member,
first_line,
options,
abort,
#[cfg(unix)]
lifeline,
);
let _ = tx.send((index, started));
});
}
drop(tx);
loop {
match rx.recv_timeout(POLL_INTERVAL) {
Ok((index, outcome)) => {
if matches!(outcome.1, PeerOutcome::Failed(_)) {
abort.store(true, Ordering::SeqCst);
}
outcomes[index] = Some(outcome);
}
Err(mpsc::RecvTimeoutError::Timeout) => {}
Err(mpsc::RecvTimeoutError::Disconnected) => break,
}
if interrupted_by.is_none() {
if let Some(signal) = caught_signal() {
interrupted_by = Some(signal);
abort.store(true, Ordering::SeqCst);
}
}
}
});
let mut members = MemberSet(Vec::new());
let mut peers = Vec::new();
let mut failed = Vec::new();
for (member, outcome) in plan.peers.iter().zip(outcomes) {
let Some((process, outcome)) = outcome else {
continue;
};
members.0.extend(process);
match outcome {
PeerOutcome::Ready(peer) => peers.push(peer),
PeerOutcome::Failed(reason) => failed.push(failure(member, reason)),
PeerOutcome::Abandoned => {}
}
}
if !failed.is_empty() || interrupted_by.is_some() || peers.len() != plan.peers.len() {
let stopped = members.stop_all();
return Err(MemberLaunchFailure {
failed,
stopped,
interrupted_by,
});
}
for member in &mut members.0 {
let addresses = addresses_of(&peers, plan.callees_of(&member.name));
if !addresses.is_empty() {
member.send_line(&render_address_line(&addresses));
}
}
Ok(RunningFormation {
formation_id: options.formation_id.clone(),
entry: plan.entry.clone(),
plan,
peers,
members,
entry_index: None,
#[cfg(unix)]
entry_lifeline,
binary,
options,
authority,
launcher,
})
}
fn addresses_of(ready: &[ReadyPeer], callees: &[String]) -> Vec<FormationPeer> {
ready
.iter()
.filter(|peer| callees.contains(&peer.name))
.map(|peer| FormationPeer {
name: peer.name.clone(),
url: peer.url.clone(),
})
.collect()
}
#[cfg(unix)]
fn member_lifelines(count: usize) -> std::io::Result<Vec<crate::lifeline::MemberLifeline>> {
(0..count)
.map(|_| crate::lifeline::MemberLifeline::new())
.collect()
}
fn failure(member: &PlannedMember, reason: MemberFailureReason) -> MemberFailure {
MemberFailure {
name: member.name.clone(),
capsule: member.capsule.clone(),
version: member.version.clone(),
reason,
}
}
fn not_started(error: &str) -> MemberFailureReason {
MemberFailureReason::NotStarted {
error: error.to_string(),
}
}
pub struct RunningFormation {
formation_id: FormationId,
entry: PlannedMember,
plan: FormationPlan,
peers: Vec<ReadyPeer>,
members: MemberSet,
entry_index: Option<usize>,
#[cfg(unix)]
entry_lifeline: Option<crate::lifeline::MemberLifeline>,
binary: PathBuf,
options: FormationLaunchOptions,
authority: FormationAuthority,
launcher: Option<ProcessIdentity>,
}
impl std::fmt::Debug for RunningFormation {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RunningFormation")
.field("formation_id", &self.formation_id)
.field("entry", &self.entry.name)
.field("peers", &self.peers)
.finish()
}
}
impl RunningFormation {
pub fn formation_id(&self) -> &FormationId {
&self.formation_id
}
pub fn entry_name(&self) -> &str {
&self.entry.name
}
pub fn peers(&self) -> &[ReadyPeer] {
&self.peers
}
pub fn callees_of(&self, member: &str) -> &[String] {
self.plan.callees_of(member)
}
pub fn entry_pid(&self) -> Option<u32> {
self.entry_index.map(|index| self.members.0[index].pid)
}
pub fn start_entry(&mut self) -> Result<u32, MemberLaunchFailure> {
let args = entry_args(&self.entry, &self.options);
let channel = Channel::open(&self.plan.first_line(
&self.authority,
self.launcher.as_ref(),
&self.entry.name,
))
.and_then(|mut channel| {
let addresses = addresses_of(&self.peers, self.plan.callees_of(&self.entry.name));
if !addresses.is_empty() {
channel
.writer
.write_all(format!("{}\n", render_address_line(&addresses)).as_bytes())
.map_err(|error| {
not_started(&format!(
"its formation channel could not be written: {error}"
))
})?;
}
Ok(channel)
});
let channel = match channel {
Ok(channel) => channel,
Err(reason) => {
let stopped = self.members.stop_all();
return Err(MemberLaunchFailure {
failed: vec![failure(&self.entry, reason)],
stopped,
interrupted_by: None,
});
}
};
let set = channel.member_env(&self.formation_id);
let started = child_launch::start_process(&ProcessLaunch {
binary: &self.binary,
args: &args,
cwd: None,
env: ProcessEnv::Inherited {
set: &set,
remove: &[SPAWNER_ENV, FORMATION_PEERS_ENV, SPAWNER_LIFELINE_ENV],
},
stdin_line: None,
stderr_prefix: None,
inherit_output: true,
own_process_group: false,
inherit_fd: Some(channel.fd()),
#[cfg(unix)]
lifeline: self
.entry_lifeline
.as_ref()
.map(crate::child_launch::Lifeline::Member),
});
let Channel { writer, reader } = channel;
drop(reader);
match started {
Ok(started) => {
#[cfg(unix)]
let lifeline = self.entry_lifeline.take().map(|mut lifeline| {
lifeline.spawned();
lifeline
});
let process = MemberProcess::new(
&self.entry.name,
started.child,
false,
self.options.stop_grace,
writer,
#[cfg(unix)]
lifeline,
);
let pid = process.pid;
self.members.0.push(process);
self.entry_index = Some(self.members.0.len() - 1);
Ok(pid)
}
Err(error) => {
let stopped = self.members.stop_all();
Err(MemberLaunchFailure {
failed: vec![failure(
&self.entry,
not_started(&format!("{}: {error}", self.binary.display())),
)],
stopped,
interrupted_by: None,
})
}
}
}
pub fn wait(mut self) -> FormationExit {
let ending = match self.entry_index {
None => caught_signal()
.map(FormationEnding::Signalled)
.unwrap_or(FormationEnding::EntryExited(ExitStatus::default())),
Some(index) => loop {
if let Some(status) = self.members.0[index].exit_status_now() {
break FormationEnding::EntryExited(status);
}
if let Some(signal) = caught_signal() {
break FormationEnding::Signalled(signal);
}
std::thread::sleep(POLL_INTERVAL);
},
};
let stopped = self.members.stop_all();
FormationExit { ending, stopped }
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::{Read, Write};
use std::net::TcpListener;
use std::sync::atomic::AtomicUsize;
use std::sync::Arc;
const LIVENESS_BOUND: Duration = Duration::from_secs(120);
fn load_average() -> String {
std::fs::read_to_string("/proc/loadavg")
.unwrap_or_default()
.split(' ')
.take(3)
.collect::<Vec<_>>()
.join(" ")
}
#[test]
fn the_production_bounds_are_180s_and_a_grace_past_the_runtime_teardown() {
assert_eq!(MEMBER_READY_TIMEOUT, Duration::from_secs(180));
assert_eq!(MEMBER_STOP_GRACE, Duration::from_secs(25));
assert!(MEMBER_STOP_GRACE > crate::runtime::TERMINATE_TEARDOWN_DEADLINE);
}
fn member(name: &str) -> PlannedMember {
PlannedMember {
name: name.to_string(),
capsule: name.to_string(),
version: "0.1.0".to_string(),
sha256: "ab".repeat(32),
}
}
#[test]
fn every_member_is_told_host_warnings_were_reported() {
let channel = Channel::open("first").expect("a channel opens");
let env = channel.member_env(&FormationId::mint());
let values: Vec<&str> = env
.iter()
.filter(|(key, _)| key == crate::host_warnings::HOST_WARNINGS_REPORTED_ENV)
.map(|(_, value)| value.as_str())
.collect();
assert_eq!(values, vec!["1"], "{env:?}");
}
fn options(deadline: Duration) -> FormationLaunchOptions {
let mut options =
FormationLaunchOptions::new(FormationId::mint(), PathBuf::from("/tmp/project"));
options.ready_timeout = deadline;
options.stop_grace = Duration::from_secs(2);
options.formations_dir = Some(tempfile::tempdir().unwrap().keep());
options
}
#[test]
fn member_argv_is_exactly_the_documented_shape() {
let mut options = options(MEMBER_READY_TIMEOUT);
let coder = member("coder");
let workdir = PathBuf::from("/home/u/.murmur/formations/frm_x/coder");
assert_eq!(
peer_args(&coder, &options, &workdir),
[
"run",
"--capsule",
"coder",
"--capsule-version",
"0.1.0",
"--capsule-sha256",
&"ab".repeat(32),
"--workdir",
"/home/u/.murmur/formations/frm_x/coder",
"--store-root",
"/tmp/project",
"--json",
"--readiness-tokens",
"--bind",
"127.0.0.1",
]
);
options.task = Some("probe".to_string());
options.json = true;
options.verbose = true;
options.no_env_file = true;
options.containment = Some("scoped".to_string());
let entry = entry_args(&member("planner"), &options);
assert_eq!(&entry[7..9], ["--workdir", "/tmp/project"]);
assert_eq!(
&entry[9..],
[
"--lifecycle-task-acceptance",
"single",
"--lifecycle-after-task",
"exit",
"--task",
"probe",
"--json",
"--verbose",
"--no-env-file",
"--containment",
"scoped",
]
);
assert!(
!entry.iter().any(|arg| arg == "--bind"
|| arg == STORE_ROOT_FLAG
|| arg == child_launch::READINESS_TOKENS_FLAG),
"{entry:?}"
);
let peer = peer_args(&coder, &options, &workdir);
assert!(peer.ends_with(&[
"--no-env-file".to_string(),
"--containment".to_string(),
"scoped".to_string()
]));
assert!(!peer.iter().any(|arg| arg == "--task" || arg == "--verbose"));
}
fn binary_guard() -> std::sync::MutexGuard<'static, ()> {
crate::formation::FORMATION_ENV_LOCK
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
fn stand_in(dir: &Path, line: Option<&str>) -> PathBuf {
let script = dir.join("mur");
let print = match line {
Some(line) => {
std::fs::write(dir.join("line"), format!("{line}\n")).unwrap();
format!("cat '{}'\n", dir.join("line").display())
}
None => String::new(),
};
let body = format!(
"#!/bin/sh\ntrap 'echo TERM >> \"{dir}/sigterm\"' TERM\n{print}\
cat \"/dev/fd/$MURMUR_FORMATION_LIFELINE\" >/dev/null\n",
dir = dir.display()
);
std::fs::write(&script, body).unwrap();
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
script
}
fn plan_of(peer: &str) -> FormationPlan {
FormationPlan {
peers: vec![member(peer)],
entry: member("planner"),
callees: vec![("planner".to_string(), vec![peer.to_string()])],
}
}
fn refused(line: Option<&str>, deadline: Duration) -> (MemberLaunchFailure, FormationId) {
let _guard = binary_guard();
let dir = tempfile::tempdir().unwrap();
let options = options(deadline);
let id = options.formation_id.clone();
let line = line.map(|line| line.replace("{formation}", id.as_str()));
std::env::set_var(
child_launch::MUR_BINARY_ENV,
stand_in(dir.path(), line.as_deref()),
);
let result = launch_plan(plan_of("coder"), options);
std::env::remove_var(child_launch::MUR_BINARY_ENV);
match result {
Ok(_) => panic!("the launch was expected to be refused"),
Err(failure) => (failure, id),
}
}
fn assert_reaped(failure: &MemberLaunchFailure) {
assert_eq!(failure.stopped.len(), 1, "{failure:?}");
let pid = failure.stopped[0].pid;
assert_eq!(failure.stopped[0].name, "coder");
assert!(
!crate::running::pid_is_alive(pid),
"the stand-in {pid} is still running"
);
assert!(
!Path::new(&format!("/proc/{pid}")).exists() || cfg!(not(target_os = "linux")),
"the stand-in {pid} was not reaped"
);
}
fn only_reason(failure: &MemberLaunchFailure) -> &MemberFailureReason {
assert_eq!(failure.failed.len(), 1, "{failure:?}");
assert_eq!(failure.failed[0].name, "coder");
&failure.failed[0].reason
}
fn silent_port() -> u16 {
crate::pinned_port::claim()
}
#[test]
fn a_door_nothing_answers_on_is_refused_at_the_deadline() {
let deadline = Duration::from_secs(10);
let port = silent_port();
let started = Instant::now();
let (failure, _) = refused(
Some(&format!(
r#"{{"formation_id":"{{formation}}","session_id":"ses_standin","url":"localhost:{port}"}}"#
)),
deadline,
);
assert!(started.elapsed() >= deadline);
match only_reason(&failure) {
MemberFailureReason::DoorDidNotAnswer {
url, session_id, ..
} => {
assert_eq!(url, &format!("http://localhost:{port}"));
assert_eq!(session_id, "ses_standin");
}
MemberFailureReason::DidNotReport { .. } => panic!(
"the stand-in's readiness line did not arrive within {deadline:?} (load average {}); \
host load is the likely cause, since the stand-in prints it at once: {failure:?}",
load_average()
),
other => panic!("{other:?}"),
}
let message = failure.to_string();
assert!(
message.contains("did not answer as session ses_standin"),
"{message}"
);
assert!(message.contains("'coder' (coder@0.1.0)"), "{message}");
assert_reaped(&failure);
}
fn card_door(mut answer: impl FnMut() -> Option<&'static str> + Send + 'static) -> u16 {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
std::thread::spawn(move || {
for stream in listener.incoming().flatten() {
let mut stream = stream;
let mut buffer = [0u8; 4096];
let _ = stream.read(&mut buffer);
let body = match answer() {
Some(session_id) => serde_json::json!({
"name": "coder",
"capabilities": {"extensions": [{
"uri": crate::identity::CAPSULE_EXTENSION_URI,
"params": {"sessionId": session_id},
}]},
})
.to_string(),
None => "{}".to_string(),
};
let _ = write!(
stream,
"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\n\
connection: close\r\n\r\n{body}",
body.len()
);
}
});
port
}
fn door(session_id: &'static str, answer_after: Duration) -> u16 {
let opened = Instant::now();
card_door(move || (opened.elapsed() >= answer_after).then_some(session_id))
}
const WRONG_ANSWERS: usize = 3;
fn door_naming_another_session_first() -> (u16, Arc<AtomicUsize>) {
let served = Arc::new(AtomicUsize::new(0));
let counted = Arc::clone(&served);
let port = card_door(move || {
if counted.fetch_add(1, Ordering::SeqCst) < WRONG_ANSWERS {
Some("ses_somebody_else")
} else {
Some("ses_standin")
}
});
(port, served)
}
#[test]
fn a_door_naming_another_session_is_not_ready() {
let _guard = binary_guard();
let (port, served) = door_naming_another_session_first();
let dir = tempfile::tempdir().unwrap();
let options = options(LIVENESS_BOUND);
let line = format!(
r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
options.formation_id
);
std::env::set_var(
child_launch::MUR_BINARY_ENV,
stand_in(dir.path(), Some(&line)),
);
let result = launch_plan(plan_of("coder"), options);
std::env::remove_var(child_launch::MUR_BINARY_ENV);
let formation = match result {
Ok(formation) => formation,
Err(failure) => match only_reason(&failure) {
MemberFailureReason::DidNotReport { .. } => panic!(
"the stand-in's readiness line did not arrive within {LIVENESS_BOUND:?} \
(load average {}); host load is the likely cause: {failure:?}",
load_average()
),
_ => panic!(
"the launch was refused although the door answered as ses_standin from its \
request {} on: {failure:?}",
WRONG_ANSWERS + 1
),
},
};
let peer = &formation.peers()[0];
assert_eq!(peer.session_id, "ses_standin");
let served = served.load(Ordering::SeqCst);
assert!(
served > WRONG_ANSWERS,
"an answer naming another session was counted as ready: the door served {served} \
request(s), and its first {WRONG_ANSWERS} named ses_somebody_else"
);
let pid = peer.pid;
drop(formation);
assert!(
!crate::running::pid_is_alive(pid),
"dropping the owner stops the peer"
);
}
#[test]
fn a_door_counts_as_ready_only_when_it_answers_as_the_reported_session() {
let port = door("ses_somebody_else", Duration::ZERO);
let probed = crate::running::probe_door_session_id(&format!("localhost:{port}"), None);
match door_answers_as(probed, "ses_standin") {
Err(why) => assert!(
why.contains("the door answers as session ses_somebody_else"),
"{why}"
),
Ok(()) => panic!("a door answering as ses_somebody_else counted as ses_standin"),
}
assert_eq!(
door_answers_as(Ok("ses_standin".to_string()), "ses_standin"),
Ok(())
);
assert_eq!(
door_answers_as(Err("connection refused".to_string()), "ses_standin"),
Err("connection refused".to_string())
);
}
#[test]
fn an_empty_url_is_refused_as_no_door() {
let (failure, _) = refused(
Some(r#"{"formation_id":"{formation}","session_id":"ses_standin","url":""}"#),
Duration::from_secs(10),
);
assert_eq!(only_reason(&failure), &MemberFailureReason::ReportedNoDoor);
assert!(failure.to_string().contains("it reported no door"));
assert_reaped(&failure);
}
#[test]
fn a_wrong_formation_or_none_is_refused_naming_what_was_reported() {
let other = FormationId::mint();
let (failure, id) = refused(
Some(&format!(
r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:1"}}"#,
other.as_str()
)),
Duration::from_secs(10),
);
assert_eq!(
only_reason(&failure),
&MemberFailureReason::WrongFormation {
reported: Some(other.as_str().to_string()),
expected: id.as_str().to_string(),
}
);
assert!(failure
.to_string()
.contains(&format!("it reported formation {other} instead of {id}")));
assert_reaped(&failure);
let (failure, id) = refused(
Some(r#"{"session_id":"ses_standin","url":"localhost:1"}"#),
Duration::from_secs(10),
);
assert!(failure
.to_string()
.contains(&format!("it reported no formation instead of {id}")));
assert_reaped(&failure);
}
#[test]
fn no_line_at_all_is_refused_at_the_deadline() {
let started = Instant::now();
let (failure, _) = refused(None, Duration::from_millis(600));
assert!(started.elapsed() >= Duration::from_millis(600));
assert_eq!(
only_reason(&failure),
&MemberFailureReason::DidNotReport {
deadline: Duration::from_millis(600)
}
);
assert_reaped(&failure);
}
#[test]
fn a_door_that_starts_answering_before_the_deadline_is_ready() {
let _guard = binary_guard();
let port = door("ses_standin", Duration::from_millis(400));
let dir = tempfile::tempdir().unwrap();
let options = options(Duration::from_secs(20));
let line = format!(
r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
options.formation_id
);
std::env::set_var(
child_launch::MUR_BINARY_ENV,
stand_in(dir.path(), Some(&line)),
);
let result = launch_plan(plan_of("coder"), options);
std::env::remove_var(child_launch::MUR_BINARY_ENV);
let formation = result.expect("the door answered before the deadline");
assert_eq!(formation.peers().len(), 1);
let peer = &formation.peers()[0];
assert_eq!(peer.session_id, "ses_standin");
assert_eq!(peer.url, format!("http://localhost:{port}"));
assert_eq!(formation.callees_of("planner"), ["coder"]);
assert!(formation.callees_of("coder").is_empty());
let pid = peer.pid;
assert!(crate::running::pid_is_alive(pid));
let dropped = Instant::now();
drop(formation);
assert!(
!crate::running::pid_is_alive(pid),
"dropping the owner stops the peer"
);
assert!(
dropped.elapsed() < Duration::from_secs(2),
"the peer ended on its lifeline's EOF, inside the grace: {:?}",
dropped.elapsed()
);
assert!(
!dir.path().join("sigterm").exists(),
"the launcher sent the peer SIGTERM"
);
}
#[test]
fn a_member_deaf_to_its_lifeline_is_killed_after_the_grace() {
let _guard = binary_guard();
let port = door("ses_standin", Duration::ZERO);
let dir = tempfile::tempdir().unwrap();
let mut options = options(Duration::from_secs(20));
options.stop_grace = Duration::from_millis(600);
let line = format!(
r#"{{"formation_id":"{}","session_id":"ses_standin","url":"localhost:{port}"}}"#,
options.formation_id
);
let script = dir.path().join("mur");
std::fs::write(dir.path().join("line"), format!("{line}\n")).unwrap();
std::fs::write(
&script,
format!(
"#!/bin/sh\ncat '{}'\nexec sleep 600\n",
dir.path().join("line").display()
),
)
.unwrap();
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
std::env::set_var(child_launch::MUR_BINARY_ENV, &script);
let result = launch_plan(plan_of("coder"), options);
std::env::remove_var(child_launch::MUR_BINARY_ENV);
let formation = result.expect("the door answers");
let pid = formation.peers()[0].pid;
let dropped = Instant::now();
drop(formation);
assert!(dropped.elapsed() >= Duration::from_millis(600));
assert!(!crate::running::pid_is_alive(pid));
}
#[test]
fn each_member_reads_only_its_own_credentials_and_callees_on_its_channel() {
use crate::formation_credentials::{parse_address_line, FormationBundle};
let _guard = binary_guard();
let dir = tempfile::tempdir().unwrap();
let coder_port = door("ses_coder", Duration::ZERO);
let reviewer_port = door("ses_reviewer", Duration::ZERO);
let options = options(Duration::from_secs(20));
let id = options.formation_id.clone();
let line = |session: &str, port: u16| {
format!(
r#"{{"formation_id":"{id}","session_id":"{session}","url":"localhost:{port}"}}"#
)
};
let out = dir.path().display().to_string();
let script = dir.path().join("mur");
std::fs::write(
&script,
format!(
"#!/bin/sh\n\
case \" $* \" in\n\
*' --capsule coder '*) name=coder; line='{coder}';;\n\
*' --capsule reviewer '*) name=reviewer; line='{reviewer}';;\n\
*) name=planner; line='';;\n\
esac\n\
env > '{out}/'$name.env\n\
printf '%s\\n' \"$*\" > '{out}/'$name.argv\n\
cat /dev/fd/$MURMUR_FORMATION_CHANNEL >> '{out}/'$name.channel &\n\
[ -n \"$line\" ] && printf '%s\\n' \"$line\"\n\
exec sleep 600\n",
coder = line("ses_coder", coder_port),
reviewer = line("ses_reviewer", reviewer_port),
),
)
.unwrap();
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
std::env::set_var(child_launch::MUR_BINARY_ENV, &script);
let plan = FormationPlan {
peers: vec![member("coder"), member("reviewer")],
entry: member("planner"),
callees: vec![
("planner".to_string(), vec!["coder".to_string()]),
("coder".to_string(), vec!["reviewer".to_string()]),
],
};
let result = launch_plan(plan, options);
let mut formation = result.expect("both stand-ins came up");
formation.start_entry().expect("the entry stand-in starts");
std::env::remove_var(child_launch::MUR_BINARY_ENV);
let read_lines = |name: &str, count: usize| -> Vec<String> {
let path = dir.path().join(format!("{name}.channel"));
let deadline = Instant::now() + Duration::from_secs(10);
loop {
let lines: Vec<String> = std::fs::read_to_string(&path)
.unwrap_or_default()
.lines()
.map(str::to_string)
.collect();
if lines.len() >= count || Instant::now() >= deadline {
return lines;
}
std::thread::sleep(POLL_INTERVAL);
}
};
let coder = read_lines("coder", 2);
let planner = read_lines("planner", 2);
let reviewer = read_lines("reviewer", 1);
std::thread::sleep(Duration::from_millis(300));
assert_eq!(read_lines("reviewer", 1).len(), 1, "reviewer calls nobody");
let bundle = |line: &str| FormationBundle::parse_line(line).expect("a member's bundle");
let names = |bundle: &FormationBundle| -> Vec<String> {
bundle.calls.iter().map(|(name, _)| name.clone()).collect()
};
let (coder_bundle, planner_bundle, reviewer_bundle) =
(bundle(&coder[0]), bundle(&planner[0]), bundle(&reviewer[0]));
assert_eq!(coder_bundle.member, "coder");
assert_eq!(names(&coder_bundle), ["reviewer"]);
assert_eq!(names(&planner_bundle), ["coder"]);
assert!(names(&reviewer_bundle).is_empty());
let launcher = crate::running::ProcessIdentity::of_this_process();
assert!(launcher.is_some());
for bundle in [&coder_bundle, &planner_bundle, &reviewer_bundle] {
assert_eq!(bundle.formation_id, id);
assert_eq!(bundle.verify_key, coder_bundle.verify_key);
assert_eq!(bundle.launcher, launcher, "{}", bundle.member);
}
assert!(!planner.join("\n").contains("reviewer"), "{planner:?}");
let addresses = |line: &str| -> Vec<(String, String)> {
parse_address_line(line)
.unwrap()
.into_iter()
.map(|peer| (peer.name, peer.url))
.collect()
};
assert_eq!(
addresses(&coder[1]),
[(
"reviewer".to_string(),
format!("http://localhost:{reviewer_port}")
)]
);
assert_eq!(
addresses(&planner[1]),
[(
"coder".to_string(),
format!("http://localhost:{coder_port}")
)]
);
for name in ["coder", "reviewer", "planner"] {
let env = std::fs::read_to_string(dir.path().join(format!("{name}.env"))).unwrap();
let argv = std::fs::read_to_string(dir.path().join(format!("{name}.argv"))).unwrap();
assert!(!env.contains("mft1.") && !argv.contains("mft1."), "{name}");
assert!(!env.contains(FORMATION_PEERS_ENV), "{name}: {env}");
assert!(
env.contains(&format!("{FORMATION_CHANNEL_ENV}=")),
"{name}: {env}"
);
}
let pids: Vec<u32> = formation.peers().iter().map(|peer| peer.pid).collect();
drop(formation);
for pid in pids {
assert!(!crate::running::pid_is_alive(pid), "{pid}");
}
}
#[test]
fn a_registered_member_is_killed_by_the_panic_hooks_sweep() {
use std::os::unix::process::CommandExt;
let _guard = binary_guard();
let mut child = std::process::Command::new("sleep")
.arg("600")
.process_group(0)
.spawn()
.unwrap();
let slot = RegistrySlot::register(child.id(), true);
assert!(slot.0.is_some());
kill_registered_members();
let status = child.wait().unwrap();
use std::os::unix::process::ExitStatusExt;
assert_eq!(status.signal(), Some(libc::SIGKILL));
drop(slot);
assert!(MEMBER_REGISTRY
.iter()
.all(|slot| slot.load(Ordering::SeqCst) != i64::from(child.id())));
}
#[test]
fn a_launch_failure_names_each_member_and_every_stopped_one() {
let failure = MemberLaunchFailure {
failed: vec![failure(
&member("broken"),
MemberFailureReason::ExitedBeforeReporting {
status: "status 1".to_string(),
stderr_tail: vec!["error[E-RUN-008]: missing".to_string()],
},
)],
stopped: vec![
StoppedMember {
name: "coder".to_string(),
pid: 7,
},
StoppedMember {
name: "reviewer".to_string(),
pid: 8,
},
],
interrupted_by: None,
};
assert_eq!(
failure.to_string(),
"the formation could not be launched\n 'broken' (broken@0.1.0): its process exited \
with status 1 before reporting; its last stderr lines were:\n \
error[E-RUN-008]: missing\n stopped: coder (pid 7), reviewer (pid 8)"
);
}
#[test]
fn ready_peer_debug_prints_no_door_token() {
let peer = ReadyPeer {
name: "coder".to_string(),
capsule: "coder".to_string(),
version: "1.2.0".to_string(),
pid: 1,
session_id: "ses_x".to_string(),
url: "http://localhost:1".to_string(),
workdir: PathBuf::from("/home/u/.murmur/formations/frm_x/coder"),
door_token: Some(DoorToken::new("mdt1.secretpayload.mac".to_string())),
};
let debug = format!("{peer:?}");
assert!(!debug.contains("secretpayload"), "{debug}");
assert!(debug.contains("<redacted>"), "{debug}");
}
#[test]
fn a_member_directory_is_made_owner_only_under_its_formation() {
use std::os::unix::fs::PermissionsExt;
let home = tempfile::tempdir().unwrap();
let formations = home.path().join(FORMATIONS_DIR);
let id = FormationId::mint();
let workdir = make_member_workdir(&formations, &id, "worker").unwrap();
assert_eq!(workdir, formations.join(id.as_str()).join("worker"));
for dir in [&formations, &formations.join(id.as_str()), &workdir] {
let mode = std::fs::metadata(dir).unwrap().permissions().mode() & 0o777;
assert_eq!(mode, 0o700, "{}", dir.display());
}
}
#[test]
fn an_existing_member_path_is_refused_and_named() {
let home = tempfile::tempdir().unwrap();
let formations = home.path().join(FORMATIONS_DIR);
let id = FormationId::mint();
make_member_workdir(&formations, &id, "worker").unwrap();
let again = make_member_workdir(&formations, &id, "worker").unwrap_err();
let path = formations.join(id.as_str()).join("worker");
assert!(again.contains(&path.display().to_string()), "{again}");
std::os::unix::fs::symlink(home.path(), formations.join(id.as_str()).join("linked"))
.unwrap();
assert!(make_member_workdir(&formations, &id, "linked").is_err());
}
#[test]
fn a_peer_whose_directory_cannot_be_made_refuses_the_launch() {
let _guard = binary_guard();
let dir = tempfile::tempdir().unwrap();
let options = options(Duration::from_secs(5));
let formations = options.formations_dir.clone().unwrap();
let taken = formations.join(options.formation_id.as_str()).join("coder");
std::fs::create_dir_all(&taken).unwrap();
std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
let result = launch_plan(plan_of("coder"), options);
std::env::remove_var(child_launch::MUR_BINARY_ENV);
let Err(failure) = result else {
panic!("the launch was expected to be refused");
};
assert!(failure.stopped.is_empty(), "{failure:?}");
match only_reason(&failure) {
MemberFailureReason::NotStarted { error } => {
assert!(error.contains(&taken.display().to_string()), "{error}")
}
other => panic!("expected NotStarted, got {other:?}"),
}
}
#[test]
fn member_roots_are_the_member_directories_that_hold_sessions() {
assert!(formation_member_roots(&FormationId::mint()).is_empty());
let id = FormationId::mint();
let home = tempfile::tempdir().unwrap();
let formations = home.path().join(FORMATIONS_DIR);
for member in ["worker", "reviewer", "idle"] {
assert_eq!(
make_member_workdir(&formations, &id, member).unwrap(),
formations.join(id.as_str()).join(member)
);
}
for member in ["worker", "reviewer"] {
std::fs::create_dir(formations.join(id.as_str()).join(member).join(".murmur")).unwrap();
}
assert_eq!(
formation_member_roots_in(&formations, &id),
[
formations.join(id.as_str()).join("reviewer/.murmur"),
formations.join(id.as_str()).join("worker/.murmur"),
]
);
}
#[test]
fn formation_marker_round_trips_as_three_keys() {
let marker = FormationMarker {
project_dir: PathBuf::from("/srv/project"),
launcher_pid: 4242,
launcher_start: "12345".to_string(),
};
let value = serde_json::to_value(&marker).unwrap();
assert_eq!(
value,
serde_json::json!({
"project_dir": "/srv/project",
"launcher_pid": 4242,
"launcher_start": "12345",
})
);
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join(FORMATION_MARKER_FILE), value.to_string()).unwrap();
assert_eq!(FormationMarker::read(dir.path()), Some(marker));
std::fs::write(dir.path().join(FORMATION_MARKER_FILE), "{").unwrap();
assert_eq!(FormationMarker::read(dir.path()), None);
assert_eq!(FormationMarker::read(&dir.path().join("missing")), None);
}
#[test]
fn formation_marker_is_written_owner_only_before_the_first_peer_directory() {
use std::os::unix::fs::PermissionsExt;
let _guard = binary_guard();
let dir = tempfile::tempdir().unwrap();
let project = tempfile::tempdir().unwrap();
let mut options = options(Duration::from_secs(1));
options.project_dir = project.path().join(".");
let formation_dir = options
.formations_dir
.clone()
.unwrap()
.join(options.formation_id.as_str());
std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
let result = launch_plan(plan_of("coder"), options);
std::env::remove_var(child_launch::MUR_BINARY_ENV);
assert!(result.is_err(), "the stand-in never reports itself");
let path = formation_dir.join(FORMATION_MARKER_FILE);
let metadata = std::fs::metadata(&path).unwrap();
assert_eq!(metadata.permissions().mode() & 0o777, 0o600);
let marker = FormationMarker::read(&formation_dir).unwrap();
let pid = std::process::id();
assert_eq!(
marker,
FormationMarker {
project_dir: std::fs::canonicalize(project.path()).unwrap(),
launcher_pid: pid,
launcher_start: crate::running::process_start_token(pid).unwrap_or_default(),
}
);
let peer_dir = std::fs::metadata(formation_dir.join("coder")).unwrap();
assert!(
metadata.modified().unwrap() <= peer_dir.modified().unwrap(),
"the marker was written after the peer's directory was made"
);
}
#[test]
fn formation_marker_is_never_written_without_peers() {
let _guard = binary_guard();
let dir = tempfile::tempdir().unwrap();
let options = options(Duration::from_secs(5));
let formations = options.formations_dir.clone().unwrap();
std::env::set_var(child_launch::MUR_BINARY_ENV, stand_in(dir.path(), None));
let plan = FormationPlan {
peers: Vec::new(),
entry: member("planner"),
callees: Vec::new(),
};
let result = launch_plan(plan, options);
std::env::remove_var(child_launch::MUR_BINARY_ENV);
let formation = result.expect("a formation with no peers has nothing to wait for");
drop(formation);
assert!(
std::fs::read_dir(&formations).map_or(true, |mut entries| entries.next().is_none()),
"{}",
formations.display()
);
}
fn marker_of(pid: u32, start: &str) -> FormationMarker {
FormationMarker {
project_dir: PathBuf::from("/srv/project"),
launcher_pid: pid,
launcher_start: start.to_string(),
}
}
fn reaped_pid() -> u32 {
let mut child = std::process::Command::new("true").spawn().unwrap();
let pid = child.id();
child.wait().unwrap();
pid
}
fn running_record(pid: u32, formation: Option<&FormationId>) -> crate::running::RunningRecord {
crate::running::RunningRecord {
session_id: "ses_0199c4e2f1b7712a9d3e4f5061728394".to_string(),
url: "127.0.0.1:1".to_string(),
pid,
process_start: crate::running::process_start_token(pid).unwrap_or_default(),
capsule_name: "worker".to_string(),
capsule_version: "0.1.0".to_string(),
workdir: PathBuf::from("/tmp/worker"),
outlives_launcher: false,
started_at: "2026-01-01T00:00:00Z".to_string(),
door_token: None,
credentials: Default::default(),
formation_id: formation.cloned(),
formation_lifeline: false,
formation_launcher: None,
spawned_by: None,
}
}
#[test]
fn prune_formations_liveness_reads_the_launcher_from_the_marker() {
let id = FormationId::mint();
let own = std::process::id();
let token = crate::running::process_start_token(own).unwrap();
assert!(formation_is_live(&id, &marker_of(own, &token), Some(&[])));
assert!(!formation_is_live(
&id,
&marker_of(reaped_pid(), &token),
Some(&[])
));
}
#[test]
fn prune_formations_liveness_reads_running_records_that_name_the_formation() {
let id = FormationId::mint();
let dead = marker_of(reaped_pid(), "1");
let mut sleeper = std::process::Command::new("sleep")
.arg("30")
.spawn()
.unwrap();
let live = running_record(sleeper.id(), Some(&id));
let other = running_record(sleeper.id(), Some(&FormationId::mint()));
let unnamed = running_record(sleeper.id(), None);
let gone = running_record(reaped_pid(), Some(&id));
assert!(formation_is_live(&id, &dead, Some(&[live])));
assert!(!formation_is_live(
&id,
&dead,
Some(&[other, unnamed, gone])
));
let _ = sleeper.kill();
let _ = sleeper.wait();
}
#[test]
fn prune_formations_liveness_treats_an_unlistable_running_directory_as_all_live() {
let id = FormationId::mint();
assert!(formation_is_live(&id, &marker_of(reaped_pid(), "1"), None));
let home = tempfile::tempdir().unwrap();
let project = tempfile::tempdir().unwrap();
let formations = home.path().join(FORMATIONS_DIR);
let old = FormationId::parse(&format!("frm_{:012x}{:020x}", 1_000_000u64, 1)).unwrap();
let old_dir = formations.join(old.as_str());
std::fs::create_dir_all(old_dir.join("worker")).unwrap();
let marker = FormationMarker {
project_dir: std::fs::canonicalize(project.path()).unwrap(),
..marker_of(reaped_pid(), "1")
};
std::fs::write(
old_dir.join(FORMATION_MARKER_FILE),
serde_json::to_string(&marker).unwrap(),
)
.unwrap();
let policy = murmur_artifact::TraceRetainConfig {
max_sessions: Some(1),
max_age_secs: Some(1),
};
let running = home.path().join("running");
std::fs::write(&running, "").unwrap();
let current = FormationId::mint();
assert!(prune_ended_formations_in(
&formations,
&running,
¤t,
project.path(),
&policy
)
.is_empty());
assert!(old_dir.is_dir());
std::fs::remove_file(&running).unwrap();
let pruned =
prune_ended_formations_in(&formations, &running, ¤t, project.path(), &policy);
assert_eq!(pruned.len(), 1, "{pruned:?}");
assert_eq!(pruned[0].formation_id, old);
assert_eq!(pruned[0].reason, crate::trace::RETENTION_REASON_MAX_AGE);
assert!(!old_dir.exists());
}
}