diff --git a/docs/lifecycle-supervisor.md b/docs/lifecycle-supervisor.md index 4d1938f9f..1fe06911f 100644 --- a/docs/lifecycle-supervisor.md +++ b/docs/lifecycle-supervisor.md @@ -58,33 +58,50 @@ a failed capture does not prevent that healthy guard from handling shutdown. If is lost during stop, recovery prefers a fresh capture and falls back to those admitted identities, rechecking each before TERM/KILL. Each recovered target gets its own five-second TERM grace and two-second confirmation bound. Recovery begins -at most once per target; late guard loss can extend stop to roughly fourteen seconds. Unobserved descendants -still report unknown. A failed or expired target does not abandon other admitted -targets: stop drives them to their own outcomes before returning unknown. +at most once per target; late guard loss can extend stop to roughly fourteen seconds. Descendants forked after the capture are never +signalled: once no captured member is left, stop waits until the target's own deadline for them to +exit and reports unknown only if they are still alive. A failed or expired target does not abandon other admitted +targets: stop drives them to their own outcomes before returning unknown. A receipt that +cannot be verified at admission is never signalled, but it no longer keeps the owner's +other targets from being stopped; the reply is still `unknown` with `stop-incomplete`. Replies contain `version`, `supervisorPid`, `processes`, and `outcome` (`ready`, `observed`, `stopped`, `unknown` or `refused`). A stop whose confirmation times out emits `unknown` and exits nonzero. Callers must inspect both the receipt and exit status. -Every `stop` and `status` request ends with exactly one framed receipt, also when it -fails. A failed request exits nonzero and carries a `reason`: +Every `stop` and `status` request whose frame is read ends with exactly one framed +receipt, also when it fails. A failed request exits nonzero and carries a `reason`: | `reason` | `outcome` | Meaning | |---|---|---| | `invalid-request` | `refused` | The control frame or root was rejected; nothing was read or changed. | | `stop-lock-wait-expired` | `unknown` | Another stop or a launch held `.stop.lock` for the whole wait (sixteen seconds); this request acted on nothing. | | `foreign-live-process` | `unknown` | A receipt from another owner token still has live members. | +| `foreign-unknown-process` | `unknown` | A receipt from another owner token has live members that no witness ties to it. | | `stop-incomplete` | `unknown` | Stop could not confirm that every member is gone. | | `retired-owner-live` | `unknown` | Status for a retired owner whose members are not proven absent. | | `stop-lock-failed` | `unknown` | Taking `.stop.lock` failed for a reason other than the wait expiring. | | `receipt-read-failed` | `unknown` | The process receipts could not be read. | -| `retire-failed` | `unknown` | Stop could not durably retire the owner token; nothing was signalled. | +| `retire-failed` | `unknown` | Stop could not durably retire the owner token; nothing was signalled. The retirement marker may already exist, so a later `run` with that token can be refused. | | `io-error` | `unknown` | Any other local failure. | A successful reply has no `reason`. The sixteen-second stop-lock wait exceeds the timers of a stop that recovers after late guard loss (about fourteen seconds), so a waiter normally gets its own outcome; admission and census work run outside those timers, which is why an expired wait still returns its own receipt. + +The reply's `processes`, `liveProcessNames` and `unknownProcessNames` are a snapshot +read after the stop lock is released. Another stop or launch can run in between, so +they describe the root at reply time; `outcome`, `reason` and the exit status describe +this request. + +No frame (exit nonzero with no receipt, or a partial frame) means the request may or +may not have acted. That happens only before the request frame is read or when the +reply cannot be written: the startup descriptor cleanup fails (on macOS it needs one +spare descriptor to list `/dev/fd`, so a caller at its descriptor limit is refused), a +wrong argument count, a closed stdout, a reader that does not drain within ten +seconds, a receipt over 1 MiB, or the process being killed. Re-run `status` to learn +the state. `run` must be spawned as a fresh child without making it a process-group leader: the supervisor calls `setsid` itself. A caller using `detached: true` or `setpgid(0, 0)` is refused before authored code starts. @@ -111,10 +128,22 @@ there is no total ordering guarantee between stdout and stderr. Each record is written with one `write` on an `O_APPEND` descriptor and no lock. On a local file system the kernel places each append atomically at the end of the file, so whole records from different guards never interleave; records from different guards -appear in the order their appends happened. A short write (a full disk, a file-size -limit, or a signal after partial progress) is completed by a second write and can split -one record. No guard waits for another guard to log, so a stopped or hung -guard cannot stall a sibling guard's output, exit detection or stop handling. +appear in the order their appends happened. No guard waits for another guard to log, +so a stopped or hung guard cannot stall a sibling guard's output, exit detection or +stop handling. + +Readers: a final line without a line end is incomplete (a record can be read while it +is still being appended) and must be held back until it ends. + +When a log write fails, a record can be split. If the first write is short and the +second fails (a full disk), the head of the record stays in the file without a line +end; the next record from any guard is appended directly after it, so that line has a +damaged head followed by a whole record. A failed log write is an error for the guard: +it sends TERM to its command, waits the five-second grace, sends KILL and exits, so on +a full disk each guard ends its command at its next log record. With a file-size limit +(`RLIMIT_FSIZE`) the second write raises SIGXFSZ, which the guard does not handle, so +the default action ends the guard itself and its command continues without a guard; +`stop` then recovers it as after any lost guard. ## Locks and waits @@ -125,9 +154,19 @@ The bounds below limit each wait, not the total duration of a request: | Lock | Held by | Held across a blocking call? | Who waits, and the bound | |---|---|---|---| | `.instance.lock` | The project supervisor (`run` or `serve`), for its lifetime | Yes, its whole run | Another `run` or `serve` for the root: up to 10 s, then refused. Guards, `add`, `stop` and `status` never take it. | -| `.stop.lock` | `run` or `serve` from admission until its ready receipt; `add` while it admits one command; each new guard (inherited on fd 2) until it publishes and detaches; `stop` for its duration; the project supervisor while it stops on a signal | Yes. A guard frozen before it publishes keeps the lease until it is continued or killed | `stop` waits up to 16 s, then answers `unknown` with reason `stop-lock-wait-expired` and acts on nothing. `status` waits up to 16 s only when the lock file exists (it never creates it), with the same framed answer on expiry. `add` waits up to 16 s; its forwarder reads the reply for up to 26 s. A launch (`run`) waits up to 16 s, then exits nonzero without a ready receipt and starts nothing. `run` waits up to 10 s for each guard's receipt, then rolls back. | +| `.stop.lock` | `run` or `serve` from admission until its ready receipt; `add` while it admits one command; each new guard (inherited on fd 2) until it publishes and detaches, or through its cleanup if a step before detaching fails (five seconds plus the KILL confirmation); `stop` for its duration; the project supervisor while it stops on a signal | Yes. A guard frozen before it publishes keeps the lease until it is continued or killed | `stop` waits up to 16 s, then answers `unknown` with reason `stop-lock-wait-expired` and acts on nothing. `status` waits up to 16 s only when the lock file exists (it never creates it), with the same framed answer on expiry. `add` waits up to 16 s; its forwarder reads the reply for up to 26 s. A launch (`run` or `serve`) waits up to 16 s, then exits nonzero without a ready receipt and starts nothing. `run` writes each guard's plan and reads its receipt with a 10 s deadline each, then rolls back (up to seven seconds more). The project supervisor's signal-triggered stop also waits up to 16 s, then tries again on its next loop pass. | | `.lock` | The guard of that command, for its lifetime | Yes | Only a new guard for the same command name: up to 10 s, then refused. Never a sibling, `stop` or `status`. | -| Receipts, retirement marker, `lifecycle.log` | No lock | No | Receipts are replaced atomically; a reader that races a replacement reopens the name. Log records are single appends (above). | +| Receipts, retirement marker, `lifecycle.log` | No lock | No | Receipts are replaced atomically; a reader that races a replacement reads the complete version it opened. Log records are single appends (above). | + +A guard frozen before it detaches keeps every `stop`, `run`, `serve`, `add` and (when +the lock exists) `status` of the whole root failing, each within its bound, for as long as it stays frozen, healthy siblings +included. If the project supervisor is alive it rolls back and stops the published +siblings itself; if it was killed, they keep running and cannot be stopped through +`stop`. A guard frozen before its first receipt save has no receipt, so `status` does +not show it. Recovery: find the frozen guard (a stopped `hack-lifecycle-supervisor +guard` process whose session is its own PID, a child of the project supervisor or +orphaned) and continue it (`SIGCONT`) or kill it; the lease is released when it exits +or detaches. `status` takes only the bounded stop-lock wait above. `stop` handles each target separately: a frozen live guard costs its own target the five-second grace plus the two-second confirmation bound (and the @@ -180,13 +219,15 @@ leader identity and each remaining member's UID, birth and session. Recovery delivers TERM individually, grants a fresh five-second grace, then KILL to the captured incarnations and confirms absence. The capture remains authoritative if TERM ends the witness; every signal still rechecks that member's PID, birth, UID and session (pidfd on Linux). -Readers that open a receipt while a guard atomically replaces it reopen the name, so a +Readers that open a receipt while a guard atomically replaces it read the complete version +they opened (the read takes effect just before the replacement), so a concurrent update never reads as a refusal. On Linux a thread-group leader whose main thread exited while other threads run shows -`Z` in `/proc//stat`; it counts as live until no thread in `/proc//task` is. A pidfd open that +`Z` in `/proc//stat`; it counts as live while the kernel's thread count on that same +stat line is above one. A pidfd open that reports the process missing (`ESRCH`, or `EINVAL` on Linux 6.8 when the PID survives only as a group or session ID) skips that member without a signal; liveness is still decided by identity afterwards. -Unobserved descendants never inherit this authority and retain an unknown outcome. Unverified identity never authorizes signals or a successful +Unobserved descendants never inherit this authority; stop waits for them until its deadline and reports unknown if they remain. Unverified identity never authorizes signals or a successful stop. Launch and stop share a bounded lock; a startup descriptor lease keeps admission locked through guard publication even if the project owner dies. The guard temporarily inherits this stop-lock lease on stderr (fd 2), then closes @@ -407,8 +448,8 @@ the default suite and builds the release executable without this feature. ## Manual recovery after an unknown stop Stop retires the owner token before validating all receipts. One unknown receipt -can therefore leave otherwise healthy guards running and block every new start -in that private root. Repeating up with a new token does not bypass the uncertainty. +is left running (it is never signalled) and blocks every new start in that private +root, although the owner's other, verifiable targets are still stopped. Repeating up with a new token does not bypass the uncertainty. This also applies when a guard dies after spawn but before publishing running: the receipt has no command birth evidence, so an unanchored session is unknown. diff --git a/packages/lifecycle-supervisor/src/os.rs b/packages/lifecycle-supervisor/src/os.rs index 4fc71b969..a29a5d869 100644 --- a/packages/lifecycle-supervisor/src/os.rs +++ b/packages/lifecycle-supervisor/src/os.rs @@ -63,68 +63,76 @@ pub struct Identity { pub zombie: bool, } +/// One `/proc//stat` read: whether the thread group is dead, group, +/// session and start time. +/// +/// A thread-group leader that called `pthread_exit` shows `Z` while its other +/// threads keep running; such a process is alive (it owns its PID and receives +/// process-directed signals). The leader's own stat line carries the kernel's +/// count of the group's threads (field 20) from the same read: a dead leader +/// with more than one counted thread is a live process. An exiting thread stays +/// counted until it is released, which only errs toward live. No listing of +/// `/proc//task` is used, because a listing cannot see a thread created +/// after it and can miss a live replacement during thread turnover. #[cfg(target_os = "linux")] -pub fn inspect(pid: i32) -> Result> { - let path = format!("/proc/{pid}/stat"); - let bytes = match fs::read(&path) { +fn linux_stat(pid: i32) -> Result> { + let bytes = match fs::read(format!("/proc/{pid}/stat")) { Ok(value) => value, Err(error) if absent_error(&error) => return Ok(None), Err(error) => return Err(error), }; let tail = proc_stat_tail(&bytes)?; let fields: Vec<_> = tail.split_whitespace().collect(); - let group = fields + let group: i32 = fields .get(2) .and_then(|v| v.parse().ok()) .ok_or_else(refused)?; - let start = fields + let session: i32 = fields + .get(3) + .and_then(|v| v.parse().ok()) + .ok_or_else(refused)?; + let start: u64 = fields .get(19) .and_then(|v| v.parse().ok()) .ok_or_else(refused)?; - let session = fields - .get(3) + let threads: u64 = fields + .get(17) .and_then(|v| v.parse().ok()) .ok_or_else(refused)?; - let uid = match linux_uid(pid)? { - Some(uid) => uid, - None => return Ok(None), - }; - Ok(Some(Identity { - pid, - group, - session, - start, - uid, - zombie: matches!(fields.first(), Some(&"Z" | &"X")) && !linux_live_thread(pid)?, - })) + let group_dead = matches!(fields.first(), Some(&"Z" | &"X")) && threads <= 1; + Ok(Some((group_dead, group, session, start))) } -/// A thread-group leader that called `pthread_exit` shows `Z` in -/// `/proc//stat` while its other threads keep running. Such a process is -/// alive: it still owns its PID and receives process-directed signals. Only a -/// group with no live thread left is a zombie. +/// The identity is assembled from two reads (`stat`, then `status`). The +/// leader's start time is read again afterwards: if it changed, the +/// PID was reused during the reads, so the reads are repeated rather than +/// combining two processes. Three unstable attempts refuse. #[cfg(target_os = "linux")] -fn linux_live_thread(pid: i32) -> Result { - let tasks = match fs::read_dir(format!("/proc/{pid}/task")) { - Ok(tasks) => tasks, - Err(error) if absent_error(&error) => return Ok(false), - Err(error) => return Err(error), - }; - for task in tasks { - let task = task?; - let stat = match fs::read(task.path().join("stat")) { - Ok(stat) => stat, - Err(error) if absent_error(&error) => continue, - Err(error) => return Err(error), +pub fn inspect(pid: i32) -> Result> { + for _ in 0..3 { + let Some((zombie, group, session, start)) = linux_stat(pid)? else { + return Ok(None); }; - if !matches!( - proc_stat_tail(&stat)?.split_whitespace().next(), - Some("Z" | "X") - ) { - return Ok(true); + let uid = match linux_uid(pid)? { + Some(uid) => uid, + None => return Ok(None), + }; + match linux_stat(pid)? { + None => return Ok(None), + Some((_, _, _, again)) if again == start => { + return Ok(Some(Identity { + pid, + group, + session, + start, + uid, + zombie, + })); + } + Some(_) => continue, } } - Ok(false) + Err(refused()) } #[cfg(target_os = "linux")] @@ -815,21 +823,34 @@ impl Recovery { Ok(()) } - pub fn live(&self) -> Result { + pub fn live(&self) -> Result { for member in &self.members { if inspect(member.pid)?.is_some_and(|p| !p.zombie && incarnation_matches(member, &p)) { - return Ok(true); + return Ok(RecoveryLiveness::Captured); } } // Unobserved descendants or an unrelated reused session cannot be // signalled by this snapshot, and must not be claimed as stopped. if live_group(self.session, Some(self.session))? { - return Err(refused()); + return Ok(RecoveryLiveness::Unobserved); } - Ok(false) + Ok(RecoveryLiveness::Gone) } } +/// What a recovery snapshot can still see of its session. +#[derive(Debug, PartialEq, Eq)] +pub enum RecoveryLiveness { + /// A captured incarnation is alive; it may still be signalled. + Captured, + /// No captured incarnation is alive, but the session still has members this + /// snapshot never observed. They may exit by themselves; they are never + /// signalled, and stop reports unknown if they outlive its deadline. + Unobserved, + /// The session has no live member. + Gone, +} + /// A successful census must contain our own readable identity. Known restricted /// proc mounts are unsupported: hidden same-user members defeat absence proofs. pub fn census_preflight() -> Result<()> { diff --git a/packages/lifecycle-supervisor/src/storage.rs b/packages/lifecycle-supervisor/src/storage.rs index f49364c4d..5ca9f3e6f 100644 --- a/packages/lifecycle-supervisor/src/storage.rs +++ b/packages/lifecycle-supervisor/src/storage.rs @@ -95,8 +95,13 @@ impl Root { let name = std::ffi::CString::new(name).map_err(|_| refused())?; // A receipt is replaced atomically (temporary file, then rename). An open // that races that rename can return the replaced inode, which is already - // unlinked (nlink 0). Reopen the name to read its current inode; any - // other link count is still refused. The bound only stops a livelock. + // unlinked (nlink 0). For a read-only open that inode is a complete + // receipt, the version current when the open happened, so it is read as + // is: the read takes effect just before the rename, and an unlinked inode + // has no other name a hard link could alias. Opens that may write retry + // the name for its current inode; the bound only stops a livelock. Any + // other link count is refused. + let read_only = flags & libc::O_ACCMODE == libc::O_RDONLY; for _ in 0..64 { // SAFETY: directory descriptor is live; name is NUL-terminated; // returned descriptor is newly owned and immediately wrapped in RAII. @@ -114,13 +119,13 @@ impl Root { // SAFETY: fd was just returned by openat and has no other owner. let file = File::from(unsafe { OwnedFd::from_raw_fd(fd) }); let metadata = file.metadata()?; - if metadata.nlink() == 0 { + if metadata.nlink() == 0 && !read_only { continue; } // SAFETY: geteuid has no preconditions. if !metadata.is_file() || metadata.uid() != unsafe { libc::geteuid() } - || metadata.nlink() != 1 + || metadata.nlink() > 1 || metadata.mode() & 0o777 != 0o600 { return Err(refused()); diff --git a/packages/lifecycle-supervisor/src/supervisor.rs b/packages/lifecycle-supervisor/src/supervisor.rs index 56564fa8a..90e418a09 100644 --- a/packages/lifecycle-supervisor/src/supervisor.rs +++ b/packages/lifecycle-supervisor/src/supervisor.rs @@ -100,7 +100,7 @@ fn control(action: &str) -> Result<()> { let mut result = control_effects(action, &request, &root, &mut reason); // Receipts are atomically replaced files, so this read is safe without the // lock, also after a lock wait expired. A failed read never reports - // success: it keeps an earlier error, or becomes `io-error` itself. + // success: it keeps an earlier error, or becomes `receipt-read-failed`. let processes = match root.states() { Ok(states) => states .into_iter() @@ -164,12 +164,22 @@ fn control_effects( *reason = "io-error"; if state.owner_token == request.owner_token { states.push(state); - } else if os::recorded_live(&state, false)? { - // Old terminal receipts survive renames and new generations. - // Guard-written terminal phases prove completed retirement; - // non-terminal receipts require native identity evidence. - *reason = "foreign-live-process"; - return Err(refused()); + continue; + } + // Old terminal receipts survive renames and new generations. + // Guard-written terminal phases prove completed retirement; + // non-terminal receipts require native identity evidence. + match os::recorded_live(&state, false) { + Ok(false) => (), + Ok(true) => { + *reason = "foreign-live-process"; + return Err(refused()); + } + Err(error) => { + // Live members that no witness ties to that receipt. + *reason = "foreign-unknown-process"; + return Err(error); + } } } *reason = "io-error"; @@ -671,51 +681,65 @@ fn stop(states: &[State]) -> Result<()> { } } let recovery = self.recovery.as_ref().ok_or_else(refused)?; - if !recovery.live()? { + let liveness = recovery.live()?; + if liveness == os::RecoveryLiveness::Gone { return Ok(false); } let elapsed = self.orphan_term.ok_or_else(refused)?.elapsed(); if elapsed >= GRACE + CONFIRM_BOUND { return Err(refused()); } - if elapsed >= GRACE { + // Only captured incarnations are signalled. Unobserved members get + // the rest of this target's deadline to exit on their own. + if elapsed >= GRACE && liveness == os::RecoveryLiveness::Captured { recovery.signal(libc::SIGKILL)?; } Ok(true) } } let mut active = Vec::new(); + let mut failure = None; // Validate admission before effects. Capture while a live guard can still // prove membership, so loss during the later liveness check retains authority. + // A receipt that cannot be verified is never signalled; it records the first + // failure, and the other admitted targets still get their stop. for state in states { - let guard_live = match os::inspect(state.guard_pid)? { - Some(identity) if !identity.zombie => { - if let Err(error) = os::same(state, &identity) { - if !os::recorded_live(state, false)? { - continue; + let admitted = (|| -> Result> { + let guard_live = match os::inspect(state.guard_pid)? { + Some(identity) if !identity.zombie => { + if let Err(error) = os::same(state, &identity) { + if !os::recorded_live(state, false)? { + return Ok(None); + } + return Err(error); } - return Err(error); + true } - true + _ => false, + }; + let (recovery, admission) = stop_capture(guard_live, || os::Recovery::capture(state))?; + match os::recorded_live(state, false) { + Ok(false) => return Ok(None), + Ok(true) => (), + Err(_) if guard_lost(state)? && (recovery.is_some() || admission.is_some()) => (), + Err(error) => return Err(error), } - _ => false, - }; - let (recovery, admission) = stop_capture(guard_live, || os::Recovery::capture(state))?; - match os::recorded_live(state, false) { - Ok(false) => continue, - Ok(true) => (), - Err(_) if guard_lost(state)? && (recovery.is_some() || admission.is_some()) => (), - Err(error) => return Err(error), - } - active.push(Target { - state: state.clone(), - recovery, - admission, - orphan_term: None, - finished: false, - }); + Ok(Some(Target { + state: state.clone(), + recovery, + admission, + orphan_term: None, + finished: false, + })) + })(); + match admitted { + Ok(Some(target)) => active.push(target), + Ok(None) => (), + Err(error) => { + failure.get_or_insert(error); + } + } } - let mut failure = None; for target in &mut active { let signalled = if let Some(recovery) = &target.recovery { target.orphan_term = Some(Instant::now()); diff --git a/packages/lifecycle-supervisor/tests/fake_caller.rs b/packages/lifecycle-supervisor/tests/fake_caller.rs index 7c61b8311..c9dfd426f 100644 --- a/packages/lifecycle-supervisor/tests/fake_caller.rs +++ b/packages/lifecycle-supervisor/tests/fake_caller.rs @@ -8,7 +8,7 @@ use std::{ os::{fd::AsRawFd, unix::fs::MetadataExt}, path::{Path, PathBuf}, process::{Child, Command, Stdio}, - sync::atomic::{AtomicU64, Ordering}, + sync::atomic::{AtomicBool, AtomicU64, Ordering}, thread, time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }; @@ -55,6 +55,7 @@ impl Fixture { json!({"version":1,"root":self.root,"ownerToken":token}) } + #[track_caller] fn request(&self, action: &str, request: &Value) -> (Child, Value) { let mut child = command(&self.root, action).spawn().unwrap(); send(child.stdin.as_mut().unwrap(), request); @@ -124,6 +125,7 @@ impl Fixture { lock } + #[track_caller] fn state(&self, name: &str) -> Value { eventually(|| { fs::read(self.root.join(format!("{name}.json"))) @@ -408,6 +410,18 @@ impl Drop for ForeignChild { } } +/// A refused `stop` or `status`: exits nonzero and still answers with one +/// framed receipt carrying the expected outcome and reason. +#[track_caller] +fn refused_framed(fixture: &Fixture, action: &str, request: Value, reason: &str) -> Value { + let (mut child, receipt) = fixture.request(action, &request); + assert!(!wait(&mut child, TIMEOUT).success(), "{receipt}"); + assert_eq!(receipt["outcome"], "unknown", "{receipt}"); + assert_eq!(receipt["reason"], reason, "{receipt}"); + receipt +} + +#[track_caller] fn refused(fixture: &Fixture, action: &str, request: Value) { let mut child = command(&fixture.root, action).spawn().unwrap(); send(child.stdin.as_mut().unwrap(), &request); @@ -487,7 +501,12 @@ fn foreign_owner_cannot_stop() { let f = Fixture::new(); let (mut caller, _) = f.start(vec![f.process("worker", "hold")]); f.await_log("fixture-ready"); - refused(&f, "stop", f.control("different-owner")); + refused_framed( + &f, + "stop", + f.control("different-owner"), + "foreign-live-process", + ); let state = f.state("worker"); // SAFETY: signal zero observes this fixture's process without delivering a signal. assert_eq!(unsafe { libc::kill(pid(&state, "commandPid"), 0) }, 0); @@ -530,21 +549,44 @@ fn output_bursts_and_long_lines_include_partial_tail() { } } -/// Without a log lock, each record is one O_APPEND write. Two guards bursting -/// at once, including 64 KiB records far above PIPE_BUF, must still leave only -/// whole records: every line parses, belongs to one guard and is not mixed. +/// Without a log lock, each record is one O_APPEND write. Two guards burst +/// with a barrier in the middle (each waits until both halves are in the log), +/// so their appends overlap in both directions by construction, including +/// 64 KiB records far above PIPE_BUF. Every line must be one whole record of +/// one guard: a split anywhere, also inside the timestamp, fails a field check. #[test] fn concurrent_guards_append_whole_records_without_a_lock() { let f = Fixture::new(); let (mut caller, _) = f.start(vec![ - f.process("alpha", "burst"), - f.process("beta", "burst"), + f.process("alpha", "burst-pair-alpha"), + f.process("beta", "burst-pair-beta"), ]); assert!(wait(&mut caller, TIMEOUT).success()); let text = f.log(); + let lines: Vec<_> = text.lines().collect(); + for line in &lines { + let fields: Vec<_> = line.splitn(4, '\t').collect(); + assert_eq!(fields.len(), 4, "split record: {line:.80}"); + assert_eq!(fields[0].len(), 20, "split timestamp: {line:.80}"); + assert!(fields[0].ends_with('Z'), "{line:.80}"); + assert!(matches!(fields[1], "alpha" | "beta"), "{line:.80}"); + assert!(matches!(fields[2], "stdout" | "stderr"), "{line:.80}"); + } + for (first, later) in [("alpha", "beta"), ("beta", "alpha")] { + let start = lines + .iter() + .position(|line| line.split('\t').nth(1) == Some(first)) + .expect("a guard logged nothing"); + assert!( + lines[start..] + .iter() + .any(|line| line.split('\t').nth(1) == Some(later)), + "no {later} record after the first {first} record" + ); + } for name in ["alpha", "beta"] { - let own: Vec<_> = text - .lines() + let own: Vec<_> = lines + .iter() .filter(|line| line.split('\t').nth(1) == Some(name)) .collect(); assert_eq!( @@ -562,12 +604,6 @@ fn concurrent_guards_append_whole_records_without_a_lock() { .sum(); assert_eq!(xs, 300_000); } - for line in text.lines() { - let fields: Vec<_> = line.splitn(4, '\t').collect(); - assert_eq!(fields.len(), 4, "split record: {line:.80}"); - assert!(matches!(fields[1], "alpha" | "beta"), "{line:.80}"); - assert!(fields[0].ends_with('Z')); - } } #[test] @@ -1176,11 +1212,24 @@ fn guard_lost_during_stop(delay: Duration, blocked: bool) { "TERM-ignoring child exited before guard loss" ); let recovery_started = Instant::now(); + // Freeze stop too, so it cannot observe the guard as an unreaped zombie: + // a zombie guard is still a witness, and a fresh capture would then succeed + // without the admission snapshot this test is about. + let stop_pid = stop.id() as i32; + // SAFETY: the stop request is this test's own child. + assert_eq!(unsafe { libc::kill(stop_pid, libc::SIGSTOP) }, 0); + // Resume the stop request even if any assertion below fails first. + let resume = ResumeOnDrop(stop_pid); + eventually(|| process_is_stopped(stop_pid).then_some(())); // SAFETY: the frozen guard belongs to this fixture. assert_eq!(unsafe { libc::kill(guard, libc::SIGKILL) }, 0); if !blocked { assert!(wait(&mut caller, TIMEOUT).success()); } + // Resume stop only after the guard incarnation has no process record left, + // zombie included, so no guard can witness a fresh capture. + eventually(|| (process_birth(guard) != Some(guard_birth)).then_some(())); + drop(resume); let receipt = receive(stop.stdout.as_mut().unwrap()); let stopped = wait(&mut stop, TIMEOUT).success(); assert_eq!(stopped, !blocked, "{receipt}"); @@ -1247,6 +1296,118 @@ fn orphan_recovery_keeps_observed_child_after_term_ends_witness() { ); } +/// Recovery captures the session before its TERM. A descendant forked after +/// that capture is never signalled; stop waits until its own deadline for it. +fn late_descendant_case(forever: bool) { + let f = Fixture::new(); + let mode = if forever { + "fork-on-term-forever" + } else { + "fork-on-term" + }; + let (mut caller, _) = f.start(vec![f.process("worker", mode)]); + f.await_log("fixture-ready"); + let state = f.state("worker"); + // SAFETY: the captured guard belongs to this fixture; its command survives. + assert_eq!( + unsafe { libc::kill(pid(&state, "guardPid"), libc::SIGKILL) }, + 0 + ); + assert!(wait(&mut caller, TIMEOUT).success()); + let start = Instant::now(); + let (mut stop, receipt) = f.request("stop", &f.control(TOKEN)); + let stopped = wait(&mut stop, TIMEOUT).success(); + let elapsed = start.elapsed(); + let late = std::fs::read_to_string(f.root.join("late-descendant")) + .ok() + .and_then(|value| value.parse::().ok()) + .expect("command did not fork after recovery TERM"); + let late_identity = identity(late); + if let Some(late) = &late_identity { + f.owned.borrow_mut().push(late.clone()); + } + assert!( + identity(pid(&state, "commandPid")).is_none() + || process_is_zombie(pid(&state, "commandPid")), + "captured command survived recovery KILL" + ); + if forever { + assert!(!stopped, "{receipt}"); + assert_eq!(receipt["outcome"], "unknown"); + assert!( + elapsed >= Duration::from_secs(7), + "stop gave up before its deadline: {elapsed:?}" + ); + let late = late_identity.expect("unobserved descendant was signalled"); + assert!( + !process_is_zombie(late.pid), + "unobserved descendant was signalled" + ); + } else { + assert!(stopped, "{receipt}"); + assert_eq!(receipt["outcome"], "stopped"); + assert!(elapsed >= Duration::from_secs(5)); + assert!(elapsed < TIMEOUT); + } +} + +#[test] +fn orphan_recovery_waits_for_unobserved_descendant_to_exit() { + late_descendant_case(false); +} + +#[test] +fn orphan_recovery_never_signals_unobserved_descendant() { + late_descendant_case(true); +} + +/// Sends SIGCONT to a stopped child of this test when dropped, so a failed +/// assertion cannot leave the stop request suspended. +struct ResumeOnDrop(i32); + +impl Drop for ResumeOnDrop { + fn drop(&mut self) { + // SAFETY: the PID is this test's own unreaped child. + unsafe { + libc::kill(self.0, libc::SIGCONT); + } + } +} + +/// Birth of any process record at `pid`, zombies included, matching the +/// production inspection (Darwin `proc_pidinfo` argument 1, no `getsid`). +fn process_birth(pid: i32) -> Option { + #[cfg(target_os = "linux")] + { + let text = fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; + text.rsplit_once(") ")? + .1 + .split_whitespace() + .nth(19)? + .parse() + .ok() + } + #[cfg(target_os = "macos")] + { + let mut info = std::mem::MaybeUninit::::zeroed(); + // SAFETY: native output buffer is writable and exactly sized. + let size = unsafe { + libc::proc_pidinfo( + pid, + libc::PROC_PIDTBSDINFO, + 1, + info.as_mut_ptr().cast(), + std::mem::size_of::() as i32, + ) + }; + if size as usize != std::mem::size_of::() { + return None; + } + let info = unsafe { info.assume_init() }; + Some(info.pbi_start_tvsec * 1_000_000 + info.pbi_start_tvusec) + } +} + #[test] fn stop_waiting_past_lock_bound_reports_unknown_without_acting() { let f = Fixture::new(); @@ -1299,9 +1460,16 @@ fn invalid_control_requests_still_get_a_framed_refusal() { #[cfg(target_os = "linux")] #[test] fn main_thread_exit_with_live_worker_is_still_stopped() { - for lose_guard in [false, true] { + // `pthread-churn` keeps replacing its only worker thread, so a scan of the + // task list can miss the live replacement; the kernel's thread count cannot. + for (mode, lose_guard) in [ + ("pthread-main-exit", false), + ("pthread-main-exit", true), + ("pthread-churn", false), + ("pthread-churn", true), + ] { let f = Fixture::new(); - let (mut caller, _) = f.start(vec![f.process("worker", "pthread-main-exit")]); + let (mut caller, _) = f.start(vec![f.process("worker", mode)]); f.await_log("fixture-ready"); let state = f.state("worker"); let child = eventually(|| { @@ -1377,6 +1545,88 @@ fn held_log_lock_never_stalls_a_guard() { drop(log); } +/// One receipt that cannot be verified (guard gone, recorded command birth +/// wrong, so no witness proves the session) must not keep stop from stopping +/// the healthy commands of the same owner. The unverifiable command is never +/// signalled, and the request answers `unknown` with `stop-incomplete`. +#[test] +fn unverifiable_receipt_does_not_block_healthy_targets() { + let f = Fixture::new(); + let (mut caller, _) = f.start(vec![ + f.process("healthy", "hold"), + f.process("unverifiable", "hold"), + ]); + eventually(|| (f.log().matches("fixture-ready").count() == 2).then_some(())); + let healthy = identity(pid(&f.state("healthy"), "commandPid")).expect("healthy command"); + let mut bad = f.state("unverifiable"); + let bad_command = identity(pid(&bad, "commandPid")).expect("unverifiable command"); + let bad_guard = pid(&bad, "guardPid"); + // SAFETY: the guard belongs to this fixture; its command survives it. + assert_eq!(unsafe { libc::kill(bad_guard, libc::SIGKILL) }, 0); + eventually(|| identity(bad_guard).is_none().then_some(())); + bad["commandStartTime"] = json!(bad["commandStartTime"].as_u64().unwrap() + 1); + fs::write( + f.root.join("unverifiable.json"), + serde_json::to_vec(&bad).unwrap(), + ) + .unwrap(); + let (mut stop, receipt) = f.request("stop", &f.control(TOKEN)); + assert!(!wait(&mut stop, TIMEOUT).success(), "{receipt}"); + assert_eq!(receipt["outcome"], "unknown"); + assert_eq!(receipt["reason"], "stop-incomplete"); + eventually(|| { + (identity(healthy.pid).as_ref() != Some(&healthy) || process_is_zombie(healthy.pid)) + .then_some(()) + }); + assert!( + identity(bad_command.pid).as_ref() == Some(&bad_command) + && !process_is_zombie(bad_command.pid), + "unverifiable command was signalled" + ); + // SAFETY: this fixture's own command, identity checked just above. + unsafe { + libc::kill(bad_command.pid, libc::SIGKILL); + } + let _ = wait(&mut caller, TIMEOUT); +} + +/// Another owner's receipt whose live members no witness ties to it (guard +/// gone, recorded command birth wrong) refuses with `foreign-unknown-process` +/// and signals nothing. +#[test] +fn unverifiable_foreign_receipt_refuses_with_its_own_reason() { + let f = Fixture::new(); + let (mut caller, _) = f.start(vec![f.process("worker", "hold")]); + f.await_log("fixture-ready"); + let mut state = f.state("worker"); + let command = identity(pid(&state, "commandPid")).expect("command identity"); + let guard = pid(&state, "guardPid"); + // SAFETY: the guard belongs to this fixture; its command survives it. + assert_eq!(unsafe { libc::kill(guard, libc::SIGKILL) }, 0); + eventually(|| identity(guard).is_none().then_some(())); + state["commandStartTime"] = json!(state["commandStartTime"].as_u64().unwrap() + 1); + fs::write( + f.root.join("worker.json"), + serde_json::to_vec(&state).unwrap(), + ) + .unwrap(); + refused_framed( + &f, + "stop", + f.control("different-owner"), + "foreign-unknown-process", + ); + assert!( + identity(command.pid).as_ref() == Some(&command), + "command was signalled" + ); + // SAFETY: this fixture's own command, identity checked just above. + unsafe { + libc::kill(command.pid, libc::SIGKILL); + } + let _ = wait(&mut caller, TIMEOUT); +} + fn process_is_zombie(pid: i32) -> bool { process_state(pid) == Some(b'Z') } @@ -1821,8 +2071,11 @@ fn unresponsive_guard_stop_remains_unknown() { assert_eq!(unsafe { libc::kill(guard, libc::SIGSTOP) }, 0); let (mut stop, receipt) = f.request("stop", &f.control(TOKEN)); assert_eq!(receipt["outcome"], "unknown"); + assert_eq!(receipt["reason"], "stop-incomplete"); assert!(!wait(&mut stop, TIMEOUT).success()); assert_eq!(unsafe { libc::kill(pid(&state, "commandPid"), 0) }, 0); + // The owner is retired but its member is not proven absent. + refused_framed(&f, "status", f.control(TOKEN), "retired-owner-live"); refused(&f, "run", f.launch(vec![f.process("worker", "hold")])); assert_eq!(unsafe { libc::kill(guard, libc::SIGCONT) }, 0); f.stop(); @@ -1997,6 +2250,37 @@ fn fixture_process() { "command inherited caller environment" ); match mode.as_str() { + "burst-pair-alpha" | "burst-pair-beta" => { + // Two guards' appends overlap by construction: each writes half its + // records, then waits until both halves are in the shared log + // before writing the rest. + let (own, other) = if mode == "burst-pair-alpha" { + ("alpha", "beta") + } else { + ("beta", "alpha") + }; + let mut out = std::io::stdout().lock(); + for i in 0..1000 { + writeln!(out, "burst-record-{i}").unwrap(); + } + out.flush().unwrap(); + let half = |name: &str| format!("\t{name}\tstdout\tburst-record-999\n"); + loop { + let log = fs::read_to_string("lifecycle.log").unwrap_or_default(); + if log.contains(&half(own)) && log.contains(&half(other)) { + break; + } + thread::sleep(Duration::from_millis(1)); + } + for i in 1000..2000 { + writeln!(out, "burst-record-{i}").unwrap(); + } + out.write_all(&vec![b'X'; 300_000]).unwrap(); + out.write_all(b"\npartial-tail").unwrap(); + out.flush().unwrap(); + eprintln!("fixture-error"); + std::process::exit(0); + } "burst" => { let mut out = std::io::stdout().lock(); for i in 0..2000 { @@ -2199,8 +2483,62 @@ fn fixture_process() { } println!("wrapper-child {child}"); } + "fork-on-term" | "fork-on-term-forever" => { + // Fork one TERM-ignoring descendant after each TERM, so recovery + // meets a session member forked after its capture. Its PID goes to + // a file: stdout belongs to the guard, which the test kills first. + static TERMINATED: AtomicBool = AtomicBool::new(false); + extern "C" fn on_term(_: libc::c_int) { + TERMINATED.store(true, Ordering::SeqCst); + } + let forever = mode == "fork-on-term-forever"; + // SAFETY: the handler only stores an atomic flag. + unsafe { + libc::signal( + libc::SIGTERM, + on_term as extern "C" fn(libc::c_int) as libc::sighandler_t, + ); + } + println!("fixture-ready"); + std::io::stdout().flush().unwrap(); + loop { + if TERMINATED.swap(false, Ordering::SeqCst) { + // Saved before the fork: a child first scheduled after this + // process died would otherwise read its reaper's PID. + // SAFETY: getpid has no preconditions. + let parent = unsafe { libc::getpid() }; + // SAFETY: the fork child uses only async-signal-safe libc + // calls and never returns into Rust state. + let child = unsafe { libc::fork() }; + assert!(child >= 0); + if child == 0 { + unsafe { + libc::signal(libc::SIGTERM, libc::SIG_IGN); + if forever { + loop { + libc::pause(); + } + } + // Outlive the parent's recovery KILL (at five + // seconds), then exit well before the seven-second + // deadline. Tie the exit to the parent's death, not + // to a fixed time, so a slow host cannot push it + // past the deadline. + while libc::getppid() == parent { + libc::usleep(10_000); + } + libc::usleep(500_000); + libc::_exit(0); + } + } + std::fs::write("late-descendant.tmp", child.to_string()).unwrap(); + std::fs::rename("late-descendant.tmp", "late-descendant").unwrap(); + } + thread::sleep(Duration::from_millis(10)); + } + } #[cfg(target_os = "linux")] - "pthread-main-exit" => { + "pthread-main-exit" | "pthread-churn" => { extern "C" fn worker(_: *mut libc::c_void) -> *mut libc::c_void { loop { // SAFETY: pause has no preconditions. @@ -2209,6 +2547,36 @@ fn fixture_process() { } } } + /// Thread turnover: each worker starts its replacement, then ends. + /// The group always has a live thread, but no single listing of + /// its tasks is guaranteed to see one. + extern "C" fn churn(_: *mut libc::c_void) -> *mut libc::c_void { + // SAFETY: libc thread calls only; a failed create keeps this + // thread alive instead of ending the group. + unsafe { + libc::usleep(2000); + let mut next: libc::pthread_t = std::mem::zeroed(); + if libc::pthread_create( + &mut next, + std::ptr::null(), + churn, + std::ptr::null_mut(), + ) != 0 + { + loop { + libc::pause(); + } + } + libc::syscall(libc::SYS_exit, 0); + } + std::ptr::null_mut() + } + let body: extern "C" fn(*mut libc::c_void) -> *mut libc::c_void = + if mode == "pthread-churn" { + churn + } else { + worker + }; // The fork child is single-threaded, so its main thread is the // thread-group leader. It starts one worker and ends only its main // thread (as `pthread_exit` does); Linux then shows the leader as @@ -2223,7 +2591,7 @@ fn fixture_process() { if libc::pthread_create( &mut thread, std::ptr::null(), - worker, + body, std::ptr::null_mut(), ) != 0 {