use std::{
cell::RefCell,
fs,
path::{Path, PathBuf},
rc::Rc,
time::Duration,
};
use crate::support::os;
pub struct RetainedInputs {
_directory: Rc<RefCell<Option<tempfile::TempDir>>>,
}
struct Handoff {
pid: u32,
identities: Vec<(u32, u64)>,
}
pub struct Launched {
root: PathBuf,
inputs: Rc<RefCell<Option<tempfile::TempDir>>>,
processes: Option<super::Processes>,
handoff: Option<Handoff>,
}
impl Launched {
pub fn new(root: tempfile::TempDir) -> Self {
Self {
root: root.path().to_owned(),
inputs: Rc::new(RefCell::new(Some(root))),
processes: Some(super::Processes::default()),
handoff: None,
}
}
pub fn path(&self) -> &Path {
&self.root
}
pub fn retain_inputs(&self) -> RetainedInputs {
RetainedInputs {
_directory: self.inputs.clone(),
}
}
pub fn own(&mut self) -> Option<u32> {
if let Some(handoff) = &self.handoff {
return Some(handoff.pid);
}
let pid = fs::read_to_string(self.path().join("pid"))
.ok()?
.parse()
.ok()?;
let pids = match fs::read(self.path().join("pids")) {
Ok(bytes) => serde_json::from_slice::<Vec<u32>>(&bytes).ok()?,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Vec::new(),
Err(_) => return None,
};
let pids = pids
.into_iter()
.chain([pid])
.collect::<std::collections::BTreeSet<_>>();
for &pid in &pids {
self.processes.as_mut().unwrap().own(pid);
}
let identities = pids
.into_iter()
.map(|pid| os::observed_start_time(pid).map(|start| (pid, start)))
.collect::<Option<Vec<_>>>()?;
self.handoff = Some(Handoff { pid, identities });
Some(pid)
}
pub fn close(mut self) -> Result<(), String> {
self.release()
}
fn release(&mut self) -> Result<(), String> {
if self.processes.is_none() {
return Ok(());
}
let _ = fs::write(self.path().join("release-handoff"), "");
const HANDOFF_BOUND: Duration = Duration::from_secs(30);
const HANDOFF_POLL: Duration = Duration::from_millis(10);
let deadline = std::time::Instant::now() + os::patience(HANDOFF_BOUND);
let adopted = loop {
if self.own().is_some() {
break true;
}
if std::time::Instant::now() >= deadline {
break false;
}
std::thread::sleep(HANDOFF_POLL);
};
let released = fs::write(self.path().join("release"), "");
let joined = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
drop(self.processes.take());
}));
let exited = self.handoff.as_ref().is_some_and(|handoff| {
handoff
.identities
.iter()
.all(|&(pid, start)| !os::same_process_running(pid, start))
});
if !adopted || released.is_err() || joined.is_err() || !exited {
let retained = self.inputs.borrow_mut().take().unwrap().keep();
return Err(format!(
"Detached launch cleanup failed; inputs retained at {}",
os::path_text(&retained)
));
}
Ok(())
}
}
impl Drop for Launched {
fn drop(&mut self) {
if let Err(message) = self.release() {
if std::thread::panicking() {
eprintln!("{message}");
} else {
panic!("{message}");
}
}
}
}
pub const LAUNCH: &str = r"import json, os, pathlib, sys, time
root = pathlib.Path(sys.argv[1]) if len(sys.argv) > 1 else pathlib.Path.cwd()
if (root / 'hold-handoff').exists():
(root / 'handoff-ready').write_text('')
while not (root / 'release-handoff').exists(): time.sleep(0.01)
for name, contents in [('pids', json.dumps(globals().get('OWNED_PIDS', [os.getpid()]))), ('pid', str(os.getpid())), ('marker', os.environ.get('ARTIFACTIZE_HUMAN_MARKER', 'launched'))]:
pending = root / (name + '.pending')
pending.write_text(contents)
pending.replace(root / name)
while not (root / 'release').exists(): time.sleep(0.01)
";