Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,12 @@ and listed with their assets on the

### Fixed

- **A hub that stopped answering no longer blocks every start** (#398): it
held its lock, so `devstack up`, `devstack ui`, `devstack hub start` and
the MCP server each failed with "another devstack instance owns …\hub\lock"
until it was ended by hand. A hub that runs, holds the lock, started more
than a minute ago and does not answer on its port is now ended by the
next start, which takes its place (`hub_hung` in the hub's log).
- **A node no longer leaves a gone lane's containers running** (#394): a
slot removed without its purge (one without `slot.json`, or by an agent
of an earlier release) left its Compose projects and networks running
Expand Down
65 changes: 65 additions & 0 deletions crates/devstack-hub/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3183,6 +3183,7 @@ pub fn ensure_running() -> Result<u16, String> {
Listener::Foreign => return Err(foreign(port)),
Listener::Nothing => {}
}
end_hung_hub(port);
let exe = devstack_exe().map_err(|e| e.to_string())?;
// What the hub prints before its own log exists (a failed start) goes to
// hub.out, like the daemon's daemon.out.
Expand Down Expand Up @@ -3228,6 +3229,48 @@ pub fn ensure_running() -> Result<u16, String> {
))
}

/// How long a hub may take to answer after it started before one that holds
/// the lock and does not answer counts as hung.
const HUNG_AFTER: Duration = Duration::from_secs(60);

/// A hub that runs, holds the state lock and does not answer on `port`
/// (seen: its runtime stuck, every start after it refused by the lock) is
/// ended, so the start that follows can take its place. One that just
/// started, or a lock nobody holds, is left alone.
fn end_hung_hub(port: u16) {
let Some(info) = read_info() else { return };
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default();
if !is_hung(&info, port, now, devstack_proc::pid_alive(info.pid)) {
return;
}
// Nobody holds the lock: that pid is no hub (any more).
if devstack_proc::StateLock::acquire(&state_dir()).is_ok() {
return;
}
diag::warn(
"hub_hung",
serde_json::json!({ "pid": info.pid, "port": port, "version": info.version }),
);
if devstack_proc::end_pid(info.pid).is_err() {
return;
}
let t0 = Instant::now();
while devstack_proc::pid_alive(info.pid) && t0.elapsed() < Duration::from_secs(10) {
std::thread::sleep(Duration::from_millis(100));
}
let _ = std::fs::remove_file(state_dir().join("hub.json"));
}

/// Whether the hub `info` names counts as hung: it runs, it said it listens
/// on `port`, and it started more than [`HUNG_AFTER`] before `now` (the
/// caller found nothing answering there).
fn is_hung(info: &HubInfo, port: u16, now: Duration, alive: bool) -> bool {
let started = Duration::from_millis(u64::try_from(info.started).unwrap_or(u64::MAX));
alive && info.port == port && now.saturating_sub(started) >= HUNG_AFTER
}

