diff --git a/DESIGN.md b/DESIGN.md index d2c532f..869f99b 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -198,7 +198,22 @@ Crash recovery needs the data directory to **outlive the process**: on a contain **Release note, default location moved.** Earlier builds defaulted to `/bugsee/`; the default is now the platform location (3) or the private temp fallback (4). When no directory was requested and the old location exists, is a real directory and is owned by the current user, the launch recovers its crashed sessions and delivers its queue once, best effort, including when no new directory could be created. Recovery runs on the worker before a `flush` is acknowledged and `flush` drains the legacy queue with the data directory's, so a flush covers it; the uploader stops polling it once its queue is empty. It is not claimed to be lossless: a legacy session whose process is still alive is skipped, and a legacy directory that fails the ownership check is left alone. Hosts that set `data_dir` were never using it. -**Unsafe.** The facade makes two `geteuid()` calls (unix) to name and vet the private temp directory, alongside core's single `kill(pid, 0)`. +**Unsafe.** The facade makes two `geteuid()` calls (unix) to name and vet the private temp directory. Core adds `fcntl(F_SETLK)` (unix) / `LockFileEx` (Windows) for claims, a `kill(pid, 0)` probe (plus `proc_pidinfo` on macOS, `OpenProcess`/`WaitForSingleObject` on Windows) for legacy markers only, and one `UnsafeCell` registry guarded by a pid-stamped spin lock (below). + +**Several processes on one directory (#14).** Every pending item is owned by exactly one process at a time, arbitrated by an **OS byte-range lock** on a `*.lock` file: a queued report by `queue/.lock`, a session being recovered by `sessions/.recover.lock`, and a *live session* by `sessions/.lock`, held for the session's whole life. The kernel releases the lock when the process dies however it dies, so nothing can go stale: there is no owner pid, start time, lease or takeover protocol to get wrong. Choices that follow from this: + +- **Pid namespaces.** Containers sharing a volume each see their app as pid 1; a pid (or pid plus start time) written in a file means nothing to the other. A lock does not depend on it, so a peer's live session is never mistaken for a crash and a dead one is never kept alive by pid collision. The `.alive` marker's pid is consulted only for a marker with no `.lock` beside it, i.e. one written by an older SDK build. +- **Why `fcntl`, not `flock`/OFD locks.** `fcntl` locks are per process and are not inherited across `fork`, so a prefork child never holds its parent's claim; `flock`/OFD locks follow the open file description and would. (MSRV 1.88 predates `File::lock` anyway, hence libc/FFI.) +- **Costs of per-process locks, handled in `claim.rs`.** They do not exclude threads, so a registry of held paths does (keyed by pid and spin-locked with the holder's pid stamped in, so a `fork` child that inherits it mid-update breaks the lock instead of deadlocking). POSIX drops a process's locks on a file when *any* descriptor for it closes, so a lock file is never opened twice by one process: the registry is checked before each open, and a claim closes its descriptor before leaving the registry. +- **Ordering at launch.** `Session::begin` creates the marker (`O_EXCL`), takes `.lock`, then writes the pid. Until the pid is written the marker is empty and fresh, which peers skip even if the lock reads free. A lock that cannot be held with kernel enforcement (contended past a short retry, or the filesystem rejects locks) is *removed*: an existing but unheld lock file reads as "that session died", so peers fall back to the marker's pid instead. On unix a probe asks `F_GETLK` and never takes the lock, so it cannot steal one from a launching peer. +- **Unclean shutdown.** If the worker or uploader thread is detached at shutdown it may still be writing the session, so the session's lock is kept until the process exits (the kernel then drops it) rather than released on drop. +- **Filesystems without locks.** If the filesystem rejects byte-range locks (`ENOLCK` on NFS without lockd; `EOPNOTSUPP`/`ENOSYS`/`EINVAL` on FUSE, 9p) a claim degrades to thread-only exclusion with a one-time message on stderr, and a probe reports unknown, and a lock file that vanishes under a peer's scan is likewise not taken as evidence of death (as a probe does for a leftover lock file that cannot be opened, e.g. on a read-only volume, so a crash is never hidden by it). A session that could not take a lock keeps a `.beat` heartbeat file, touched every 2 s by the worker (and during launch-time recovery): a beat younger than 3 minutes, or dated in the future (generous: NFS caches attributes for up to a minute and its mtime is the server's clock) means alive whatever pid the marker names, which is the only signal that works across pid namespaces without locks; with no beat the marker's pid is probed as before. The price is that a crash on such a filesystem is recovered only after its heartbeat has been quiet for three minutes, so a faster restart defers it to a later launch. Under the recovery claim the liveness decision is re-run, so a session that finished launching since the scan is never torn down. The queue keeps draining and sessions keep recovering, but processes sharing that directory are no longer kept from delivering a report twice. Locks that *succeed but do not exclude* (Docker Desktop bind mounts, WSL2 drvfs/9p, NFS mounted `nolock`) cannot be detected and behave the same way: run one SDK host per directory there. +- **Paths.** The in-process registry canonicalises the lock file's directory, so `/tmp` vs `/private/tmp` or relative vs absolute paths are one claim. A `fork` child clears the registry (`claim::reset_after_fork`, from the fork hook) before claiming anything. +- **Stale scans.** After winning a session claim the marker is re-checked, since a peer may have recovered and discarded it since the scan (this caused phantom duplicate reports). `queue::gc_orphans` only reaps files untouched for 60 s: a peer's in-flight `enqueue` looks like an orphan, and reaping it lost the report. Lock files of finished items and dead sessions are swept when free. +- **Compatibility.** The `.alive` marker is still a bare pid and the lock files are new names, so older builds sharing the directory keep working. They do not take part in the locking, so a mixed-version directory (a rolling upgrade) gives no exactly-once guarantee between old and new processes, and a new process treats an old build's session by its pid as before. Run one SDK version per directory for the guarantee. +- **Limits.** Advisory locks on network filesystems are best effort (see above). The only timestamps relied on are the 30 s grace for an empty marker and the 60 s orphan grace, which assume clocks within seconds. Delivery is at-least-once: a holder that dies after uploading but before removing the item (or after enqueueing a recovered report but before discarding the session) makes its successor do it again. Platforms with neither unix nor Windows locks exclude threads only. + +Tests: `bugsee-core/tests/multiprocess.rs` (real child processes: racing launchers, a SIGKILLed holder, a live peer whose marker names a colliding pid) and unit tests in `claim.rs`, `process.rs`, `queue.rs`. Tests: `bugsee-core/tests/storage_failures.rs`, `bugsee/tests/storage_fallback.rs`, a memory-mode case in `bugsee/tests/fork_safety.rs`, `options.rs` unit tests. A real ENOSPC (size-capped tmpfs) is not exercised, since mounting one needs root; failures are injected by blocking path components and removing the directory mid-run. diff --git a/crates/bugsee-core/src/claim.rs b/crates/bugsee-core/src/claim.rs index 341f1e0..f3d0956 100644 --- a/crates/bugsee-core/src/claim.rs +++ b/crates/bugsee-core/src/claim.rs @@ -4,115 +4,547 @@ // // Copyright © 2026 Bugsee. All rights reserved. // -//! Cross-process exclusive claims on on-disk work items. +//! Cross-process exclusive claims on on-disk work items, and session liveness. //! //! Several processes can share one data directory — a prefork server's master -//! and its workers above all — and the work items in it (a crashed session to -//! recover, a queued report to upload) must each be handled by exactly one of -//! them, or the backend receives duplicates. A claim is a small file created -//! with `O_EXCL` that records its owner's pid. +//! and its workers, containers sharing a volume — and the work items in it (a +//! crashed session to recover, a queued report to upload) must each be handled +//! by exactly one of them, or the backend receives duplicates. //! -//! A claim must never block the work forever if its owner dies, so a claim whose -//! owner is no longer alive is *stale* and is taken over. (An OS file lock would -//! be released on death automatically, but it is per-open-file-description, not -//! per-process, which makes it unreliable across `fork`.) +//! A claim is an **OS byte-range lock** on a small `*.lock` file: +//! +//! * unix: `fcntl(F_SETLK)`, a *process-scoped* lock. The kernel releases it the +//! instant the process dies, however it dies, and a `fork` child does not +//! inherit it. +//! * Windows: `LockFileEx` on the handle, released when the handle closes or the +//! process exits. +//! +//! Because the kernel is the arbiter there is nothing to go stale: no owner pid +//! to probe, no start time, no lease to expire, no takeover protocol. That is +//! also why it is correct across **pid namespaces** (two containers sharing a +//! volume, both with a pid 1) and across pid reuse, where recording a pid in a +//! file is not. The lock file itself may be left behind by a dead holder; it is +//! just an unlocked file, and anyone may lock it. +//! +//! Two costs of process-scoped locks, both handled here: +//! +//! * They do not exclude *threads* of one process, so an in-process registry of +//! held paths does that. +//! * POSIX drops a process's locks on a file when it closes **any** descriptor +//! for that file. So a lock file is never opened twice by one process: every +//! open (claim or probe) registers the path first, and a claim closes its +//! descriptor *before* leaving the registry. Registry keys are canonicalised +//! (`/tmp` and `/private/tmp`, relative and absolute, are one file). +//! +//! **Unsupported locks.** Some filesystems reject byte-range locks outright (NFS +//! without lockd: `ENOLCK`; FUSE/9p: `EOPNOTSUPP`/`ENOSYS`/`EINVAL`). The base +//! scheme (`O_EXCL` files) worked there, so refusing to claim would be a +//! regression: a queue that never drains. Instead such a claim degrades to +//! thread-only exclusion (announced once on stderr), and a probe reports +//! [`Probe::Unknown`] so the caller falls back to its pid check. +//! +//! **Locks that succeed but do not exclude** (Docker Desktop bind mounts, WSL2 +//! drvfs/9p, NFS mounted `nolock`) cannot be detected. Processes sharing a +//! directory over such a mount can deliver a report twice; run one SDK host per +//! directory there. +//! +//! The SDK relies on no file timestamps for claims; the only ones it reads are +//! the grace windows for half-written files in `recovery` and `queue`, which +//! assume clocks that agree to within seconds. +//! +//! None of this runs on the crash path; it allocates freely. -use std::io::Write; +use std::cell::UnsafeCell; +use std::fs::{File, OpenOptions}; use std::path::{Path, PathBuf}; -use std::time::{Duration, SystemTime}; - -/// A claim that has no readable owner is given this long before it is treated as -/// stale: its creator writes the pid a moment after creating the file. -const UNOWNED_GRACE: Duration = Duration::from_secs(5); +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; /// Holds a claim; dropping it releases the claim. #[derive(Debug)] pub struct Claim { - path: PathBuf, + key: PathBuf, + file: Option, + os_locked: bool, +} + +impl Claim { + /// Whether the kernel enforces this claim against other processes. `false` + /// means the filesystem rejected locks and only threads are excluded. + pub(crate) fn is_os_locked(&self) -> bool { + self.os_locked + } } impl Drop for Claim { fn drop(&mut self) { - let _ = std::fs::remove_file(&self.path); + // Close (releasing the OS lock) strictly before leaving the registry, so + // a thread that acquires next never has its lock dropped by our close. + drop(self.file.take()); + registry().remove(&self.key); } } -/// Try to claim `path`. `None` if another live process (or another thread of this -/// one) holds it, or if the claim file cannot be created at all. +/// Try to claim `path` (created if absent). `None` if another process, or another +/// thread of this one, holds it, or if it cannot be claimed. pub fn try_claim(path: &Path) -> Option { - match claim_state(path) { - ClaimState::Held(claim) => Some(claim), - ClaimState::Taken | ClaimState::Unavailable => None, + match acquire(path) { + Acquire::Got(claim) => Some(claim), + Acquire::Held | Acquire::Unavailable => None, } } /// Like [`try_claim`] for a directory that may be full or read-only, where the -/// claim file cannot be created but the work (recover a crash, deliver a queued +/// lock file cannot be created but the work (recover a crash, deliver a queued /// report) is still worth doing: `Some(None)` means "no claim could be made, /// proceed unclaimed". The cost is that two processes sharing such a directory /// may both do the work, and the backend deduplicates. `None` means another -/// process holds the claim. +/// process (or thread) holds the claim. pub fn try_claim_lenient(path: &Path) -> Option> { - match claim_state(path) { - ClaimState::Held(claim) => Some(Some(claim)), - ClaimState::Unavailable => Some(None), - ClaimState::Taken => None, + match acquire(path) { + Acquire::Got(claim) => Some(Some(claim)), + Acquire::Unavailable => Some(None), + Acquire::Held => None, + } +} + +/// Whether the lock file at `path` is held by a live process. +pub(crate) enum Probe { + Held, + Free, + /// There is no lock file (any longer). Not the same as [`Probe::Free`]: a + /// caller that saw the file exist and finds it gone must not conclude that its + /// holder died, since a session without kernel locks unlinks its lock file. + Vanished, + /// The filesystem cannot say (it rejects locks): use another liveness signal. + Unknown, +} + +/// Is `path` currently claimed? A missing file is [`Probe::Vanished`]. A file that cannot be +/// inspected is [`Probe::Unknown`], never assumed held: a crashed session's +/// leftover lock file must not hide the crash on a volume that has become +/// read-only. +/// +/// On unix this asks (`F_GETLK`) without taking the lock, so a probe can never +/// steal the lock from a creator that has made the file but not yet locked it. +pub(crate) fn probe(path: &Path) -> Probe { + let key = key_of(path); + if !registry().insert(&key) { + return Probe::Held; // held by a thread of this process, or being probed + } + // Read-only: a querying lock call needs no write access, and a leftover lock + // file on a read-only volume (or with 0444 mode) must still be probeable. + #[cfg(test)] + BEFORE_PROBE_OPEN.with(|h| { + if let Some(hook) = h.borrow().as_ref() { + hook(); + } + }); + let result = match OpenOptions::new().read(true).open(path) { + Ok(file) => { + let r = match test_is_locked(&file) { + Ok(Peek::Free) => Probe::Free, + Ok(Peek::Held) => Probe::Held, + // Cannot tell (no lock support, or the call failed): let the + // caller use another liveness signal rather than assume a live + // holder and hide a crash forever. + Ok(Peek::Unsupported) | Err(_) => Probe::Unknown, + }; + drop(file); // close before leaving the registry + r + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Probe::Vanished, + Err(_) => Probe::Unknown, + }; + registry().remove(&key); + result +} + +/// Remove the lock file at `path` if nobody holds it (a dead holder's leftover). +pub(crate) fn remove_if_free(path: &Path) { + if let Some(claim) = try_claim(path) { + let _ = std::fs::remove_file(path); + drop(claim); } } -enum ClaimState { - Held(Claim), - Taken, +/// Forget every registry entry. Called in a `fork` child before it claims +/// anything: the child inherits the parent's registry (entries it does not own, +/// possibly mid-update under the parent's spin lock) and none of its locks. +pub fn reset_after_fork() { + registry().reset(); +} + +enum Acquire { + Got(Claim), + Held, Unavailable, } -fn claim_state(path: &Path) -> ClaimState { - for _ in 0..2 { - match std::fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(path) - { - Ok(mut file) => { - let _ = write!(file, "{}", std::process::id()); - return ClaimState::Held(Claim { - path: path.to_path_buf(), +fn acquire(path: &Path) -> Acquire { + let key = key_of(path); + if !registry().insert(&key) { + return Acquire::Held; // another thread of this process + } + let file = match OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(path) + { + Ok(f) => f, + Err(_) => { + registry().remove(&key); + return Acquire::Unavailable; + } + }; + match test_try_lock(&file) { + Ok(Lock::Got) => Acquire::Got(Claim { + key, + file: Some(file), + os_locked: true, + }), + Ok(Lock::Contended) => { + drop(file); + registry().remove(&key); + Acquire::Held + } + Ok(Lock::Unsupported) => { + static ANNOUNCED: AtomicBool = AtomicBool::new(false); + crate::util::warn_once( + &ANNOUNCED, + "this filesystem does not support file locks; processes sharing the data \ + directory cannot be kept from delivering the same report twice", + ); + Acquire::Got(Claim { + key, + file: Some(file), + os_locked: false, + }) + } + Err(_) => { + drop(file); + registry().remove(&key); + Acquire::Unavailable + } + } +} + +/// The registry key for `path`: its directory canonicalised, so every spelling of +/// one file maps to one key. Falls back to the path as given. +fn key_of(path: &Path) -> PathBuf { + match (path.parent(), path.file_name()) { + (Some(parent), Some(name)) => { + let parent = if parent.as_os_str().is_empty() { + Path::new(".") + } else { + parent + }; + std::fs::canonicalize(parent) + .map(|p| p.join(name)) + .unwrap_or_else(|_| path.to_path_buf()) + } + _ => path.to_path_buf(), + } +} + +/// Outcome of an attempt to take a lock. +enum Lock { + Got, + Contended, + Unsupported, +} + +/// Outcome of asking whether a lock is held. +enum Peek { + Free, + Held, + Unsupported, +} + +#[cfg(test)] +thread_local! { + /// Test hook: make this thread's lock calls behave as on a filesystem that + /// rejects locks. + pub(crate) static FORCE_UNSUPPORTED: std::cell::Cell = const { std::cell::Cell::new(false) }; +} + +#[cfg(test)] +type Hook = Box; + +#[cfg(test)] +thread_local! { + /// Test hook: runs inside `probe` just before it opens the file, to simulate + /// the lock file vanishing between a caller's existence check and the probe. + pub(crate) static BEFORE_PROBE_OPEN: std::cell::RefCell> = + const { std::cell::RefCell::new(None) }; +} + +fn test_try_lock(file: &File) -> std::io::Result { + #[cfg(test)] + if FORCE_UNSUPPORTED.with(|f| f.get()) { + return Ok(Lock::Unsupported); + } + os::try_lock(file) +} + +fn test_is_locked(file: &File) -> std::io::Result { + #[cfg(test)] + if FORCE_UNSUPPORTED.with(|f| f.get()) { + return Ok(Peek::Unsupported); + } + os::is_locked(file) +} + +/// The paths this process currently holds, keyed by pid so that a `fork` child, +/// which inherits a copy of this memory, ignores the parent's entries. +/// +/// A hand-rolled spin lock rather than a `Mutex`: if another thread held a +/// `Mutex` at the instant of `fork`, the child's copy would stay locked forever +/// and its first claim would deadlock. The fork hook calls [`reset_after_fork`] +/// to clear it; as a second line the lock word records the holder's pid, so a +/// child that finds it held by a different pid breaks it. +struct Registry { + owner: AtomicU32, + held: UnsafeCell>, +} + +// SAFETY: `held` is only touched while `owner` is held (see `with`). +unsafe impl Sync for Registry {} + +static REGISTRY: Registry = Registry { + owner: AtomicU32::new(0), + held: UnsafeCell::new(Vec::new()), +}; + +fn registry() -> &'static Registry { + ®ISTRY +} + +impl Registry { + fn with(&self, f: impl FnOnce(&mut Vec<(u32, PathBuf)>) -> R) -> R { + let pid = std::process::id(); + loop { + match self + .owner + .compare_exchange(0, pid, Ordering::Acquire, Ordering::Relaxed) + { + Ok(_) => break, + Err(holder) if holder != pid => { + // Held by a pid that is not us: a parent's thread at the time + // of `fork`. It does not exist here. Take the lock, and + // discard (leak, never drop) the vector it may have left + // half-modified. + if self + .owner + .compare_exchange(holder, pid, Ordering::Acquire, Ordering::Relaxed) + .is_ok() + { + // SAFETY: we now hold the lock; the old value is leaked + // because its allocator state may be mid-update. + unsafe { std::ptr::write(self.held.get(), Vec::new()) }; + break; + } + } + Err(_) => std::thread::yield_now(), + } + } + // SAFETY: exclusive while `owner == pid` (held above). + let out = f(unsafe { &mut *self.held.get() }); + self.owner.store(0, Ordering::Release); + out + } + + /// Record `path` as held by this process; `false` if it already is. + fn insert(&self, path: &Path) -> bool { + let pid = std::process::id(); + self.with(|held| { + if held.iter().any(|(p, q)| *p == pid && q == path) { + return false; + } + // Entries of a different pid are a forked-from parent's; drop them. + held.retain(|(p, _)| *p == pid); + held.push((pid, path.to_path_buf())); + true + }) + } + + fn remove(&self, path: &Path) { + let pid = std::process::id(); + self.with(|held| held.retain(|(p, q)| !(*p == pid && q == path))); + } + + /// Discard everything, including a lock word left held by a thread that does + /// not exist in this process. Only for a freshly forked, single-threaded child. + fn reset(&self) { + // SAFETY: the caller is the only thread in the process, so nothing else + // can be using the registry; the old vector is leaked, not dropped, as its + // allocator state may be mid-update from the parent's other threads. + unsafe { std::ptr::write(self.held.get(), Vec::new()) }; + self.owner.store(0, Ordering::Release); + } +} + +#[cfg(unix)] +mod os { + use super::{Lock, Peek}; + use std::fs::File; + use std::io; + use std::os::unix::io::AsRawFd; + + fn whole() -> libc::flock { + // SAFETY: an all-zero `flock` is a valid plain-data value. + let mut fl: libc::flock = unsafe { std::mem::zeroed() }; + fl.l_type = libc::F_WRLCK as _; + fl.l_whence = libc::SEEK_SET as _; + fl.l_start = 0; + fl.l_len = 1; + fl + } + + /// Errors meaning "this filesystem does not do byte-range locks". + fn is_unsupported(errno: i32) -> bool { + errno == libc::ENOLCK + || errno == libc::ENOSYS + || errno == libc::EINVAL + || errno == libc::EOPNOTSUPP + || errno == libc::ENOTSUP + } + + /// Take a non-blocking exclusive lock on byte 0. + pub fn try_lock(file: &File) -> io::Result { + let fl = whole(); + loop { + // SAFETY: `fd` is a valid open descriptor and `fl` a valid `flock`. + let ret = unsafe { libc::fcntl(file.as_raw_fd(), libc::F_SETLK, &fl) }; + if ret == 0 { + return Ok(Lock::Got); + } + let err = io::Error::last_os_error(); + match err.raw_os_error() { + Some(libc::EAGAIN) | Some(libc::EACCES) => return Ok(Lock::Contended), + Some(libc::EINTR) => continue, + Some(e) if is_unsupported(e) => return Ok(Lock::Unsupported), + _ => return Err(err), + } + } + } + + /// Ask whether a peer holds the lock, without taking it. + // `l_type` and `F_UNLCK` are different integer types per platform; widening + // both to i32 is the portable spelling, a no-op cast on some of them. + #[allow(clippy::unnecessary_cast)] + pub fn is_locked(file: &File) -> io::Result { + let mut fl = whole(); + loop { + // SAFETY: as above; `F_GETLK` only writes the conflicting lock into `fl`. + let ret = unsafe { libc::fcntl(file.as_raw_fd(), libc::F_GETLK, &mut fl) }; + if ret == 0 { + return Ok(if i32::from(fl.l_type) == libc::F_UNLCK as i32 { + Peek::Free + } else { + Peek::Held }); } - Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => { - if !is_stale(path) { - return ClaimState::Taken; - } - // Take over a claim whose owner is gone, then retry the create. - // Losing the race to another taker just means `create_new` fails - // again and we report the claim as held. - match std::fs::remove_file(path) { - Ok(()) => {} - // Someone else took it over first: retry the create. - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - // A read-only or full volume will not let us clear it, so no - // claim can ever be made here: not "held by someone else". - Err(_) => return ClaimState::Unavailable, - } + let err = io::Error::last_os_error(); + match err.raw_os_error() { + Some(libc::EINTR) => continue, + Some(e) if is_unsupported(e) => return Ok(Peek::Unsupported), + _ => return Err(err), } - // Directory missing / unwritable: nothing can be claimed, and the - // caller's own I/O will fail more usefully than we can here. - Err(_) => return ClaimState::Unavailable, } } - ClaimState::Taken } -fn is_stale(path: &Path) -> bool { - let owner = std::fs::read_to_string(path).unwrap_or_default(); - match owner.trim().parse::() { - // Our own pid is a live owner: another thread of this process holds it. - Ok(pid) => pid != std::process::id() && !crate::recovery::process_is_alive(pid), - Err(_) => std::fs::metadata(path) - .and_then(|m| m.modified()) - .ok() - .and_then(|t| SystemTime::now().duration_since(t).ok()) - .is_some_and(|age| age > UNOWNED_GRACE), +#[cfg(windows)] +mod os { + use super::{Lock, Peek}; + use std::ffi::c_void; + use std::fs::File; + use std::io; + use std::os::windows::io::AsRawHandle; + + const LOCKFILE_FAIL_IMMEDIATELY: u32 = 1; + const LOCKFILE_EXCLUSIVE_LOCK: u32 = 2; + const ERROR_LOCK_VIOLATION: i32 = 33; + // The filesystem has no byte-range locking (some network shares, FAT shims). + const ERROR_INVALID_FUNCTION: i32 = 1; + const ERROR_NOT_SUPPORTED: i32 = 50; + const ERROR_CALL_NOT_IMPLEMENTED: i32 = 120; + + #[repr(C)] + struct Overlapped { + internal: usize, + internal_high: usize, + offset: u32, + offset_high: u32, + event: *mut c_void, + } + + extern "system" { + fn LockFileEx( + file: *mut c_void, + flags: u32, + reserved: u32, + len_low: u32, + len_high: u32, + overlapped: *mut Overlapped, + ) -> i32; + } + + /// Take a non-blocking exclusive lock on byte 0. + pub fn try_lock(file: &File) -> io::Result { + let mut ov = Overlapped { + internal: 0, + internal_high: 0, + offset: 0, + offset_high: 0, + event: std::ptr::null_mut(), + }; + // SAFETY: the handle is valid for the call and `ov` outlives it. + let ok = unsafe { + LockFileEx( + file.as_raw_handle(), + LOCKFILE_EXCLUSIVE_LOCK | LOCKFILE_FAIL_IMMEDIATELY, + 0, + 1, + 0, + &mut ov, + ) + }; + if ok != 0 { + return Ok(Lock::Got); + } + let err = io::Error::last_os_error(); + match err.raw_os_error() { + Some(ERROR_LOCK_VIOLATION) => Ok(Lock::Contended), + Some(ERROR_INVALID_FUNCTION | ERROR_NOT_SUPPORTED | ERROR_CALL_NOT_IMPLEMENTED) => { + Ok(Lock::Unsupported) + } + _ => Err(err), + } + } + + /// Windows has no query-only call, so this takes the lock; the caller closes + /// the handle straight after, which releases it. + pub fn is_locked(file: &File) -> io::Result { + Ok(match try_lock(file)? { + Lock::Got => Peek::Free, + Lock::Contended => Peek::Held, + Lock::Unsupported => Peek::Unsupported, + }) + } +} + +/// No file locks: claims exclude threads of this process only. +#[cfg(not(any(unix, windows)))] +mod os { + use super::{Lock, Peek}; + pub fn try_lock(_file: &std::fs::File) -> std::io::Result { + Ok(Lock::Unsupported) + } + pub fn is_locked(_file: &std::fs::File) -> std::io::Result { + Ok(Peek::Unsupported) } } @@ -129,58 +561,106 @@ mod tests { #[test] fn a_claim_is_exclusive_and_released_on_drop() { let dir = tmp(); - let path = dir.join("x.claim"); + let path = dir.join("x.lock"); let first = try_claim(&path).expect("first claim"); + assert!(first.is_os_locked()); assert!(try_claim(&path).is_none(), "already held"); + assert!(matches!(probe(&path), Probe::Held)); drop(first); + assert!(matches!(probe(&path), Probe::Free)); assert!(try_claim(&path).is_some(), "free again after release"); } #[test] - fn a_claim_whose_owner_is_dead_is_taken_over() { + fn a_leftover_lock_file_from_a_dead_holder_is_just_free() { let dir = tmp(); - let path = dir.join("x.claim"); - // A pid that cannot be alive: far above any real pid range. - std::fs::write(&path, "4294967290").unwrap(); - assert!(try_claim(&path).is_some(), "stale claim must be reclaimed"); + let path = dir.join("x.lock"); + // Contents are irrelevant: a dead holder's pid, even one that collides + // with a live process (here, our own), must not block anything. + std::fs::write(&path, std::process::id().to_string()).unwrap(); + assert!(matches!(probe(&path), Probe::Free)); + assert!(try_claim(&path).is_some()); } #[test] - fn a_claim_held_by_a_live_process_is_respected() { + fn a_missing_lock_file_probes_free_and_is_not_created() { let dir = tmp(); - let path = dir.join("x.claim"); - // pid 1 is always alive on unix; on other platforms liveness is "dead". - std::fs::write(&path, "1").unwrap(); - #[cfg(unix)] + let path = dir.join("nope.lock"); + assert!(matches!(probe(&path), Probe::Vanished)); + assert!(!path.exists()); + } + + #[test] + fn probing_never_takes_or_drops_a_lock() { + // POSIX drops a process's locks on a file when ANY descriptor for it + // closes; the registry must keep a second open from ever happening. + let dir = tmp(); + let path = dir.join("x.lock"); + let held = try_claim(&path).unwrap(); assert!(try_claim(&path).is_none()); - #[cfg(not(unix))] + assert!(matches!(probe(&path), Probe::Held)); + assert!( + try_claim(&path).is_none(), + "probing must not have closed our lock" + ); + drop(held); + // And a probe of a free file leaves it free for the next claimer. + assert!(matches!(probe(&path), Probe::Free)); assert!(try_claim(&path).is_some()); } - #[cfg(unix)] #[test] - fn a_stale_claim_that_cannot_be_removed_is_unavailable_not_taken() { - use std::os::unix::fs::PermissionsExt; + fn one_file_under_two_spellings_is_one_claim() { let dir = tmp(); - let path = dir.join("x.claim"); - std::fs::write(&path, "4294967290").unwrap(); - std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o555)).unwrap(); - if std::fs::remove_file(dir.join("nonexistent-probe")).is_ok() - || std::fs::write(dir.join("probe"), b"x").is_ok() - { - let _ = std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o755)); - return; // permission bits not enforced (root) + let path = dir.join("x.lock"); + let held = try_claim(&path).unwrap(); + // The same file through a `..` detour. + let detour = dir.join("sub").join("..").join("x.lock"); + std::fs::create_dir_all(dir.join("sub")).unwrap(); + assert!(try_claim(&detour).is_none()); + drop(held); + } + + #[test] + fn racing_claimers_of_one_path_yield_exactly_one_holder() { + for _ in 0..100 { + let dir = tmp(); + let path = dir.join("x.lock"); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(8)); + let handles: Vec<_> = (0..8) + .map(|_| { + let (path, barrier) = (path.clone(), barrier.clone()); + std::thread::spawn(move || { + barrier.wait(); + let claim = try_claim(&path); + // Hold until every thread has tried. + barrier.wait(); + claim.is_some() + }) + }) + .collect(); + let holders = handles + .into_iter() + .map(|h| h.join().unwrap()) + .filter(|held| *held) + .count(); + assert_eq!(holders, 1, "exactly one claimer may win"); + let _ = std::fs::remove_dir_all(&dir); } - // Proceed unclaimed instead of skipping the work as "held elsewhere". - assert!(matches!(try_claim_lenient(&path), Some(None))); - let _ = std::fs::set_permissions(&dir, std::fs::Permissions::from_mode(0o755)); } #[test] - fn a_freshly_created_unowned_claim_is_not_stolen() { + fn a_filesystem_that_rejects_locks_degrades_to_thread_only_claims() { let dir = tmp(); - let path = dir.join("x.claim"); - std::fs::write(&path, "").unwrap(); // creator has not written its pid yet - assert!(try_claim(&path).is_none()); + let path = dir.join("x.lock"); + FORCE_UNSUPPORTED.with(|f| f.set(true)); + let first = try_claim(&path).expect("must still claim"); + assert!(!first.is_os_locked()); + assert!(try_claim(&path).is_none(), "threads are still excluded"); + drop(first); + assert!(try_claim(&path).is_some(), "and it is released on drop"); + // A probe cannot say: the caller must use another signal. + assert!(matches!(probe(&path), Probe::Unknown)); + FORCE_UNSUPPORTED.with(|f| f.set(false)); } } diff --git a/crates/bugsee-core/src/lib.rs b/crates/bugsee-core/src/lib.rs index e5793d7..e0e9b9b 100644 --- a/crates/bugsee-core/src/lib.rs +++ b/crates/bugsee-core/src/lib.rs @@ -24,6 +24,7 @@ pub mod fork; pub mod model; pub mod origin; pub mod panic_info; +pub(crate) mod process; pub mod queue; pub mod recovery; pub mod reporting; diff --git a/crates/bugsee-core/src/process.rs b/crates/bugsee-core/src/process.rs new file mode 100644 index 0000000..c857e46 --- /dev/null +++ b/crates/bugsee-core/src/process.rs @@ -0,0 +1,142 @@ +// +// process.rs +// bugsee-core +// +// Copyright © 2026 Bugsee. All rights reserved. +// +//! Process liveness probe, used only for **legacy** session markers. +//! +//! Current sessions and claims are arbitrated by OS file locks (see `claim`), +//! which need no pid and are correct across pid namespaces. A session marker +//! written by an older SDK build has no lock beside it, so for those alone the +//! recorded pid is probed here. The probe is inherently single-namespace. + +/// Whether a process with `pid` currently exists and has not terminated. A +/// zombie (exited, not yet reaped by its parent) counts as dead. +/// +/// Unix: `kill(pid, 0)`; `0` (permitted) or `EPERM` (exists, not permitted) both +/// mean alive, `ESRCH` means gone. Windows: the process can be opened and its +/// handle is not yet signalled. +#[cfg(unix)] +pub(crate) fn is_alive(pid: u32) -> bool { + if pid == 0 { + return false; + } + // SAFETY: `kill` with signal 0 performs only an error/permission check and + // has no side effects; it is safe to call with any pid value. + let ret = unsafe { libc::kill(pid as libc::pid_t, 0) }; + (ret == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)) + && !is_zombie(pid) +} + +#[cfg(any(target_os = "linux", target_os = "android"))] +fn is_zombie(pid: u32) -> bool { + let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else { + return false; + }; + // `pid (comm) state ...`: comm may contain spaces and parens, so look after + // the LAST ')'. + stat.rfind(')') + .and_then(|i| stat[i + 1..].split_whitespace().next()) + .is_some_and(|state| state == "Z" || state == "X") +} + +#[cfg(target_os = "macos")] +fn is_zombie(pid: u32) -> bool { + // The "short" BSD info, unlike the full one, is still available for a zombie. + // SAFETY: an all-zero `proc_bsdshortinfo` is valid plain data, and + // `proc_pidinfo` writes at most `size_of` bytes into it. + let mut info: libc::proc_bsdshortinfo = unsafe { std::mem::zeroed() }; + let size = std::mem::size_of::() as libc::c_int; + let n = unsafe { + libc::proc_pidinfo( + pid as libc::c_int, + libc::PROC_PIDT_SHORTBSDINFO, + 0, + (&mut info as *mut libc::proc_bsdshortinfo).cast(), + size, + ) + }; + const SZOMB: u32 = 5; + // Observed on macOS: for an exited-but-unreaped child `kill(pid, 0)` still + // succeeds while `proc_pidinfo` returns 0 (no info), whereas any live process, + // including other users', reports its info. So "no info" means zombie here. + n == 0 || (n == size && info.pbsi_status == SZOMB) +} + +#[cfg(all( + unix, + not(any(target_os = "linux", target_os = "android", target_os = "macos")) +))] +fn is_zombie(_pid: u32) -> bool { + false +} + +#[cfg(windows)] +pub(crate) fn is_alive(pid: u32) -> bool { + if pid == 0 { + return false; + } + // SAFETY: plain Win32 calls; the handle is closed on every path. + unsafe { + let handle = win::OpenProcess(win::SYNCHRONIZE, 0, pid); + if handle.is_null() { + // Access denied means the process exists but belongs to someone else. + return std::io::Error::last_os_error().raw_os_error() + == Some(win::ERROR_ACCESS_DENIED); + } + // A process handle is signalled once the process has terminated. + let alive = win::WaitForSingleObject(handle, 0) == win::WAIT_TIMEOUT; + win::CloseHandle(handle); + alive + } +} + +#[cfg(not(any(unix, windows)))] +pub(crate) fn is_alive(_pid: u32) -> bool { + false +} + +#[cfg(windows)] +mod win { + //! The few kernel32 calls needed, declared directly: kernel32 is always + //! linked, so this avoids a dependency for three functions. + use std::ffi::c_void; + + pub type Handle = *mut c_void; + pub const SYNCHRONIZE: u32 = 0x0010_0000; + pub const WAIT_TIMEOUT: u32 = 258; + pub const ERROR_ACCESS_DENIED: i32 = 5; + + extern "system" { + pub fn OpenProcess(access: u32, inherit: i32, pid: u32) -> Handle; + pub fn CloseHandle(handle: Handle) -> i32; + pub fn WaitForSingleObject(handle: Handle, millis: u32) -> u32; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn own_process_is_alive_and_an_impossible_pid_is_not() { + assert!(is_alive(std::process::id())); + assert!(!is_alive(0)); + assert!(!is_alive(4_294_967_290)); + } + + #[cfg(unix)] + #[test] + fn a_zombie_is_dead() { + let mut child = std::process::Command::new("true").spawn().unwrap(); + let pid = child.id(); + // Wait for it to exit without reaping it (a non-consuming probe). + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while is_alive(pid) && std::time::Instant::now() < deadline { + std::thread::sleep(std::time::Duration::from_millis(10)); + } + assert!(!is_alive(pid), "exited but unreaped child counts as dead"); + let _ = child.wait(); + } +} diff --git a/crates/bugsee-core/src/queue.rs b/crates/bugsee-core/src/queue.rs index 2a1cbd2..e6298ff 100644 --- a/crates/bugsee-core/src/queue.rs +++ b/crates/bugsee-core/src/queue.rs @@ -35,21 +35,21 @@ impl QueuedReport { fn endpoint_path(&self) -> PathBuf { sibling(&self.bundle, "endpoint") } - fn claim_path(&self) -> PathBuf { - sibling(&self.bundle, "claim") + fn lock_path(&self) -> PathBuf { + sibling(&self.bundle, "lock") } } /// Claim a queued report for delivery, so that of several processes draining one /// shared queue exactly one uploads it. Released when the guard drops. pub fn try_claim(report: &QueuedReport) -> Option { - crate::claim::try_claim(&report.claim_path()) + crate::claim::try_claim(&report.lock_path()) } /// [`try_claim`] that still lets delivery proceed when the queue directory cannot /// accept a claim file (full or read-only); see [`crate::claim::try_claim_lenient`]. pub fn try_claim_lenient(report: &QueuedReport) -> Option> { - crate::claim::try_claim_lenient(&report.claim_path()) + crate::claim::try_claim_lenient(&report.lock_path()) } fn sibling(bundle: &Path, ext: &str) -> PathBuf { @@ -157,6 +157,9 @@ pub fn remove(report: &QueuedReport) { let _ = std::fs::remove_file(report.req_path()); let _ = std::fs::remove_file(report.meta_path()); let _ = std::fs::remove_file(report.endpoint_path()); + // The delivering claimant's own lock file (removing a held lock file is fine: + // the item is gone, so no one will look for it). + let _ = std::fs::remove_file(report.lock_path()); } /// The bundle file's base name (for reconstructing an [`AssembledReport`]). @@ -237,6 +240,11 @@ pub fn blacklist_add(data_dir: &Path, sigs: &[String]) { /// crash-atomic across its multiple files). Without this they accumulate /// unbounded. Call once at startup on the worker thread, before any enqueue, so /// it never races this process's own queue writes. +/// +/// A peer sharing the directory may be mid-`enqueue` right now (sidecars written, +/// bundle not yet renamed into place), and its in-flight files look exactly like +/// orphans. Reaping them would strand the bundle without its `.req`, losing the +/// report, so only files untouched for [`ORPHAN_GRACE`] are reaped. pub fn gc_orphans(data_dir: &Path) { let dir = queue_dir(data_dir); let read = match std::fs::read_dir(&dir) { @@ -255,16 +263,33 @@ pub fn gc_orphans(data_dir: &Path) { .strip_suffix(".req") .or_else(|| name.strip_suffix(".meta")) .or_else(|| name.strip_suffix(".endpoint")) + .or_else(|| name.strip_suffix(".lock")) + // Leftovers of an earlier claim scheme. .or_else(|| name.strip_suffix(".claim")) + .or_else(|| name.strip_suffix(".claim.takeover")) .or_else(|| name.strip_suffix(".tmp")); if let Some(bundle_name) = bundle_name { - if !dir.join(bundle_name).exists() { + if !dir.join(bundle_name).exists() && untouched_for(&path, ORPHAN_GRACE) { let _ = std::fs::remove_file(&path); } } } } +/// How long an orphan-looking queue file must sit untouched before it is reaped +/// (see [`gc_orphans`]); far longer than any `enqueue` takes. +const ORPHAN_GRACE: std::time::Duration = std::time::Duration::from_secs(60); + +/// Whether `path` was last modified at least `grace` ago. Unknown or future-dated +/// times count as recent: the conservative answer never disturbs a live peer. +pub(crate) fn untouched_for(path: &Path, grace: std::time::Duration) -> bool { + std::fs::metadata(path) + .and_then(|m| m.modified()) + .ok() + .and_then(|t| t.elapsed().ok()) + .is_some_and(|age| age >= grace) +} + /// Whether any of `sigs` is locally blacklisted (report should be suppressed). pub fn any_blacklisted(data_dir: &Path, sigs: &[String]) -> bool { if sigs.is_empty() { @@ -273,3 +298,30 @@ pub fn any_blacklisted(data_dir: &Path, sigs: &[String]) -> bool { let set = load_blacklist(data_dir); sigs.iter().any(|s| set.contains(s)) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn gc_spares_a_peers_in_flight_enqueue_and_reaps_old_orphans() { + let data = std::env::temp_dir().join(format!("bugsee-qgc-{}", crate::util::random_hex(8))); + let dir = queue_dir(&data); + std::fs::create_dir_all(&dir).unwrap(); + // A peer mid-enqueue: sidecars written, bundle not yet renamed into place. + let fresh = dir.join("a.bundle.zip.req"); + std::fs::write(&fresh, "{}").unwrap(); + let old = dir.join("b.bundle.zip.req"); + std::fs::write(&old, "{}").unwrap(); + let f = std::fs::OpenOptions::new().write(true).open(&old).unwrap(); + f.set_modified(std::time::SystemTime::now() - ORPHAN_GRACE * 2) + .unwrap(); + drop(f); + + gc_orphans(&data); + + assert!(fresh.exists(), "an in-flight enqueue must not be reaped"); + assert!(!old.exists(), "an old orphan must be reaped"); + let _ = std::fs::remove_dir_all(&data); + } +} diff --git a/crates/bugsee-core/src/recovery.rs b/crates/bugsee-core/src/recovery.rs index 65f8fd0..fe50447 100644 --- a/crates/bugsee-core/src/recovery.rs +++ b/crates/bugsee-core/src/recovery.rs @@ -80,32 +80,9 @@ pub fn find_pending(data_dir: &Path, current_generation: u64) -> Vec= current_generation { continue; } - // Decide from the marker's recorded owner pid whether this is a real - // crash or a still-running shared-dir peer. - let owner = std::fs::read_to_string(&marker).unwrap_or_default(); - let owner = owner.trim(); - if owner.is_empty() { - // Present but not yet populated: `Session::begin` creates the marker - // (O_EXCL) and writes the pid a moment later, so a peer launching - // concurrently can observe this empty window. Skip only while the - // marker is FRESH — a peer writes its pid within microseconds. An empty - // marker older than the grace window is a session that died in the - // create→write gap (or whose pid write silently failed on a full - // disk); recover it so its crash is surfaced and its generation - // reclaimed instead of being re-scanned and leaked on every future - // launch (F28). No pid ⇒ no liveness check; fall through to recover. - if marker_is_fresh(&marker) { - continue; - } - } else if let Ok(pid) = owner.parse::() { - // Skip a marker still owned by a live *other* process. Our own pid is - // never skipped (a live process cannot share our pid, so a marker - // bearing it is a dead prior incarnation — recover it). - if pid != self_pid && process_is_alive(pid) { - continue; - } + if !session_is_crashed(data_dir, generation, &marker, self_pid) { + continue; } - // Non-empty but unparseable content ⇒ the owner is gone; recover. let parts_dir = data_dir.join("parts").join(generation.to_string()); let minidump = exists_opt(parts_dir.join(MINIDUMP_NAME)); let crash_info = exists_opt(parts_dir.join(CRASH_INFO_NAME)); @@ -125,6 +102,90 @@ fn exists_opt(p: PathBuf) -> Option { p.exists().then_some(p) } +/// Whether the session `generation`, whose marker survives, has ended abnormally. +/// +/// A session whose `.lock` exists is arbitrated by that OS lock: held means a +/// live process (in any pid namespace), free means it died. The marker's pid is +/// not consulted, since it is only meaningful inside the writer's own pid +/// namespace and can be reused. Otherwise (no lock file, or the filesystem cannot +/// say) the marker is judged by [`legacy_marker_is_dead`]. +fn session_is_crashed(data_dir: &Path, generation: u64, marker: &Path, self_pid: u32) -> bool { + let lock = crate::session::Session::lock_path(data_dir, generation); + if lock.exists() { + match crate::claim::probe(&lock) { + crate::claim::Probe::Held => return false, + // A launching session locks BEFORE it writes its pid, and a marker + // still empty and fresh may be one that is between creating the lock + // file and locking it. + crate::claim::Probe::Free => { + return !(marker_is_empty(marker) && marker_is_fresh(marker)); + } + // No lock file after all (it vanished since `exists`, e.g. a session + // without kernel locks removing it), or the filesystem cannot say: + // that is not evidence of death. Use the heartbeat / pid. + crate::claim::Probe::Vanished | crate::claim::Probe::Unknown => {} + } + } + legacy_marker_is_dead(marker, self_pid) +} + +/// How recently a lock-less session's heartbeat file must have been touched for +/// the session to count as alive. Generous because network filesystems cache file +/// attributes (Linux NFS `acregmax` defaults to 60 s), so a live beat can look much +/// older than it is. +const HEARTBEAT_STALE: std::time::Duration = std::time::Duration::from_secs(180); + +/// The heartbeat file beside a marker. Only a session that could not take a lock +/// (the filesystem rejects them) keeps one: it is touched every couple of seconds. +pub(crate) fn heartbeat_path(marker: &Path) -> PathBuf { + marker.with_extension("beat") +} + +/// Whether a marker with no usable lock beside it (written by an older SDK build, +/// or by a session on a filesystem without locks) is a crashed session rather +/// than a live peer. +/// +/// A fresh heartbeat file means alive, whatever pid the marker names: that is the +/// only signal that works across pid namespaces when there are no locks. Without +/// one, the recorded pid is probed, which is only meaningful within one pid +/// namespace. A crash is therefore recovered only once the heartbeat has gone +/// quiet for [`HEARTBEAT_STALE`] (three minutes; a restart faster than that defers recovery to a +/// later launch). +fn legacy_marker_is_dead(marker: &Path, self_pid: u32) -> bool { + // A beat whose age cannot be established (a modification time in the future: + // the file server's clock runs ahead of ours) counts as fresh, like + // `marker_is_fresh`: only a beat demonstrably quiet condemns a session. + if std::fs::metadata(heartbeat_path(marker)) + .and_then(|m| m.modified()) + .is_ok_and(|t| t.elapsed().map(|age| age < HEARTBEAT_STALE).unwrap_or(true)) + { + return false; + } + let owner = std::fs::read_to_string(marker).unwrap_or_default(); + let owner = owner.trim(); + if owner.is_empty() { + // Present but not yet populated: `Session::begin` creates the marker + // (O_EXCL) and writes the pid a moment later, so a peer launching + // concurrently can observe this empty window. Skip only while the + // marker is FRESH. An empty marker older than the grace window is a + // session that died in the create-to-write gap (or whose pid write failed + // on a full disk); recover it so its crash is surfaced and its + // generation reclaimed instead of being re-scanned on every launch. + return !marker_is_fresh(marker); + } + match owner.parse::() { + // Our own pid is never skipped (a live process cannot share our pid, so a + // marker bearing it is a dead prior incarnation). + Ok(pid) => pid == self_pid || !crate::process::is_alive(pid), + // Non-empty but unparseable content: the owner is gone. + Err(_) => true, + } +} + +fn marker_is_empty(marker: &Path) -> bool { + std::fs::read_to_string(marker).is_ok_and(|s| s.trim().is_empty()) +} + /// Grace window for an empty (pid-not-yet-written) liveness marker: a launching /// peer writes its pid within microseconds of the O_EXCL create, so a marker /// still empty after this is a dead session, not a live peer mid-launch. @@ -146,30 +207,6 @@ fn marker_is_fresh(marker: &Path) -> bool { .unwrap_or(true) } -/// Whether a process with `pid` currently exists. -/// -/// Uses `kill(pid, 0)`, which sends no signal and only probes existence: `0` -/// (permitted) or `EPERM` (exists, not permitted) both mean *alive*; `ESRCH` -/// means gone. A rare pid-reuse false-positive can defer recovering a real crash -/// to a later launch — acceptable vs. corrupting a running peer's live session. -#[cfg(unix)] -pub(crate) fn process_is_alive(pid: u32) -> bool { - if pid == 0 { - return false; - } - // SAFETY: `kill` with signal 0 performs only an error/permission check and - // has no side effects; it is safe to call with any pid value. - let ret = unsafe { libc::kill(pid as libc::pid_t, 0) }; - ret == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) -} - -/// No portable liveness probe off-unix: treat as dead so recovery still proceeds -/// (single-process is the common case; a live shared-dir peer is a unix concern). -#[cfg(not(unix))] -pub(crate) fn process_is_alive(_pid: u32) -> bool { - false -} - /// Build the crash report for a pending session. Precedence: a *fresh* aborting /// panic (`SIGABRT` + a recent panic snapshot) correlates into one managed event; /// any other native signal yields the thin native variant; a lone panic snapshot @@ -320,6 +357,7 @@ fn read_minidump(pending: &PendingSession) -> Vec { pub fn discard(data_dir: &Path, pending: &PendingSession) { let sessions = data_dir.join("sessions"); let _ = std::fs::remove_file(sessions.join(format!("{}.alive", pending.generation))); + let _ = std::fs::remove_file(sessions.join(format!("{}.beat", pending.generation))); let _ = std::fs::remove_file(sessions.join(format!("{}.attempts", pending.generation))); let _ = std::fs::remove_dir_all(&pending.parts_dir); } @@ -352,16 +390,30 @@ pub fn bump_attempts(data_dir: &Path, pending: &PendingSession) -> u32 { /// /// Lenient (see [`crate::claim::try_claim_lenient`]): a full or read-only /// directory cannot take a claim file, and that is exactly when a stranded crash -/// most needs recovering. +/// most needs recovering. `None` if a peer holds the claim or has already +/// recovered the session; `Some(None)` means proceed without a claim. pub fn claim_pending( data_dir: &Path, pending: &PendingSession, ) -> Option> { - crate::claim::try_claim_lenient( + let claim = crate::claim::try_claim_lenient( &data_dir .join("sessions") - .join(format!("{}.claim", pending.generation)), - ) + .join(format!("{}.recover.lock", pending.generation)), + )?; + // The listing that produced `pending` may be stale: a peer that held this + // claim before us can have recovered and discarded the session in between + // (its claim released, the marker gone). Recovering it now would deliver a + // phantom second report for it. The marker is removed first by `discard`, so + // its absence, re-checked while holding the claim, is conclusive. + // + // It also re-evaluates liveness under the claim: between the scan and now a + // launching session may have finished taking its lock (the scan saw it free, + // then the pid appeared), and a live session must not be torn down. + let marker = crate::session::Session::marker_path(data_dir, pending.generation); + (marker.exists() + && session_is_crashed(data_dir, pending.generation, &marker, std::process::id())) + .then_some(claim) } /// Reap leftover on-disk state from generations that no longer have a liveness @@ -388,13 +440,44 @@ pub fn gc_orphans(data_dir: &Path, current_generation: u64) { if let Ok(read) = std::fs::read_dir(&sessions) { for entry in read.flatten() { let name = entry.file_name().to_string_lossy().into_owned(); - if let Some(gen_str) = name.strip_suffix(".attempts") { + if let Some(gen_str) = name + .strip_suffix(".attempts") + .or_else(|| name.strip_suffix(".beat")) + { if let Ok(generation) = gen_str.parse::() { if generation < current_generation && !alive(generation) { let _ = std::fs::remove_file(entry.path()); } } } + // Lock files of a generation with no marker: the session lock of a + // crashed process (its marker already recovered) and the recovery + // claim. Removed only if free, so a live holder's is left alone. + let lock_gen = name + .strip_suffix(".recover.lock") + .or_else(|| name.strip_suffix(".lock")) + .and_then(|g| g.parse::().ok()); + if let Some(generation) = lock_gen { + if generation < current_generation && !alive(generation) { + crate::claim::remove_if_free(&entry.path()); + } + } + // Leftovers of an earlier claim scheme, once they are plainly dead. + let old_gen = name + .strip_suffix(".claim.takeover") + .or_else(|| name.strip_suffix(".claim")) + .and_then(|g| g.parse::().ok()); + if let Some(generation) = old_gen { + if generation < current_generation + && !alive(generation) + && crate::queue::untouched_for( + &entry.path(), + std::time::Duration::from_secs(60), + ) + { + let _ = std::fs::remove_file(entry.path()); + } + } } } } @@ -709,3 +792,164 @@ fn crash_meta(signatures: Vec, trigger: TriggerType) -> ReportMeta { attachments: Vec::new(), } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::claim::{try_claim, FORCE_UNSUPPORTED}; + use crate::session::Session; + + fn tmp() -> PathBuf { + let p = std::env::temp_dir().join(format!("bugsee-rec-{}", crate::util::random_hex(8))); + std::fs::create_dir_all(p.join("sessions")).unwrap(); + p + } + + fn marker(dir: &Path, generation: u64, content: &str) -> PathBuf { + let m = Session::marker_path(dir, generation); + std::fs::write(&m, content).unwrap(); + m + } + + fn gens(dir: &Path) -> Vec { + find_pending(dir, u64::MAX) + .iter() + .map(|p| p.generation) + .collect() + } + + #[test] + fn a_free_lock_means_the_session_died_whatever_pid_its_marker_names() { + let dir = tmp(); + marker(&dir, 1, "1"); // a pid that is alive here + std::fs::write(Session::lock_path(&dir, 1), "").unwrap(); + assert_eq!(gens(&dir), vec![1]); + } + + #[test] + fn a_held_lock_means_alive_whatever_pid_its_marker_names() { + let dir = tmp(); + marker(&dir, 1, "4294967290"); // not a pid in this namespace + let _held = try_claim(&Session::lock_path(&dir, 1)).unwrap(); + assert!(gens(&dir).is_empty()); + } + + #[test] + fn a_launching_session_between_lock_file_and_lock_is_not_recovered() { + // Created the lock file, not yet locked it, not yet written its pid. + let dir = tmp(); + let m = marker(&dir, 1, ""); + std::fs::write(Session::lock_path(&dir, 1), "").unwrap(); + assert!(gens(&dir).is_empty(), "fresh and empty: still launching"); + let f = std::fs::OpenOptions::new().write(true).open(&m).unwrap(); + f.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(120)) + .unwrap(); + drop(f); + assert_eq!(gens(&dir), vec![1], "an old empty marker is a dead launch"); + } + + #[test] + fn without_locks_the_marker_pid_decides() { + let dir = tmp(); + marker(&dir, 1, "4294967290"); + std::fs::write(Session::lock_path(&dir, 1), "").unwrap(); + #[cfg(unix)] + { + marker(&dir, 2, "1"); + std::fs::write(Session::lock_path(&dir, 2), "").unwrap(); + } + FORCE_UNSUPPORTED.with(|f| f.set(true)); + let found = gens(&dir); + FORCE_UNSUPPORTED.with(|f| f.set(false)); + // Dead pid: recovered. Live pid (unix): left alone. + assert_eq!(found, vec![1]); + } + + #[cfg(unix)] + #[test] + fn an_unwritable_leftover_lock_file_does_not_hide_a_crash() { + use std::os::unix::fs::PermissionsExt; + let dir = tmp(); + marker(&dir, 1, "4294967290"); + let lock = Session::lock_path(&dir, 1); + std::fs::write(&lock, "").unwrap(); + std::fs::set_permissions(&lock, std::fs::Permissions::from_mode(0o444)).unwrap(); + assert_eq!(gens(&dir), vec![1]); + } + + #[test] + fn a_fresh_heartbeat_keeps_a_lockless_session_alive_whatever_its_pid() { + // No lock file (the filesystem has none): the marker's pid, even our own + // (a peer in another pid namespace), must not condemn a beating session. + let dir = tmp(); + let m = marker(&dir, 1, &std::process::id().to_string()); + let beat = heartbeat_path(&m); + std::fs::write(&beat, "").unwrap(); + assert!(gens(&dir).is_empty(), "heartbeat is fresh: alive"); + let f = std::fs::OpenOptions::new().write(true).open(&beat).unwrap(); + f.set_modified(std::time::SystemTime::now() - HEARTBEAT_STALE * 2) + .unwrap(); + drop(f); + assert_eq!(gens(&dir), vec![1], "quiet heartbeat: crashed"); + } + + #[test] + fn a_lock_file_that_vanishes_under_the_scan_is_not_evidence_of_a_crash() { + // A lock-less session unlinks its `.lock` before it writes its + // heartbeat; a peer that saw the file must fall back to the beat. + let dir = tmp(); + let m = marker(&dir, 1, "4294967290"); + let lock = Session::lock_path(&dir, 1); + std::fs::write(&lock, "").unwrap(); + std::fs::write(heartbeat_path(&m), "").unwrap(); + let lock2 = lock.clone(); + crate::claim::BEFORE_PROBE_OPEN.with(|h| { + *h.borrow_mut() = Some(Box::new(move || { + let _ = std::fs::remove_file(&lock2); + })); + }); + let found = gens(&dir); + crate::claim::BEFORE_PROBE_OPEN.with(|h| *h.borrow_mut() = None); + assert!(found.is_empty(), "fresh heartbeat: the session is alive"); + } + + #[test] + fn a_future_dated_heartbeat_counts_as_alive() { + let dir = tmp(); + let m = marker(&dir, 1, "4294967290"); + let beat = heartbeat_path(&m); + std::fs::write(&beat, "").unwrap(); + let f = std::fs::OpenOptions::new().write(true).open(&beat).unwrap(); + f.set_modified(std::time::SystemTime::now() + std::time::Duration::from_secs(3600)) + .unwrap(); + drop(f); + assert!(gens(&dir).is_empty()); + } + + #[test] + fn claim_pending_gives_up_if_the_session_took_its_lock_since_the_scan() { + let dir = tmp(); + marker(&dir, 1, "4294967290"); + std::fs::write(Session::lock_path(&dir, 1), "").unwrap(); + let pending = find_pending(&dir, u64::MAX); + assert_eq!(pending.len(), 1); + // The launching session finishes locking after the scan. + let _live = try_claim(&Session::lock_path(&dir, 1)).unwrap(); + assert!(claim_pending(&dir, &pending[0]).is_none()); + } + + #[test] + fn old_scheme_claim_files_of_dead_sessions_are_swept() { + let dir = tmp(); + let claim = dir.join("sessions").join("1.claim"); + let guard = dir.join("sessions").join("1.claim.takeover"); + for p in [&claim, &guard] { + std::fs::write(p, "").unwrap(); + let f = std::fs::OpenOptions::new().write(true).open(p).unwrap(); + f.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(300)) + .unwrap(); + } + gc_orphans(&dir, 5); + assert!(!claim.exists() && !guard.exists()); + } +} diff --git a/crates/bugsee-core/src/runtime.rs b/crates/bugsee-core/src/runtime.rs index 45bffa0..02d241a 100644 --- a/crates/bugsee-core/src/runtime.rs +++ b/crates/bugsee-core/src/runtime.rs @@ -471,6 +471,9 @@ impl Recorder { &mut self, sampler: Option>, ) -> std::io::Result<()> { + // The registry of held claims is the parent's, and may be mid-update. + crate::claim::reset_after_fork(); + // The parent's threads are not ours to join; drop their handles without // running anything. std::mem::forget(self.worker.take()); @@ -698,6 +701,13 @@ impl Drop for Recorder { if let Some(session) = &self.session { session.end(); } + } else if !(worker_acked && uploader_acked) { + // A detached thread may still be writing this session's data: keep + // the liveness lock until the process exits so no peer recovers (and + // deletes) the session underneath it. + if let Some(session) = &mut self.session { + session.keep_lock_until_exit(); + } } } } @@ -910,6 +920,9 @@ fn spawn_pipeline( let worker_data_dir = data_dir.map(Path::to_path_buf); let worker_drain_dir = drain_dir.map(Path::to_path_buf); let marker = data_dir.map(|d| Session::marker_path(d, generation)); + // Recovery can run long (a host `before_send`, many sessions); it keeps this + // session's heartbeat going meanwhile, as the worker loop is not yet running. + let beat_marker = marker.clone(); let worker_upload_tx = upload_tx.clone(); let worker = std::thread::Builder::new() @@ -940,7 +953,7 @@ fn spawn_pipeline( // it must, or that crash waits for the master to restart. Claims // keep siblings from recovering the same session twice. if let Some(dir) = &worker_drain_dir { - run_recovery(&worker_shared, dir, generation); + run_recovery(&worker_shared, dir, generation, beat_marker.as_deref()); } // Best effort, once: sessions a previous default location left // behind. No generation of ours lives there, so none is skipped. @@ -950,7 +963,7 @@ fn spawn_pipeline( worker_shared.mem_blacklist.lock().extend(known); } queue::gc_orphans(dir); - run_recovery(&worker_shared, dir, u64::MAX); + run_recovery(&worker_shared, dir, u64::MAX, beat_marker.as_deref()); } })); legacy_ready.store(true, Ordering::SeqCst); @@ -979,8 +992,20 @@ fn spawn_pipeline( /// Recover any prior generation that ended abnormally, queueing it for delivery. /// The crashed session's state is discarded ONLY after the report is durably /// enqueued — a failure leaves it on disk for a later attempt (no crash loss). -fn run_recovery(shared: &Shared, data_dir: &Path, current_generation: u64) { +fn run_recovery( + shared: &Shared, + data_dir: &Path, + current_generation: u64, + heartbeat_of: Option<&Path>, +) { + let beat = || { + if let Some(marker) = heartbeat_of { + touch_heartbeat(marker); + } + }; + beat(); for pending in recovery::find_pending(data_dir, current_generation) { + beat(); // Several processes share this directory (a prefork master and its // workers, above all): exactly one recovers a given session. let Some(_claim) = recovery::claim_pending(data_dir, &pending) else { @@ -1095,8 +1120,11 @@ fn worker_loop( // sampled (F17). Instead we recv with a timeout bounded by the next rotation // deadline and run the tick whenever that deadline passes, message or not. let mut next_rotate = Instant::now() + rotate_interval; + let mut next_heartbeat = Instant::now() + HEARTBEAT_EVERY; loop { - let timeout = next_rotate.saturating_duration_since(Instant::now()); + let timeout = next_rotate + .saturating_duration_since(Instant::now()) + .min(HEARTBEAT_EVERY); match rx.recv_timeout(timeout) { Ok(Msg::Capture(entry)) => { shared.queued.fetch_sub(1, Ordering::Relaxed); @@ -1176,14 +1204,33 @@ fn worker_loop( // elapsed — via a message OR a timeout — so steady traffic can't starve it. if Instant::now() >= next_rotate { rotate_tick(store.as_mut(), &shared, &mut sampler); + next_rotate = Instant::now() + rotate_interval; + } + // Independent of the rotation interval (which a host may set large). + if Instant::now() >= next_heartbeat { if let Some(marker) = &marker { ensure_marker(&shared, marker); + touch_heartbeat(marker); } - next_rotate = Instant::now() + rotate_interval; + next_heartbeat = Instant::now() + HEARTBEAT_EVERY; } } } +/// How often the worker checks the liveness marker and, for a session that could +/// not take a file lock, touches its heartbeat file. +const HEARTBEAT_EVERY: Duration = Duration::from_secs(2); + +/// Touch this session's heartbeat file, if it keeps one (see +/// `recovery::heartbeat_path`). Never creates it: a session with a kernel lock has +/// none, and one that has ended must not leave one behind. +fn touch_heartbeat(marker: &Path) { + let beat = recovery::heartbeat_path(marker); + if let Ok(file) = std::fs::OpenOptions::new().write(true).open(&beat) { + let _ = file.set_modified(std::time::SystemTime::now()); + } +} + /// Make sure this session's liveness marker still exists. If the data directory /// was removed and recreated under us (`create_dir_all` heals the tree but not /// the marker), the next launch would not see this session as live, and its diff --git a/crates/bugsee-core/src/session.rs b/crates/bugsee-core/src/session.rs index b4751b0..b92515a 100644 --- a/crates/bugsee-core/src/session.rs +++ b/crates/bugsee-core/src/session.rs @@ -27,6 +27,13 @@ const SESSIONS_DIR: &str = "sessions"; pub struct Session { data_dir: PathBuf, generation: u64, + /// The OS lock on `.lock`, held for the session's life. Peers sharing the + /// data dir (including from another pid namespace) probe it to tell a live + /// session from a crashed one; the kernel drops it when this process dies. + /// `None` when the session has no lock file (the filesystem rejects locks, or + /// the file could not be locked), in which case the pid in the marker is all + /// a peer has and `.lock` is deliberately absent. + _lock: Option, } impl Session { @@ -38,7 +45,7 @@ impl Session { std::fs::create_dir_all(&markers)?; let mut candidate = floor_generation(data_dir) + 1; - let generation = loop { + let (generation, lock) = loop { let marker = markers.join(format!("{candidate}.alive")); match std::fs::OpenOptions::new() .write(true) @@ -46,8 +53,22 @@ impl Session { .open(&marker) { Ok(mut f) => { + // Lock BEFORE writing the pid: the marker is empty (and fresh, + // so peers skip it) until then. + let lock = Session::take_lock(data_dir, candidate); + if lock.is_none() { + // No kernel lock to tell peers we are alive: a heartbeat + // file, touched by the worker, is the substitute. + let _ = std::fs::write( + crate::recovery::heartbeat_path(&Session::marker_path( + data_dir, candidate, + )), + b"", + ); + } + // The marker stays a bare pid, readable by older builds. let _ = write!(f, "{}", std::process::id()); - break candidate; + break (candidate, lock); } Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => { candidate += 1; @@ -62,9 +83,42 @@ impl Session { Ok(Session { data_dir: data_dir.to_path_buf(), generation, + _lock: lock, }) } + /// Take `.lock`, or leave no lock file at all. + /// + /// A peer's probe never takes the lock (unix), but on Windows it momentarily + /// does, so a contended claim is retried briefly. If it still cannot be held + /// with kernel enforcement the file is removed: a lock file that exists but + /// is not held reads as "that session died" to every peer, which would have + /// them recover (and delete) this live session. Without the file peers fall + /// back to the pid in the marker. + fn take_lock(data_dir: &Path, generation: u64) -> Option { + let path = Session::lock_path(data_dir, generation); + for attempt in 0..20 { + match crate::claim::try_claim(&path) { + Some(claim) if claim.is_os_locked() => return Some(claim), + // Thread-only (the filesystem rejects locks): useless to peers. + Some(_) => break, + None if attempt < 19 => std::thread::sleep(std::time::Duration::from_millis(2)), + None => break, + } + } + let _ = std::fs::remove_file(&path); + None + } + + /// Keep this session's lock until the process exits instead of releasing it + /// when the session is dropped. For an unclean shutdown, where a detached + /// worker or uploader may still be writing this session's data: releasing the + /// lock would let a peer recover the session as a crash and delete its parts + /// underneath them. The kernel drops the lock when the process ends. + pub(crate) fn keep_lock_until_exit(&mut self) { + std::mem::forget(self._lock.take()); + } + /// This session's generation number. pub fn generation(&self) -> u64 { self.generation @@ -77,9 +131,22 @@ impl Session { .join(format!("{generation}.alive")) } + /// The liveness lock-file path for `generation`. Its presence marks a session + /// whose liveness is arbitrated by the lock rather than the marker's pid. + pub(crate) fn lock_path(data_dir: &Path, generation: u64) -> PathBuf { + data_dir + .join(SESSIONS_DIR) + .join(format!("{generation}.lock")) + } + /// Mark a clean shutdown: remove this session's liveness marker. pub fn end(&self) { let _ = std::fs::remove_file(Session::marker_path(&self.data_dir, self.generation)); + let _ = std::fs::remove_file(Session::lock_path(&self.data_dir, self.generation)); + let _ = std::fs::remove_file(crate::recovery::heartbeat_path(&Session::marker_path( + &self.data_dir, + self.generation, + ))); } } @@ -121,3 +188,67 @@ fn max_numeric_entry(dir: &Path, strip_suffix: Option<&str>) -> u64 { } max } + +#[cfg(test)] +mod tests { + use super::*; + use crate::claim::{probe, Probe}; + + fn tmp() -> PathBuf { + let p = std::env::temp_dir().join(format!("bugsee-sess-{}", crate::util::random_hex(8))); + std::fs::create_dir_all(&p).unwrap(); + p + } + + #[test] + fn a_live_session_holds_its_lock_and_a_clean_end_removes_it() { + let dir = tmp(); + let session = Session::begin(&dir).unwrap(); + let lock = Session::lock_path(&dir, session.generation()); + assert!(matches!(probe(&lock), Probe::Held)); + session.end(); + assert!(!lock.exists()); + } + + #[test] + fn a_session_that_cannot_hold_its_lock_leaves_no_lock_file_behind() { + // A lock file that exists but is not held reads as "that session died". + let dir = tmp(); + std::fs::create_dir_all(dir.join(SESSIONS_DIR)).unwrap(); + let peer = crate::claim::try_claim(&Session::lock_path(&dir, 1)).unwrap(); + let session = Session::begin(&dir).unwrap(); + assert_eq!(session.generation(), 1); + assert!(session._lock.is_none()); + drop(peer); + assert!(!Session::lock_path(&dir, 1).exists()); + } + + #[test] + fn a_session_on_a_filesystem_without_locks_leaves_no_lock_file() { + let dir = tmp(); + crate::claim::FORCE_UNSUPPORTED.with(|f| f.set(true)); + let session = Session::begin(&dir).unwrap(); + crate::claim::FORCE_UNSUPPORTED.with(|f| f.set(false)); + assert!(session._lock.is_none()); + assert!(!Session::lock_path(&dir, session.generation()).exists()); + // Instead it keeps a heartbeat file, removed on a clean end. + let beat = + crate::recovery::heartbeat_path(&Session::marker_path(&dir, session.generation())); + assert!(beat.exists()); + session.end(); + assert!(!beat.exists()); + } + + #[test] + fn keeping_the_lock_until_exit_survives_dropping_the_session() { + let dir = tmp(); + let mut session = Session::begin(&dir).unwrap(); + let lock = Session::lock_path(&dir, session.generation()); + session.keep_lock_until_exit(); + drop(session); + assert!( + matches!(probe(&lock), Probe::Held), + "a detached thread may still be writing: peers must not recover it" + ); + } +} diff --git a/crates/bugsee-core/tests/multiprocess.rs b/crates/bugsee-core/tests/multiprocess.rs new file mode 100644 index 0000000..5b26c83 --- /dev/null +++ b/crates/bugsee-core/tests/multiprocess.rs @@ -0,0 +1,353 @@ +// +// multiprocess.rs +// bugsee-core +// +// Copyright © 2026 Bugsee. All rights reserved. +// +//! Several processes sharing one data directory (a prefork server's workers, a CLI +//! run beside a daemon): every pending report is delivered exactly once, and a +//! holder that dies mid-work never blocks the others (issue #14). +//! +//! The "processes" are real: this test binary re-executes itself in a child role +//! selected by environment variables. + +use std::io::Write; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use bugsee_core::transport::{Transport, TransportError}; +use bugsee_core::{queue, recovery, AssembledReport, Recorder, RecorderConfig}; + +const ROLE: &str = "BUGSEE_MP_ROLE"; +const DIR: &str = "BUGSEE_MP_DIR"; +const LOG: &str = "BUGSEE_MP_LOG"; + +/// A transport that records each upload by appending a line to a shared log file +/// (one `write` under `O_APPEND`, atomic across processes). +struct LogTransport { + log: PathBuf, + lock: Mutex<()>, +} +impl Transport for LogTransport { + fn register_session(&self, _: &str, _: &[u8]) -> Result { + Ok("t".into()) + } + fn create_issue(&self, _: &str, _: Option<&str>, _: &[u8]) -> Result { + Ok("endpoint".into()) + } + fn upload_bundle(&self, _: &str, zip: &[u8]) -> Result<(), TransportError> { + let _g = self.lock.lock().unwrap(); + // Give a racing peer a real window to double-deliver. + std::thread::sleep(Duration::from_millis(20)); + let mut f = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(&self.log) + .unwrap(); + // Queued test items carry a readable name; recovered crashes are binary + // ZIPs, logged as a tag (each recovered session yields one). + let id = match std::str::from_utf8(zip) { + Ok(text) if text.starts_with("queued-") => text.to_string(), + _ => "recovered-crash".to_string(), + }; + // One `write` call: `writeln!` would split text and newline. + f.write_all(format!("{id}\n").as_bytes()).unwrap(); + Ok(()) + } +} + +/// Child entry point. A no-op in a normal test run. +#[test] +fn child_entry() { + let Ok(role) = std::env::var(ROLE) else { + return; + }; + let dir = PathBuf::from(std::env::var(DIR).unwrap()); + match role.as_str() { + "worker" => { + // Spin until the parent releases every worker at once. + let go = dir.join("GO"); + while !go.exists() { + std::thread::sleep(Duration::from_millis(1)); + } + let transport = Arc::new(LogTransport { + log: PathBuf::from(std::env::var(LOG).unwrap()), + lock: Mutex::new(()), + }); + let rec = Recorder::launch(RecorderConfig::new(&dir, "tok"), transport).unwrap(); + rec.flush(Duration::from_secs(3)); + } + "holder" => { + // Hold every queue item and every recoverable session, then wait to + // be killed. + let mut held = Vec::new(); + for q in queue::list_pending(&dir) { + held.push(queue::try_claim(&q).expect("claim queue item")); + } + for p in recovery::find_pending(&dir, u64::MAX) { + held.push( + recovery::claim_pending(&dir, &p) + .expect("claim session") + .expect("a real claim"), + ); + } + // Signal readiness through the filesystem: libtest owns stdout. + std::fs::write(dir.join("READY.tmp"), held.len().to_string()).unwrap(); + std::fs::rename(dir.join("READY.tmp"), dir.join("READY")).unwrap(); + std::thread::sleep(Duration::from_secs(120)); + } + "session" => { + // A live session of its own, as a peer's launch would create. + let session = bugsee_core::session::Session::begin(&dir).unwrap(); + std::fs::write(dir.join("READY.tmp"), session.generation().to_string()).unwrap(); + std::fs::rename(dir.join("READY.tmp"), dir.join("READY")).unwrap(); + std::thread::sleep(Duration::from_secs(120)); + } + "lockholder" => { + // Hold `sessions/1.lock`, as a peer's probe or launch would for a moment. + let claim = bugsee_core::claim::try_claim(&dir.join("sessions").join("1.lock")); + std::fs::write(dir.join("READY.tmp"), claim.is_some().to_string()).unwrap(); + std::fs::rename(dir.join("READY.tmp"), dir.join("READY")).unwrap(); + std::thread::sleep(Duration::from_secs(120)); + } + other => panic!("unknown role {other}"), + } +} + +struct TempDir(PathBuf); +impl TempDir { + fn new() -> Self { + let p = + std::env::temp_dir().join(format!("bugsee-mp-{}", bugsee_core::util::random_hex(8))); + std::fs::create_dir_all(&p).unwrap(); + TempDir(p) + } +} +impl Drop for TempDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +fn spawn(role: &str, dir: &Path, log: &Path) -> Child { + Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "child_entry", "--nocapture"]) + .env(ROLE, role) + .env(DIR, dir) + .env(LOG, log) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap() +} + +/// `queued` reports in the queue and `crashed` dead sessions awaiting recovery. +fn seed(dir: &Path, queued: usize, crashed: u64) { + for i in 0..queued { + queue::enqueue( + dir, + &AssembledReport { + bundle_name: format!("{}.bundle.zip", bugsee_core::util::random_hex(10)), + zip: format!("queued-{i}").into_bytes(), + request_json: br#"{"signatures":[]}"#.to_vec(), + }, + ) + .unwrap(); + } + let sessions = dir.join("sessions"); + std::fs::create_dir_all(&sessions).unwrap(); + for g in 1..=crashed { + // "0" is the portable "owner gone" sentinel. + std::fs::write(sessions.join(format!("{g}.alive")), "0").unwrap(); + // The lock file a crashed session leaves behind: present, held by no one. + std::fs::write(sessions.join(format!("{g}.lock")), "").unwrap(); + std::fs::create_dir_all(dir.join("parts").join(g.to_string())).unwrap(); + } + std::fs::write(dir.join("gen"), crashed.to_string()).unwrap(); +} + +fn delivered(log: &Path) -> Vec { + std::fs::read_to_string(log) + .map(|s| s.lines().map(String::from).collect()) + .unwrap_or_default() +} + +/// Every file under `dir`, relative, for failure messages. +fn tree(dir: &Path) -> Vec { + let mut out = Vec::new(); + let mut stack = vec![dir.to_path_buf()]; + while let Some(d) = stack.pop() { + for e in std::fs::read_dir(&d).into_iter().flatten().flatten() { + let p = e.path(); + if p.is_dir() { + stack.push(p); + } else { + out.push(p.strip_prefix(dir).unwrap().display().to_string()); + } + } + } + out.sort(); + out +} + +fn run_workers(dir: &Path, log: &Path, n: usize) { + let kids: Vec = (0..n).map(|_| spawn("worker", dir, log)).collect(); + std::thread::sleep(Duration::from_millis(300)); // all spinning on GO + std::fs::write(dir.join("GO"), "").unwrap(); + for mut k in kids { + let _ = k.wait(); + } +} + +#[test] +fn racing_launchers_deliver_each_pending_report_exactly_once() { + // Repeated: the double-delivery windows are narrow, so one round can pass by luck. + for round in 0..8 { + let tmp = TempDir::new(); + let dir = tmp.0.join("data"); + std::fs::create_dir_all(&dir).unwrap(); + let log = tmp.0.join("delivered.log"); + seed(&dir, 6, 4); + + run_workers(&dir, &log, 5); + + let got = delivered(&log); + // 6 queued reports + 4 recovered crashes, no more, no fewer. + assert_eq!( + got.len(), + 10, + "round {round}, delivery log: {got:?}, left on disk: {:?}", + tree(&dir) + ); + for i in 0..6 { + let n = got.iter().filter(|l| **l == format!("queued-{i}")).count(); + assert_eq!(n, 1, "round {round}: queued-{i} delivered {n} times"); + } + assert_eq!(got.iter().filter(|l| *l == "recovered-crash").count(), 4); + assert!(queue::list_pending(&dir).is_empty()); + } +} + +#[test] +fn a_holder_killed_mid_work_does_not_block_the_others() { + let tmp = TempDir::new(); + let dir = tmp.0.join("data"); + std::fs::create_dir_all(&dir).unwrap(); + let log = tmp.0.join("delivered.log"); + seed(&dir, 4, 3); + + let mut holder = spawn("holder", &dir, &log); + let ready = dir.join("READY"); + let deadline = Instant::now() + Duration::from_secs(20); + while !ready.exists() { + assert!(Instant::now() < deadline, "holder never reported ready"); + std::thread::sleep(Duration::from_millis(10)); + } + assert_eq!(std::fs::read_to_string(&ready).unwrap(), "7"); + + // While the holder lives, nothing it holds is delivered by anyone else. + run_workers(&dir, &log, 2); + assert!(delivered(&log).is_empty(), "live holder's work was taken"); + std::fs::remove_file(dir.join("GO")).unwrap(); + + // SIGKILL / TerminateProcess: no destructor runs, its claim files survive. + holder.kill().unwrap(); + let _ = holder.wait(); + + let deadline = Instant::now() + Duration::from_secs(20); + run_workers(&dir, &log, 3); + while delivered(&log).len() < 7 && Instant::now() < deadline { + run_workers(&dir, &log, 1); + } + assert_eq!(delivered(&log).len(), 7, "log: {:?}", delivered(&log)); +} + +/// Containers with separate pid namespaces share a volume, and each app is pid 1, +/// so a peer's recorded pid can name a live process HERE that is not the peer. +/// The marker's pid must therefore play no part in deciding the peer is dead. +#[test] +fn a_live_peers_session_is_not_mistaken_for_a_crash_whatever_pid_its_marker_names() { + let tmp = TempDir::new(); + let dir = tmp.0.join("data"); + std::fs::create_dir_all(&dir).unwrap(); + let log = tmp.0.join("delivered.log"); + + let mut peer = spawn("session", &dir, &log); + let ready = dir.join("READY"); + let deadline = Instant::now() + Duration::from_secs(20); + while !ready.exists() { + assert!(Instant::now() < deadline, "peer never started"); + std::thread::sleep(Duration::from_millis(10)); + } + let generation: u64 = std::fs::read_to_string(&ready).unwrap().parse().unwrap(); + + // Make the marker name a pid that is alive in THIS namespace and is not the + // peer: the parent's own, then init's. Neither may make the peer look dead, + // nor make it look alive once it is gone. + let marker = dir.join("sessions").join(format!("{generation}.alive")); + // A reaped child's pid: gone from this namespace, as a peer's own pid is. + let mut reaped = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "no_such_test"]) + .stdout(Stdio::null()) + .spawn() + .unwrap(); + let _ = reaped.wait(); + let foreign_pids = [ + std::process::id().to_string(), + "1".to_string(), + "4294967290".to_string(), + reaped.id().to_string(), + ]; + for foreign in foreign_pids { + std::fs::write(&marker, &foreign).unwrap(); + run_workers(&dir, &log, 3); + let _ = std::fs::remove_file(dir.join("GO")); + assert!(delivered(&log).is_empty(), "live peer reported as a crash"); + assert!(marker.exists(), "live peer's marker was reclaimed"); + } + + // Once the peer really dies, its session is recovered exactly once, even + // though the marker still names a live pid. + peer.kill().unwrap(); + let _ = peer.wait(); + run_workers(&dir, &log, 3); + assert_eq!(delivered(&log), vec!["recovered-crash".to_string()]); +} + +/// A launch that finds `.lock` momentarily held by someone else must not leave +/// an unheld lock file behind: every peer would read it as "that session died" and +/// recover (and delete) the live session. +#[test] +fn a_launch_that_loses_the_lock_race_is_not_recovered_as_a_crash() { + let tmp = TempDir::new(); + let dir = tmp.0.join("data"); + std::fs::create_dir_all(dir.join("sessions")).unwrap(); + let log = tmp.0.join("delivered.log"); + + let mut holder = spawn("lockholder", &dir, &log); + let ready = dir.join("READY"); + let deadline = Instant::now() + Duration::from_secs(20); + while !ready.exists() { + assert!(Instant::now() < deadline, "holder never started"); + std::thread::sleep(Duration::from_millis(10)); + } + assert_eq!(std::fs::read_to_string(&ready).unwrap(), "true"); + std::fs::remove_file(&ready).unwrap(); + + // Generation 1 is ours, but its lock file is held by the other process. + let session = bugsee_core::session::Session::begin(&dir).unwrap(); + assert_eq!(session.generation(), 1); + holder.kill().unwrap(); + let _ = holder.wait(); + + // Peers now look at a live session (this process) with no usable lock file. + run_workers(&dir, &log, 3); + assert!( + delivered(&log).is_empty(), + "live session reported as a crash" + ); + assert!(dir.join("sessions").join("1.alive").exists()); + drop(session); +}