server: extract a ProcessGroup guard for teardown
Assisted-by: Claude Opus 5.5 via Claude Code
diff --git a/src/server.rs b/src/server.rs
index e39f246..484b1ab 100644
--- a/src/server.rs
+++ b/src/server.rs
@@ -18,7 +18,6 @@ use std::time::Duration;
use std::time::Instant;
use camino::Utf8Path;
-use camino::Utf8PathBuf;
use serde::Deserialize;
use serde_json::Value;
@@ -51,18 +50,11 @@ pub(crate) struct Handshake {
pub(crate) data: Value,
}
-/// A running app: the parsed handshake plus everything teardown
-/// needs. Dropping without calling `stop` SIGKILLs the group.
+/// A running app. Dropping it without calling `stop` SIGKILLs its group.
#[derive(Debug)]
pub(crate) struct Server {
- handshake: Option<Handshake>,
- child: Child,
- /// The child's pid, which is also its pgid: it was spawned as the
- /// leader of a fresh group.
- pgid: i32,
- log: Utf8PathBuf,
- stopped: bool,
- /// Keeps the directory holding the handshake file alive.
+ handshake: Handshake,
+ group: ProcessGroup,
_dir: camino_tempfile::Utf8TempDir,
}
@@ -86,40 +78,13 @@ impl Server {
expect(dead_code, reason = "read once the flow context lands")
)]
pub(crate) fn handshake(&self) -> &Handshake {
- #[expect(
- clippy::expect_used,
- reason = "start() only yields a Server once the handshake parsed"
- )]
- self.handshake
- .as_ref()
- .expect("a started server has a handshake")
+ &self.handshake
}
- /// SIGTERM the group, wait out the grace period, then SIGKILL
- /// whatever is left — including stragglers a wrapper left behind
- /// after its own exit.
- #[expect(
- clippy::arithmetic_side_effects,
- reason = "Instant + Duration only panics near heat death"
- )]
- pub(crate) async fn stop(mut self) -> Result<(), Error> {
- tokio::task::spawn_blocking(move || {
- signal_group(self.pgid, libc::SIGTERM).map_err(Error::Io)?;
- let deadline = Instant::now() + GRACE;
- loop {
- if self.child.try_wait().map_err(Error::Io)?.is_some() {
- break;
- }
- if Instant::now() >= deadline {
- break;
- }
- std::thread::sleep(POLL);
- }
- self.reap();
- Ok(())
- })
- .await
- .map_err(|error| Error::Io(io::Error::other(error)))?
+ pub(crate) async fn stop(self) -> Result<(), Error> {
+ tokio::task::spawn_blocking(move || self.group.terminate().map_err(Error::Io))
+ .await
+ .map_err(|error| Error::Io(io::Error::other(error)))?
}
fn spawn_and_wait(
@@ -133,86 +98,88 @@ impl Server {
let variant_dir = output.join(variant);
fs_err::create_dir_all(&variant_dir).map_err(Error::Io)?;
let log = variant_dir.join("server.log");
- let file = fs_err::File::create(&log).map_err(Error::Io)?;
+ let (file, _) = fs_err::File::create(&log).map_err(Error::Io)?.into_parts();
- // A fresh process group makes the child its own leader, so the
- // group signal reaches everything a wrapper like `sh -c` or
- // `bundle exec` spawned, and only those processes.
let mut command = Command::new(&spec.program);
command
.args(&spec.args)
.env("PARALLAX_HANDSHAKE", &handshake_path)
.stdin(Stdio::null())
- .stdout({
- #[expect(
- clippy::disallowed_types,
- reason = "Stdio converts only from std files"
- )]
- let clone: std::fs::File = file.try_clone().map_err(Error::Io)?.into();
- Stdio::from(clone)
- })
- .stderr({
- #[expect(
- clippy::disallowed_types,
- reason = "Stdio converts only from std files"
- )]
- let log: std::fs::File = file.into();
- Stdio::from(log)
- })
- .process_group(0);
- let child = command.spawn().map_err(|error| Error::ServerSpawn {
+ .stdout(file.try_clone().map_err(Error::Io)?)
+ .stderr(file);
+ let mut group = ProcessGroup::spawn(&mut command).map_err(|error| Error::ServerSpawn {
cmd: spec.command_line(),
source: error,
})?;
- let mut server = Server {
- handshake: None,
- pgid: child.id() as i32,
- child,
- log,
- stopped: false,
- _dir: dir,
- };
- let child = &mut server.child;
- let outcome = wait_for_handshake(&handshake_path, spec.timeout, || child.try_wait())
+ let outcome = wait_for_handshake(&handshake_path, spec.timeout, || group.try_wait())
.map_err(Error::Io)?;
- match outcome {
- Ok(handshake) => {
- server.handshake = Some(handshake);
- Ok(server)
- }
- Err(HandshakeError::Exited(status)) => {
- server.reap();
- Err(Error::ServerExited {
- variant: variant.to_owned(),
- status,
- tail: tail(&server.log),
- })
+ let handshake = match outcome {
+ Ok(handshake) => handshake,
+ Err(failure) => {
+ // Reaping the group first means its log is complete.
+ drop(group);
+ let tail = tail(&log);
+ let variant = variant.to_owned();
+ return Err(match failure {
+ HandshakeError::Exited(status) => Error::ServerExited {
+ variant,
+ status,
+ tail,
+ },
+ HandshakeError::Timeout => Error::ServerTimeout {
+ variant,
+ timeout: spec.timeout,
+ tail,
+ },
+ });
}
- Err(HandshakeError::Timeout) => {
- server.reap();
- Err(Error::ServerTimeout {
- variant: variant.to_owned(),
- timeout: spec.timeout,
- tail: tail(&server.log),
- })
- }
- }
+ };
+ Ok(Server {
+ handshake,
+ group,
+ _dir: dir,
+ })
}
+}
- /// SIGKILLs the group and reaps the child, making the log complete.
- fn reap(&mut self) {
- self.stopped = true;
- let _ = signal_group(self.pgid, libc::SIGKILL);
- let _ = self.child.wait();
+/// A child spawned as the leader of its own process group, so signals
+/// reach everything a wrapper like `sh -c` or `bundle exec` started.
+/// Dropping it SIGKILLs the group and reaps the leader.
+#[derive(Debug)]
+struct ProcessGroup {
+ leader: Child,
+ pgid: i32,
+}
+
+impl ProcessGroup {
+ fn spawn(command: &mut Command) -> io::Result<ProcessGroup> {
+ let leader = command.process_group(0).spawn()?;
+ Ok(ProcessGroup {
+ pgid: leader.id() as i32,
+ leader,
+ })
+ }
+
+ fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
+ self.leader.try_wait()
+ }
+
+ /// SIGTERM, then the grace period; the drop SIGKILLs any stragglers.
+ fn terminate(mut self) -> io::Result<()> {
+ signal_group(self.pgid, libc::SIGTERM)?;
+ let started = Instant::now();
+ while self.try_wait()?.is_none() && started.elapsed() < GRACE {
+ std::thread::sleep(POLL);
+ }
+ Ok(())
}
}
-impl Drop for Server {
+impl Drop for ProcessGroup {
fn drop(&mut self) {
- if !self.stopped {
- self.reap();
- }
+ let _ = signal_group(self.pgid, libc::SIGKILL);
+ let _ = self.leader.wait();
}
}
@@ -227,16 +194,12 @@ enum HandshakeError {
/// unreadable or unparseable — including a missing `url` — keeps
/// waiting until the deadline. `exited` is `Child::try_wait`; the outer
/// error is parallax failing to check, the inner one the app failing.
-#[expect(
- clippy::arithmetic_side_effects,
- reason = "Instant + Duration only panics near heat death"
-)]
fn wait_for_handshake(
path: &Utf8Path,
timeout: Duration,
mut exited: impl FnMut() -> io::Result<Option<ExitStatus>>,
) -> io::Result<Result<Handshake, HandshakeError>> {
- let deadline = Instant::now() + timeout;
+ let started = Instant::now();
loop {
if let Some(handshake) = read_handshake(path) {
return Ok(Ok(handshake));
@@ -244,7 +207,7 @@ fn wait_for_handshake(
if let Some(status) = exited()? {
return Ok(Err(HandshakeError::Exited(status)));
}
- if Instant::now() >= deadline {
+ if started.elapsed() >= timeout {
return Ok(Err(HandshakeError::Timeout));
}
std::thread::sleep(POLL);
@@ -295,11 +258,7 @@ fn tail(path: &Utf8Path) -> String {
#[cfg(test)]
mod tests {
- #![allow(
- clippy::arithmetic_side_effects,
- clippy::expect_used,
- clippy::unwrap_used
- )]
+ #![allow(clippy::expect_used)]
use super::*;
use serde_json::json;
@@ -375,7 +334,7 @@ mod tests {
)
.await
.expect("server starts");
- let pgid = server.pgid;
+ let pgid = server.group.pgid;
server.stop().await.expect("server stops");
assert_group_gone(pgid);
}
@@ -391,7 +350,7 @@ mod tests {
)
.await
.expect("server starts");
- let pgid = server.pgid;
+ let pgid = server.group.pgid;
drop(server);
assert_group_gone(pgid);
}
@@ -410,7 +369,7 @@ mod tests {
)
.await
.expect("server starts");
- let log = server.log.clone();
+ let log = output.path().join("desktop").join("server.log");
server.stop().await.expect("server stops");
let text = fs_err::read_to_string(&log).expect("log readable");
assert!(text.contains("terminated-by-term"));
@@ -429,7 +388,7 @@ mod tests {
)
.await
.expect("server starts");
- let pgid = server.pgid;
+ let pgid = server.group.pgid;
let started = Instant::now();
server.stop().await.expect("server stops");
let waited = started.elapsed();
@@ -512,8 +471,7 @@ mod tests {
)
.await
.expect("server starts");
- let log = server.log.clone();
- assert_eq!(log, output.path().join("desktop").join("server.log"));
+ let log = output.path().join("desktop").join("server.log");
server.stop().await.expect("server stops");
let text = fs_err::read_to_string(&log).expect("log readable");
assert!(text.contains("an-out-line"));