/// Is a hub (or anything) answering on `port`?
pub fn answers(port: u16) -> bool {
std::net::TcpStream::connect_timeout(
Expand Down Expand Up @@ -3848,3 +3891,25 @@ mod daemon_send_tests {
assert_eq!(daemon_in(d.path()), DaemonNow::Absent { starting: false });
}
}

#[cfg(test)]
mod hung_hub_tests {
use super::*;

#[test]
fn a_hub_is_hung_when_it_runs_listened_here_and_is_old_enough() {
let info = HubInfo {
pid: 42,
port: 4250,
version: "0.32.0".into(),
started: 1_000_000,
};
let at = |s: u64| Duration::from_millis(1_000_000) + Duration::from_secs(s);
assert!(is_hung(&info, 4250, at(61), true));
// Just started: it may still be coming up.
assert!(!is_hung(&info, 4250, at(10), true));
// Gone, or a hub of another port.
assert!(!is_hung(&info, 4250, at(61), false));
assert!(!is_hung(&info, 4251, at(61), true));
}
}
41 changes: 41 additions & 0 deletions crates/devstack-proc/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,47 @@ pub fn pid_alive(pid: u32) -> bool {
backend::pid_alive(pid)
}

/// End the process `pid`, one devstack did not start (a daemon or a hub
/// that does not answer): Windows terminates it, which closes its job
/// objects and with them every process tree it owns (ADR-0009); elsewhere
/// `SIGTERM`, then `SIGKILL` after 5 s.
pub fn end_pid(pid: u32) -> std::io::Result<()> {
#[cfg(windows)]
{
use windows_sys::Win32::Foundation::CloseHandle;
use windows_sys::Win32::System::Threading::{
OpenProcess, PROCESS_TERMINATE, TerminateProcess,
};
// SAFETY: plain Win32 calls on a handle we open and close here.
unsafe {
let h = OpenProcess(PROCESS_TERMINATE, 0, pid);
if h.is_null() {
return Err(std::io::Error::last_os_error());
}
let ok = TerminateProcess(h, 1) != 0;
let err = std::io::Error::last_os_error();
CloseHandle(h);
if ok { Ok(()) } else { Err(err) }
}
}
#[cfg(not(windows))]
{
// SAFETY: a signal to one process id.
if unsafe { libc::kill(pid as i32, libc::SIGTERM) } != 0 {
return Err(std::io::Error::last_os_error());
}
let t0 = std::time::Instant::now();
while pid_alive(pid) && t0.elapsed() < std::time::Duration::from_secs(5) {
std::thread::sleep(std::time::Duration::from_millis(100));
}
if pid_alive(pid) {
// SAFETY: as above.
unsafe { libc::kill(pid as i32, libc::SIGKILL) };
}
Ok(())
}
}

/// What a process tree costs right now (plan §12). Callers derive CPU load
/// from two samples of `cpu_us`.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)]
Expand Down
42 changes: 1 addition & 41 deletions crates/devstack/src/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -446,46 +446,6 @@ pub(crate) fn pid_alive(pid: u32) -> bool {
}
}

/// End the process `pid` (a daemon that does not answer): Windows
/// terminates it, which closes its job objects and with them every process
/// tree it owns (ADR-0009); elsewhere `SIGTERM`, then `SIGKILL` after 5 s.
fn end_process(pid: u32) -> std::io::Result<()> {
#[cfg(windows)]
{
use windows_sys::Win32::Foundation::CloseHandle;
use windows_sys::Win32::System::Threading::{
OpenProcess, PROCESS_TERMINATE, TerminateProcess,
};
// SAFETY: plain Win32 calls on a handle we open and close here.
unsafe {
let h = OpenProcess(PROCESS_TERMINATE, 0, pid);
if h.is_null() {
return Err(std::io::Error::last_os_error());
}
let ok = TerminateProcess(h, 1) != 0;
let err = std::io::Error::last_os_error();
CloseHandle(h);
if ok { Ok(()) } else { Err(err) }
}
}
#[cfg(not(windows))]
{
// SAFETY: a signal to one process id.
if unsafe { libc::kill(pid as i32, libc::SIGTERM) } != 0 {
return Err(std::io::Error::last_os_error());
}
let t0 = Instant::now();
while pid_alive(pid) && t0.elapsed() < Duration::from_secs(5) {
std::thread::sleep(Duration::from_millis(100));
}
if pid_alive(pid) {
// SAFETY: as above.
unsafe { libc::kill(pid as i32, libc::SIGKILL) };
}
Ok(())
}
}

/// Nothing listens (yet): the connection was refused, unlike a daemon that
/// accepted and did not answer.
fn is_refused(e: &anyhow::Error) -> bool {
Expand Down Expand Up @@ -2498,7 +2458,7 @@ fn end_unanswering_daemon(state_dir: &Path, why: anyhow::Error, all: bool) -> Re
}
eprintln!("{why:#}");
println!("ending the daemon by its pid ({})", info.pid);
end_process(info.pid).with_context(|| format!("cannot end process {}", info.pid))?;
devstack_proc::end_pid(info.pid).with_context(|| format!("cannot end process {}", info.pid))?;
let t0 = Instant::now();
while pid_alive(info.pid) && t0.elapsed() < Duration::from_secs(15) {
std::thread::sleep(Duration::from_millis(200));
Expand Down
Loading