diff --git a/DESIGN.md b/DESIGN.md index 0513278..74c6955 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -302,7 +302,7 @@ That last case previously produced **no report whatsoever** — the SDK shut dow ## 12. Crash-time safety, best-effort flush & edge cases -**Best-effort in-flight flush:** at crash, secure the guaranteed artifact first (`PanicSnapshot` / minidump), then attempt flush, then terminate. The crashing thread never touches the queue lock/allocator — it does an async-signal-safe wake (atomic flag + `sem_post`/eventfd/self-pipe) and waits ≤~50 ms for the worker to drain→append→`write_all`→ack; no ack → give up. (This best-effort in-flight flush is a **planned** hardening step; the current worker already durably appends each record as it drains, so completed records survive without it.) Panic(unwind): usually succeeds. panic=abort: hook wakes worker before abort. Native fault: defensive only. +**Best-effort in-flight flush (implemented, `bugsee_core::crash_flush`):** a producer only queues; the worker appends later, so entries still in the channel (or mid-append) when the process dies are lost. At crash the handler first secures the guaranteed artifact (marker + frames), then does one atomic ticket increment, one non-blocking `write(2)` (Windows: `SetEvent`) to a process-wide wake handle (one pipe/event per process, never closed, so no descriptor-reuse race), and polls an atomic ack for at most 50 ms (`crash_flush::DEFAULT_BUDGET`, shared with the panic hook). It first raises a `dying` flag so the uploader and recovery start no new claims in the window (a kill mid-claim would strand it for the claim grace period). One helper thread per process (created at launch, re-created in a fork child, never in the handler) is blocked on that handle; it posts a `Flush` message through the normal FIFO channel and, when the worker acks, publishes the ticket. FIFO order means an ack implies every entry queued before the crash is appended. The crashing thread never takes the queue lock, allocates or serializes, and a stuck/dead worker or poisoned allocator costs exactly the budget (the 5 s watchdog is not involved). The pid recorded at registration makes a fork child skip the flush until its revived recorder re-registers. A fault on the worker or helper thread skips the wait; the last registered recorder is the flush target. Panic: the hook runs the same bounded flush for a main-thread panic no guard will catch (covers `panic = "abort"` builds with no native handler; with one, the SIGABRT handler flushes too); an unwinding panic is also drained by the launch guard's `Drop`. Native fault: flushed from every fatal exit of the signal handler / the Windows exception filter, after the marker is complete. **Tradeoffs:** zero hot-path cost and no new on-disk format (no spill file: a torn append is the already-tolerated torn part tail, pinned by a recovery test), but the flush needs a live worker and allocator, so a fault that corrupts the heap or hits the worker thread forfeits it, and what is drained is bounded by throughput (about 3.5k ~450-byte events in 50 ms on a debug build; the remainder is lost). A producer-side pre-serialized ring spilled by the handler would survive a dead worker but taxes every event; not built. **Recursion guard:** `NORMAL→PANICKING→HANDLING_FATAL→SECONDARY`; second fatal → tiny marker + terminate, no re-entry. **Quarantine:** Class A (continue) / B (restart) / C (rebuild instance) / D (disable); restart budget (1 immediate, 3/10 min, then disable) — never infinite. diff --git a/PROGRESS.md b/PROGRESS.md index 776aaab..746cd3d 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -33,7 +33,7 @@ in [`DESIGN.md`](./DESIGN.md); the wire contract is `bugsee/report-bundle-struct | 0 — Harness | ✅ | Workspace, mock 3-step API server, bundle validator, subprocess crash runner, CI **matrix across ubuntu/macOS/windows** (build/test/clippy per platform, fmt once) + pinned-MSRV job. | | 1 — Ingestible bundle (core) | ✅ | Sliding-window capture (time/byte/event caps), length-prefixed part streams, envelope/manifest, Zstd method-93 ZIP, signatures. | | 2 — Handled errors + caught panics | ✅ | `catch_unwind` guards, panic snapshot, error/cause chains. Follow-ups: `tracing` Layer + `log` adapter done. Subsystem quarantine still open. | -| 3 — Fatal + next-launch + correlation | ✅ (core) | Native handler, session generations + liveness markers, next-launch recovery, panic↔SIGABRT correlation, client-side native dedup signature from frame offsets. **New:** Windows native capture — SEH/VEH handler, PE CodeView module map, marker + frame capture — so all three desktop platforms report fatal crashes. Open: best-effort in-flight flush, full out-of-process minidump. | +| 3 — Fatal + next-launch + correlation | ✅ (core) | Native handler, session generations + liveness markers, next-launch recovery, panic↔SIGABRT correlation, client-side native dedup signature from frame offsets. **New:** Windows native capture — SEH/VEH handler, PE CodeView module map, marker + frame capture — so all three desktop platforms report fatal crashes. In-flight events are now flushed at crash (bounded 50 ms, signal-safe wake of the capture worker; #13). Open: full out-of-process minidump. | | 4 — Integrations & APM | ✅ (core) | APM transactions/spans, sysinfo telemetry, reqwest network capture. **New:** the environment is actually populated — OS version, kernel, memory, disk, locale, and app identity (`package_id`/`version`/`build`) via `LaunchOptions::app_version`. Open: sessions / release-health. | | 5 — Hardening | ✅ (core) | Durable retry queue (backoff, retry cap, blacklist), `before_send`/`before_breadcrumb`, event sampling. **New:** `on_report_dropped` — delivery failure is observable at last; `flush()` alone never could distinguish delivered from discarded. Open: general PII scrubber, rate limits. | | 6 — Mobile/FFI | ✅ (core) | `bugsee-ffi` C ABI (`include/bugsee.h`). Open: actual iOS/Android target builds + Swift/Kotlin wrappers. | @@ -538,7 +538,7 @@ Rust and are worth evaluating before building one. ### Remaining phase work - Phase 2: subsystem quarantine (auto-disable a repeatedly-faulting integration). -- Phase 3: best-effort in-flight flush at crash; full out-of-process minidump (desktop) — see the backend note below, which is the same machinery. +- Phase 3: full out-of-process minidump (desktop) — see the backend note below, which is the same machinery. - Phase 4: sessions / release-health. - Phase 5: general PII scrubber, rate limits. - Phase 6: iOS/Android target builds; Swift/Kotlin language wrappers. diff --git a/crates/bugsee-core/src/claim.rs b/crates/bugsee-core/src/claim.rs index f3d0956..3ebcb4b 100644 --- a/crates/bugsee-core/src/claim.rs +++ b/crates/bugsee-core/src/claim.rs @@ -87,6 +87,10 @@ impl Drop for Claim { /// 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 { + // A dying process must not start work a kill would leave half-claimed. + if crate::crash_flush::is_dying() { + return None; + } match acquire(path) { Acquire::Got(claim) => Some(claim), Acquire::Held | Acquire::Unavailable => None, @@ -100,6 +104,9 @@ pub fn try_claim(path: &Path) -> Option { /// may both do the work, and the backend deduplicates. `None` means another /// process (or thread) holds the claim. pub fn try_claim_lenient(path: &Path) -> Option> { + if crate::crash_flush::is_dying() { + return None; + } match acquire(path) { Acquire::Got(claim) => Some(Some(claim)), Acquire::Unavailable => Some(None), diff --git a/crates/bugsee-core/src/crash_flush.rs b/crates/bugsee-core/src/crash_flush.rs new file mode 100644 index 0000000..afd9d78 --- /dev/null +++ b/crates/bugsee-core/src/crash_flush.rs @@ -0,0 +1,498 @@ +// +// crash_flush.rs +// bugsee-core +// +// Copyright © 2026 Bugsee. All rights reserved. +// + +//! Best-effort flush of in-flight capture entries at crash time (DESIGN §12). +//! +//! # What is lost without this +//! +//! A producer thread never writes to disk: it pushes the entry into the capture +//! worker's channel and returns. The worker serializes and appends it to the +//! part file a little later. Anything still in the channel (or in the worker's +//! hands) when the process dies is gone, and those are exactly the last +//! breadcrumbs, log lines and network events before the fault. The window is +//! microseconds when idle, but grows to many milliseconds under a burst or when +//! the worker is descheduled. +//! +//! # What this does +//! +//! A fatal-crash handler cannot take the channel lock, allocate, or serialize +//! (the faulting thread may hold the allocator or the lock). So it does not +//! touch the queue at all. It performs one `write(2)` / `SetEvent` on a +//! process-wide wake handle created at first launch, then waits for an atomic +//! acknowledgement for a bounded time: +//! +//! 1. one helper thread per process (created at launch, long before any crash; +//! re-created in a `fork()` child) is blocked on that handle; +//! 2. woken, it calls the registered flush target, which posts a `Flush` marker +//! to the capture worker through the normal channel. The channel is FIFO, so +//! when the worker acknowledges the marker every entry queued before the +//! crash has been serialized and appended; +//! 3. the helper publishes the acknowledgement in an atomic, which the waiting +//! handler observes before letting the process die. +//! +//! The handler's side ([`flush_for_crash`]) uses only atomics, `getpid`, one +//! non-blocking `write`, `clock_gettime` and `poll` (Windows: `SetEvent`, +//! `WaitForSingleObject`, `QueryPerformanceCounter`), all async-signal-safe: no +//! allocation, no lock. The wait is capped by the caller's budget, so a stuck or +//! dead worker (or a corrupted heap that blocks the helper in the allocator) +//! costs the budget and no more. +//! +//! The wake handles are **never closed** (one pipe or event pair per process, +//! not per recorder), so a handler racing a recorder's shutdown can never write +//! to a recycled descriptor, and the read end outlives every writer, so a C host +//! (through `bugsee-ffi`) never sees `SIGPIPE`. +//! +//! # The dying flag +//! +//! [`flush_for_crash`] first raises a process-wide flag. During the (up to +//! budget) window the process is still running other threads, and an uploader +//! that started a report now would be killed halfway through a claim. While the +//! flag is up the runtime starts no claims and no recoveries ([`is_dying`]). +//! +//! # Limits (deliberate, documented tradeoffs) +//! +//! * Costs nothing on the happy path (no extra copy or serialization on the +//! producer) but depends on the worker and allocator being alive. A fault that +//! corrupts the heap, or kills the worker thread itself, forfeits the flush. +//! A fault on the worker or the helper thread skips the wait entirely. +//! * Entries are flushed to the OS page cache like every other append; this +//! survives the process dying, not the machine. +//! * **Last registered wins.** The flush target is process-wide: a second +//! `Recorder` launched in the same process takes over, and dropping a recorder +//! that is not the current target does nothing. +//! * A fork child has none of the parent's threads: the pid recorded at +//! registration makes the handler skip the flush there until the child's +//! revived recorder registers again (which re-creates the pipe and helper). +//! * **Windows ordering:** the flush runs inside the unhandled-exception filter, +//! before the host's previously installed filter (if any) is called, so a host +//! reporter that runs afterwards sees the process as it was. +//! * A panic hook that calls [`flush_in_flight`] for a main-thread panic a host +//! goes on to catch outside our guards pays up to the budget once for it. + +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use crate::fork::ForkMutex; + +/// How long a dying process waits for the capture worker by default. +pub const DEFAULT_BUDGET: Duration = Duration::from_millis(50); + +/// Flush requests issued by crash handlers (a ticket counter). +static REQUESTED: AtomicU64 = AtomicU64::new(0); +/// Highest ticket the helper has seen acknowledged by the worker. +static ACKED: AtomicU64 = AtomicU64::new(0); +/// Raised by a fatal-crash handler: the process is about to die. +static DYING: AtomicBool = AtomicBool::new(false); + +/// Threads whose own crash must not wait for the flush (the worker and helper: +/// they cannot answer while handling a fault). +const PIPELINE_SLOTS: usize = 8; +static PIPELINE_THREADS: [AtomicUsize; PIPELINE_SLOTS] = + [const { AtomicUsize::new(0) }; PIPELINE_SLOTS]; + +type Target = Arc bool + Send + Sync>; + +struct State { + target: Option<(u64, Target)>, + next_id: u64, + /// Process that created the helper and pipe (`0`: none yet). A different + /// value means we are in a `fork()` child, where they do not exist. + helper_pid: u32, +} + +static STATE: ForkMutex = ForkMutex::new(|| State { + target: None, + next_id: 1, + helper_pid: 0, +}); + +/// The registered flush target. Dropping it deregisters (if still current). +pub struct CrashFlush { + id: u64, +} + +impl CrashFlush { + /// Make `flush` the process's crash-time flush target, starting the helper + /// thread and wake handle if this process has none yet. + /// + /// `flush` is called on the helper thread each time a crash handler asks, and + /// must return `true` once everything queued before the call is on disk. It + /// must bound its own waiting. + pub fn install( + flush: impl Fn() -> bool + Send + Sync + 'static, + ) -> std::io::Result { + let mut st = STATE.lock(); + let pid = std::process::id(); + if st.helper_pid != pid { + imp::start_helper()?; + st.helper_pid = pid; + } + let id = st.next_id; + st.next_id += 1; + st.target = Some((id, Arc::new(flush))); + Ok(CrashFlush { id }) + } +} + +impl Drop for CrashFlush { + fn drop(&mut self) { + let mut st = STATE.lock(); + if st.target.as_ref().is_some_and(|(id, _)| *id == self.id) { + st.target = None; + } + } +} + +/// The current flush target, for the helper thread. +fn current_target() -> Option { + STATE.lock().target.as_ref().map(|(_, t)| Arc::clone(t)) +} + +/// Marks the current thread as part of the capture pipeline for its lifetime. +pub struct PipelineThread(usize); + +/// Record that the calling thread is a pipeline thread (see [`PIPELINE_SLOTS`]). +pub fn register_pipeline_thread() -> PipelineThread { + let me = imp::thread_id(); + for slot in &PIPELINE_THREADS { + if slot + .compare_exchange(0, me, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + { + return PipelineThread(me); + } + } + PipelineThread(0) +} + +impl Drop for PipelineThread { + fn drop(&mut self) { + if self.0 == 0 { + return; + } + for slot in &PIPELINE_THREADS { + let _ = slot.compare_exchange(self.0, 0, Ordering::AcqRel, Ordering::Acquire); + } + } +} + +fn on_pipeline_thread() -> bool { + let me = imp::thread_id(); + PIPELINE_THREADS + .iter() + .any(|s| s.load(Ordering::Acquire) == me) +} + +/// Has a fatal-crash handler declared the process about to die? The runtime +/// starts no report claims or recoveries while this is set. +pub fn is_dying() -> bool { + DYING.load(Ordering::Acquire) +} + +/// Clear the dying flag. Tests only. +#[doc(hidden)] +pub fn reset_dying_for_test() { + DYING.store(false, Ordering::Release); +} + +/// For a fatal-crash handler: declare the process dying, then ask the registered +/// capture pipeline to drain everything queued so far, waiting at most `budget`. +/// Returns whether it confirmed. +/// +/// **Async-signal-safe**: callable from a signal handler or exception filter. +/// No allocation, no locks, bounded time. +pub fn flush_for_crash(budget: Duration) -> bool { + DYING.store(true, Ordering::Release); + flush_in_flight(budget) +} + +/// Like [`flush_for_crash`] without declaring the process dying: for a caller +/// that only *expects* to die (a main-thread panic hook). Returns immediately +/// when no pipeline is registered, the registration belongs to another process +/// (a `fork()` child), or the caller is itself a pipeline thread. +pub fn flush_in_flight(budget: Duration) -> bool { + if on_pipeline_thread() { + return false; + } + imp::request_and_wait(budget) +} + +/// Drain: bump the ack to `target` after a successful flush. +fn ack(target: u64) { + ACKED.fetch_max(target, Ordering::AcqRel); +} + +/// What the helper does each time it is woken. +fn serve_request() { + let target = REQUESTED.load(Ordering::Acquire); + if target <= ACKED.load(Ordering::Acquire) { + return; + } + let ok = match current_target() { + Some(flush) => { + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| flush())).unwrap_or(false) + } + // Nothing registered: no recorder is accepting entries (a recorder stays + // registered until its worker has stopped), so there is nothing to drain. + None => true, + }; + if ok { + ack(target); + imp::signal_ack(); + } +} + +#[cfg(unix)] +mod imp { + use super::{serve_request, ACKED, PIPELINE_THREADS, REQUESTED}; + use std::sync::atomic::{AtomicI32, Ordering}; + use std::time::Duration; + + /// Write end of the wake pipe (`-1` when none). Never closed once set. + static WAKE_FD: AtomicI32 = AtomicI32::new(-1); + /// Read end, kept only so a fork child can close the parent's copy. + static READ_FD: AtomicI32 = AtomicI32::new(-1); + /// Pid that owns `WAKE_FD` (a fork child must not use it). + static OWNER_PID: AtomicI32 = AtomicI32::new(0); + + pub fn thread_id() -> usize { + // SAFETY: pthread_self is async-signal-safe and has no preconditions. + unsafe { libc::pthread_self() as usize } + } + + fn make_pipe() -> std::io::Result<(i32, i32)> { + let mut fds = [0i32; 2]; + #[cfg(any(target_os = "linux", target_os = "android"))] + // SAFETY: `fds` is a valid two-element array. + let rc = unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC) }; + #[cfg(not(any(target_os = "linux", target_os = "android")))] + // SAFETY: as above. + let rc = unsafe { libc::pipe(fds.as_mut_ptr()) }; + if rc != 0 { + return Err(std::io::Error::last_os_error()); + } + // SAFETY: both descriptors were just created and are ours. + unsafe { + #[cfg(not(any(target_os = "linux", target_os = "android")))] + for fd in fds { + libc::fcntl(fd, libc::F_SETFD, libc::FD_CLOEXEC); + } + // The handler must never block on a full pipe. + let fl = libc::fcntl(fds[1], libc::F_GETFL); + libc::fcntl(fds[1], libc::F_SETFL, fl | libc::O_NONBLOCK); + } + Ok((fds[0], fds[1])) + } + + /// Create the pipe and the one helper thread for this process. In a fork + /// child this replaces the parent's (which are closed here, after the new + /// ones are published). + pub fn start_helper() -> std::io::Result<()> { + // Unpublish first, so a handler that runs mid-switch sees nothing. + let old_w = WAKE_FD.swap(-1, Ordering::AcqRel); + let old_r = READ_FD.swap(-1, Ordering::AcqRel); + // SAFETY: getpid has no preconditions. + OWNER_PID.store(unsafe { libc::getpid() }, Ordering::Release); + for slot in &PIPELINE_THREADS { + slot.store(0, Ordering::Release); + } + // Tickets of the parent's requests are meaningless here. + ACKED.store(REQUESTED.load(Ordering::Acquire), Ordering::Release); + + let (r, w) = make_pipe()?; + let spawned = std::thread::Builder::new() + .name("bugsee-crash-flush".into()) + .spawn(move || { + let _me = super::register_pipeline_thread(); + let mut buf = [0u8; 64]; + loop { + // SAFETY: reads into a valid buffer from a descriptor that is + // never closed while this thread runs. + let n = unsafe { libc::read(r, buf.as_mut_ptr().cast(), buf.len()) }; + if n > 0 { + serve_request(); + } else if n < 0 + && std::io::Error::last_os_error().kind() == std::io::ErrorKind::Interrupted + { + continue; + } else { + break; + } + } + }); + if let Err(e) = spawned { + // SAFETY: closing descriptors made above and shared with nobody. + unsafe { + libc::close(r); + libc::close(w); + } + return Err(e); + } + READ_FD.store(r, Ordering::Release); + WAKE_FD.store(w, Ordering::Release); + // The parent's copies are of no use here. + for fd in [old_w, old_r] { + if fd >= 0 { + // SAFETY: inherited descriptors only this child's code could use, + // unpublished above. + unsafe { libc::close(fd) }; + } + } + Ok(()) + } + + pub fn signal_ack() {} + + fn now_ns() -> u64 { + let mut ts = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + // SAFETY: clock_gettime is async-signal-safe and `ts` is valid. + unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut ts) }; + (ts.tv_sec as u64) + .saturating_mul(1_000_000_000) + .saturating_add(ts.tv_nsec as u64) + } + + pub fn request_and_wait(budget: Duration) -> bool { + let fd = WAKE_FD.load(Ordering::Acquire); + // SAFETY: getpid is async-signal-safe. + if fd < 0 || OWNER_PID.load(Ordering::Acquire) != unsafe { libc::getpid() } { + return false; + } + let ticket = REQUESTED.fetch_add(1, Ordering::AcqRel) + 1; + let b = 1u8; + // SAFETY: async-signal-safe write of one byte. A full pipe (EAGAIN) means + // wakeups are already pending, and the helper reads the ticket counter. + unsafe { libc::write(fd, (&b as *const u8).cast(), 1) }; + let deadline = now_ns().saturating_add(budget.as_nanos().min(u64::MAX as u128) as u64); + while ACKED.load(Ordering::Acquire) < ticket { + if now_ns() >= deadline { + return false; + } + // SAFETY: poll with no descriptors is a plain async-signal-safe sleep. + unsafe { libc::poll(core::ptr::null_mut(), 0, 1) }; + } + true + } +} + +#[cfg(windows)] +mod imp { + use super::{serve_request, ACKED, REQUESTED}; + use crate::win32::{ + CreateEventW, GetCurrentThreadId, Handle, QueryPerformanceCounter, + QueryPerformanceFrequency, SetEvent, WaitForSingleObject, WAIT_OBJECT_0, + }; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; + + /// Auto-reset events, never closed: the handler signals `WAKE`, the helper + /// signals `ACK`. (`0` when none.) + static WAKE: AtomicUsize = AtomicUsize::new(0); + static ACK: AtomicUsize = AtomicUsize::new(0); + + pub fn thread_id() -> usize { + // SAFETY: plain Win32 call. + unsafe { GetCurrentThreadId() as usize } + } + + pub fn start_helper() -> std::io::Result<()> { + // SAFETY: plain Win32 calls; auto-reset, initially unsignalled, unnamed. + let (wake, ack) = unsafe { + ( + CreateEventW(core::ptr::null(), 0, 0, core::ptr::null()), + CreateEventW(core::ptr::null(), 0, 0, core::ptr::null()), + ) + }; + if wake.is_null() || ack.is_null() { + return Err(std::io::Error::last_os_error()); + } + // Raw handles are not `Send`; carry them as integers. + let (wake, ack) = (wake as usize, ack as usize); + std::thread::Builder::new() + .name("bugsee-crash-flush".into()) + .spawn(move || { + let _me = super::register_pipeline_thread(); + loop { + // SAFETY: waits on an event that is never closed. + if unsafe { WaitForSingleObject(wake as Handle, u32::MAX) } != WAIT_OBJECT_0 { + break; + } + serve_request(); + } + })?; + ACK.store(ack, Ordering::Release); + WAKE.store(wake, Ordering::Release); + Ok(()) + } + + pub fn signal_ack() { + let h = ACK.load(Ordering::Acquire); + if h != 0 { + // SAFETY: signals a live event. + unsafe { SetEvent(h as Handle) }; + } + } + + /// `(counter, frequency)`; integer math only downstream (no floating point + /// in an exception filter). + fn now_ticks() -> (i64, i64) { + let (mut t, mut f) = (0i64, 1i64); + // SAFETY: valid out-pointers; both calls cannot fail on supported Windows. + unsafe { + QueryPerformanceCounter(&mut t); + QueryPerformanceFrequency(&mut f); + } + (t, f.max(1)) + } + + pub fn request_and_wait(budget: Duration) -> bool { + let wake = WAKE.load(Ordering::Acquire); + let ack = ACK.load(Ordering::Acquire); + if wake == 0 || ack == 0 { + return false; + } + let ticket = REQUESTED.fetch_add(1, Ordering::AcqRel) + 1; + let (start, freq) = now_ticks(); + let budget_ms = budget.as_millis().min(i64::MAX as u128 / freq as u128 / 2) as i64; + // SAFETY: SetEvent/WaitForSingleObject take no locks we could be holding. + unsafe { SetEvent(wake as Handle) }; + while ACKED.load(Ordering::Acquire) < ticket { + let (now, _) = now_ticks(); + let elapsed_ms = (now - start).saturating_mul(1000) / freq; + let left_ms = budget_ms - elapsed_ms; + if left_ms <= 0 { + return false; + } + // Wake on the helper's ack; the slice cap covers another waiter having + // consumed the (auto-reset) signal. + let ms = left_ms.clamp(1, 5) as u32; + // SAFETY: as above. + unsafe { WaitForSingleObject(ack as Handle, ms) }; + } + true + } +} + +#[cfg(not(any(unix, windows)))] +mod imp { + use std::time::Duration; + + pub fn thread_id() -> usize { + 0 + } + pub fn start_helper() -> std::io::Result<()> { + Err(std::io::Error::from(std::io::ErrorKind::Unsupported)) + } + pub fn signal_ack() {} + pub fn request_and_wait(_: Duration) -> bool { + false + } +} diff --git a/crates/bugsee-core/src/lib.rs b/crates/bugsee-core/src/lib.rs index e0e9b9b..5f2f989 100644 --- a/crates/bugsee-core/src/lib.rs +++ b/crates/bugsee-core/src/lib.rs @@ -19,6 +19,7 @@ pub mod bundle; pub mod capture; pub mod claim; +pub mod crash_flush; pub mod errors; pub mod fork; pub mod model; @@ -33,6 +34,8 @@ pub mod session; pub mod signature; pub mod transport; pub mod util; +#[cfg(windows)] +mod win32; pub use model::{ entry::{ diff --git a/crates/bugsee-core/src/process.rs b/crates/bugsee-core/src/process.rs index c857e46..a30b8bc 100644 --- a/crates/bugsee-core/src/process.rs +++ b/crates/bugsee-core/src/process.rs @@ -98,22 +98,7 @@ pub(crate) fn is_alive(_pid: u32) -> bool { } #[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; - } -} +use crate::win32 as win; #[cfg(test)] mod tests { diff --git a/crates/bugsee-core/src/runtime.rs b/crates/bugsee-core/src/runtime.rs index 51d1c7a..7e707c7 100644 --- a/crates/bugsee-core/src/runtime.rs +++ b/crates/bugsee-core/src/runtime.rs @@ -36,6 +36,7 @@ use std::time::{Duration, Instant}; use serde_json::Value; use crate::capture::{export, PartStore, WindowCaps}; +use crate::crash_flush::CrashFlush; use crate::fork::{ForkLock, ForkLockable}; use crate::model::entry::{Breadcrumb, CaptureEntry, TraceEntry}; use crate::model::enums::IssueType; @@ -375,6 +376,9 @@ pub struct Recorder { backoff_base: Duration, worker: Option>, uploader: Option>, + /// Lets a crash handler ask the worker to drain in-flight entries + /// ([`crate::crash_flush`]). `None` in memory mode (nothing to flush to). + crash_flush: Option, } /// The two pipeline threads and the channels that feed them. @@ -383,6 +387,7 @@ struct Pipeline { upload_tx: Sender, worker: JoinHandle<()>, uploader: JoinHandle<()>, + crash_flush: Option, } impl Recorder { @@ -520,6 +525,7 @@ impl Recorder { backoff_base: config.upload_backoff_base, worker: Some(pipeline.worker), uploader: Some(pipeline.uploader), + crash_flush: pipeline.crash_flush, }) } @@ -552,6 +558,9 @@ impl Recorder { // running anything. std::mem::forget(self.worker.take()); std::mem::forget(self.uploader.take()); + // Likewise the parent's crash-flush helper thread (and its wake handle, + // which the parent still uses). + std::mem::forget(self.crash_flush.take()); // Entries queued to the dead worker will never be processed, so the // back-pressure counters they incremented would stay high forever and @@ -607,6 +616,7 @@ impl Recorder { std::mem::forget(std::mem::replace(&mut self.upload_tx, pipeline.upload_tx)); self.worker = Some(pipeline.worker); self.uploader = Some(pipeline.uploader); + self.crash_flush = pipeline.crash_flush; Ok(()) } @@ -766,6 +776,9 @@ impl Drop for Recorder { let _ = handle.join(); } } + // Only now stop answering crash handlers: until the worker has acked + // `Stop` it can still hold queued entries a crash flush must wait for. + drop(self.crash_flush.take()); let (utx, urx) = channel(); let uploader_acked = self.upload_tx.send(UploadMsg::Stop(utx)).is_ok() && urx.recv_timeout(Duration::from_secs(2)).is_ok(); @@ -1050,10 +1063,13 @@ fn spawn_pipeline( // 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 crash_flush = store.is_some().then(|| install_crash_flush(&tx)).flatten(); let worker = std::thread::Builder::new() .name("bugsee-capture".into()) .spawn(move || { + // A crash on this thread cannot wait for itself to flush. + let _pipeline = crate::crash_flush::register_pipeline_thread(); // Queue any crashed prior session before capturing this one. // Contained: a panic here (e.g. a host before_send on a recovered // crash) must not abort the thread before worker_loop starts. @@ -1093,6 +1109,9 @@ fn spawn_pipeline( // Best effort, once: sessions a previous default location left // behind. No generation of ours lives there, so none is skipped. for dir in &legacy_data_dirs { + if crate::crash_flush::is_dying() { + break; + } let known = queue::blacklisted_signatures(dir); if !known.is_empty() { worker_shared.mem_blacklist.lock().extend(known); @@ -1123,7 +1142,27 @@ fn spawn_pipeline( upload_tx, worker, uploader, + crash_flush, + }) +} + +/// How long the helper waits for the worker to acknowledge a crash-time flush. +/// The crash handler gives up on its own, far sooner (its budget); this only +/// bounds how long a permanently wedged worker can occupy the helper. It must +/// outlast any legitimate drain (the queue is capped), or a slow-but-healthy +/// worker would make the helper report failure for a request whose budget was +/// longer than this. +const CRASH_FLUSH_ACK_TIMEOUT: Duration = Duration::from_secs(10); + +/// Register this pipeline as the process's crash-time flush target. A failure +/// (no pipe/event available) just forgoes the best-effort flush. +fn install_crash_flush(tx: &Sender) -> Option { + let tx = tx.clone(); + CrashFlush::install(move || { + let (ack, done) = channel(); + tx.send(Msg::Flush(ack)).is_ok() && done.recv_timeout(CRASH_FLUSH_ACK_TIMEOUT).is_ok() }) + .ok() } /// Recover any prior generation that ended abnormally, queueing it for delivery. @@ -1140,6 +1179,11 @@ fn run_recovery( heartbeat_of: Option<&Path>, skip_launch_pass: bool, ) { + // A dying process starts no scan or GC a kill cannot finish: the worker must + // reach its loop to answer the crash handler's in-flight flush (crash_flush). + if crate::crash_flush::is_dying() { + return; + } let beat = || { if let Some(marker) = heartbeat_of { touch_heartbeat(marker); @@ -1147,6 +1191,9 @@ fn run_recovery( }; beat(); for pending in recovery::find_pending(data_dir, current_generation) { + if crate::crash_flush::is_dying() { + return; + } beat(); if skip_launch_pass && shared.launch_attempted.lock().contains(&pending.generation) { continue; @@ -1154,7 +1201,9 @@ fn run_recovery( recover_claimed(shared, data_dir, &pending, false); } // Reap parts/attempts left by cleanly-exited or already-recovered generations. - recovery::gc_orphans(data_dir, current_generation); + if !crate::crash_flush::is_dying() { + recovery::gc_orphans(data_dir, current_generation); + } } /// Claim one pending session and recover it. Returns whether this process won the @@ -1206,7 +1255,13 @@ fn wait_for_launch_pass(shared: &Shared) { } let wait = Duration::from_millis(shared.launch_wait_ms.load(Ordering::SeqCst)); let deadline = Instant::now() + wait + LAUNCH_PASS_GRACE; - while !shared.launch_done.load(Ordering::SeqCst) && Instant::now() < deadline { + // A dying process stops waiting at once: the worker must be free to answer the + // crash handler's in-flight flush (crash_flush), which a crash INSIDE the + // launch pass would otherwise leave queued behind this wait. + while !shared.launch_done.load(Ordering::SeqCst) + && !crate::crash_flush::is_dying() + && Instant::now() < deadline + { std::thread::sleep(Duration::from_millis(2)); } } @@ -1231,6 +1286,10 @@ fn run_launch_recovery(shared: &Shared, plan: &LaunchPlan) { let started = Instant::now(); let mut claimed = 0; for pending in recovery::find_pending_bounded(&plan.dir, plan.generation, plan.max_sessions) { + // A crash is in progress: start no more sessions (claims are refused too). + if crate::crash_flush::is_dying() { + break; + } if let Some(marker) = &plan.heartbeat_of { touch_heartbeat(marker); } @@ -2055,6 +2114,10 @@ fn drain_memory( if deadline.is_some_and(|d| Instant::now() >= d) { break; } + // A dying process starts no delivery it cannot finish (crash_flush). + if crate::crash_flush::is_dying() { + break; + } let now = epoch_ms(); let next = shared .mem_queue diff --git a/crates/bugsee-core/src/win32.rs b/crates/bugsee-core/src/win32.rs new file mode 100644 index 0000000..066123b --- /dev/null +++ b/crates/bugsee-core/src/win32.rs @@ -0,0 +1,39 @@ +// +// win32.rs +// bugsee-core +// +// Copyright © 2026 Bugsee. All rights reserved. +// + +//! The few kernel32 calls this crate needs, declared ONCE. +//! +//! kernel32 is always linked, so declaring them directly avoids a dependency. +//! They live in one module on purpose: two declarations of the same symbol with +//! different Rust types (a `HANDLE` as `*mut c_void` in one place, `usize` in +//! another) trip `clashing_extern_declarations`, a crate-wide lint. Any new +//! kernel32 call in this crate goes here. + +use std::ffi::c_void; + +pub type Handle = *mut c_void; + +pub const SYNCHRONIZE: u32 = 0x0010_0000; +pub const WAIT_OBJECT_0: u32 = 0; +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; + pub fn CreateEventW( + attrs: *const c_void, + manual_reset: i32, + initial: i32, + name: *const u16, + ) -> Handle; + pub fn SetEvent(event: Handle) -> i32; + pub fn GetCurrentThreadId() -> u32; + pub fn QueryPerformanceCounter(value: *mut i64) -> i32; + pub fn QueryPerformanceFrequency(value: *mut i64) -> i32; +} diff --git a/crates/bugsee-core/tests/crash_flush_e2e.rs b/crates/bugsee-core/tests/crash_flush_e2e.rs new file mode 100644 index 0000000..790dd2d --- /dev/null +++ b/crates/bugsee-core/tests/crash_flush_e2e.rs @@ -0,0 +1,406 @@ +// +// crash_flush_e2e.rs +// bugsee-core +// +// Copyright © 2026 Bugsee. All rights reserved. +// +//! The crash-time flush handshake against a real capture pipeline (issue #13). +//! The crash handler's side is exactly `flush_in_flight`; the subprocess crash +//! tests in the `bugsee` crate prove it from a real fault. +//! +//! The registration is process-global, so these tests run one at a time. + +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use bugsee_core::capture::record::RecordIter; +use bugsee_core::crash_flush::{flush_in_flight, CrashFlush}; +use bugsee_core::model::entry::{CaptureEntry, EventEntry}; +use bugsee_core::runtime::TelemetrySampler; +use bugsee_core::{MockTransport, Recorder, RecorderConfig}; +use serde_json::{Map, Value}; + +static SERIAL: Mutex<()> = Mutex::new(()); + +struct TempDir(PathBuf); +impl TempDir { + fn new() -> Self { + let p = std::env::temp_dir().join(format!( + "bugsee-cflush-{}", + 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 event(i: usize) -> CaptureEntry { + let mut params = Map::new(); + params.insert("pad".into(), Value::String("x".repeat(300))); + CaptureEntry::UserEvent(EventEntry { + timestamp: 1_000 + i as i64, + name: Some(format!("e{i}")), + params, + custom: Map::new(), + }) +} + +/// Count the records the part files hold RIGHT NOW (no flush of ours involved). +fn records_on_disk(data: &Path) -> usize { + let mut n = 0; + let mut stack = vec![data.join("parts")]; + while let Some(dir) = stack.pop() { + let Ok(read) = std::fs::read_dir(&dir) else { + continue; + }; + for e in read.flatten() { + let p = e.path(); + if p.is_dir() { + stack.push(p); + } else if p.to_string_lossy().ends_with("events.user.part") { + n += RecordIter::new(&std::fs::read(&p).unwrap()).count(); + } + } + } + n +} + +#[test] +fn everything_queued_before_the_request_is_on_disk_when_it_confirms() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let dir = TempDir::new(); + let recorder = Recorder::launch( + RecorderConfig::new(&dir.0, "TOKEN"), + Arc::new(MockTransport::default()), + ) + .unwrap(); + + const N: usize = 1500; + for i in 0..N { + recorder.capture(event(i)); + } + // The producer only queued; the worker is still behind. + let behind = records_on_disk(&dir.0); + + assert!( + flush_in_flight(Duration::from_secs(5)), + "a healthy worker confirms" + ); + assert_eq!(records_on_disk(&dir.0), N, "all {N} are durable on confirm"); + eprintln!("on disk before the request: {behind}/{N}"); + drop(recorder); +} + +/// Blocks the worker inside `sample` until released, and says when it got there. +struct Blocker { + release: Arc, + entered: Arc, +} +impl TelemetrySampler for Blocker { + fn sample(&mut self) -> Vec<(String, Value)> { + self.entered.store(true, Ordering::SeqCst); + while !self.release.load(Ordering::SeqCst) { + std::thread::sleep(Duration::from_millis(5)); + } + Vec::new() + } +} + +#[test] +fn a_stuck_worker_costs_only_the_budget() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let dir = TempDir::new(); + let release = Arc::new(AtomicBool::new(false)); + let entered = Arc::new(AtomicBool::new(false)); + let mut cfg = RecorderConfig::new(&dir.0, "TOKEN"); + cfg.rotate_interval = Duration::from_millis(20); + cfg.sampler = Some(Box::new(Blocker { + release: Arc::clone(&release), + entered: Arc::clone(&entered), + })); + let recorder = Recorder::launch(cfg, Arc::new(MockTransport::default())).unwrap(); + // Wait (observably, not by sleeping) until the worker is wedged in the sampler. + let deadline = Instant::now() + Duration::from_secs(60); + while !entered.load(Ordering::SeqCst) { + assert!(Instant::now() < deadline, "the worker never reached a tick"); + std::thread::sleep(Duration::from_millis(5)); + } + for i in 0..100 { + recorder.capture(event(i)); + } + + let t = Instant::now(); + assert!(!flush_in_flight(Duration::from_millis(50))); + let took = t.elapsed(); + assert!(took < Duration::from_secs(5), "bounded, took {took:?}"); + + // Un-wedge: the same request now succeeds (the helper is not poisoned). The + // budget is generous: this proves the handshake recovers, not how fast a + // loaded CI machine drains. + release.store(true, Ordering::SeqCst); + assert!(flush_in_flight(Duration::from_secs(60))); + assert_eq!(records_on_disk(&dir.0), 100); + drop(recorder); +} + +/// A flush target that only counts its calls. +fn counter() -> (Arc, impl Fn() -> bool + Send + Sync + 'static) { + let n = Arc::new(AtomicUsize::new(0)); + let c = Arc::clone(&n); + (n, move || { + c.fetch_add(1, Ordering::SeqCst); + true + }) +} + +#[test] +fn dropping_the_recorder_stops_flushing_into_it() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let dir = TempDir::new(); + let (calls, f) = counter(); + let _sentinel = CrashFlush::install(f).unwrap(); + let recorder = Recorder::launch( + RecorderConfig::new(&dir.0, "TOKEN"), + Arc::new(MockTransport::default()), + ) + .unwrap(); + assert!(flush_in_flight(Duration::from_secs(30))); + assert_eq!( + calls.load(Ordering::SeqCst), + 0, + "the recorder is the target" + ); + drop(recorder); + // Nothing is registered now, so a request is answered at once rather than + // waited out, and the replaced sentinel is not called. + let t = Instant::now(); + assert!(flush_in_flight(Duration::from_secs(30))); + assert!(t.elapsed() < Duration::from_secs(10), "no wait at all"); + assert_eq!(calls.load(Ordering::SeqCst), 0); +} + +#[test] +fn memory_mode_does_not_take_over_the_target() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let dir = TempDir::new(); + let (calls, f) = counter(); + let _sentinel = CrashFlush::install(f).unwrap(); + let mut cfg = RecorderConfig::new(&dir.0, "TOKEN"); + cfg.memory_only = Some("test".into()); + let recorder = Recorder::launch(cfg, Arc::new(MockTransport::default())).unwrap(); + assert!(flush_in_flight(Duration::from_secs(30))); + assert_eq!(calls.load(Ordering::SeqCst), 1, "still the sentinel"); + drop(recorder); +} + +// The handshake itself, against stand-in flushers (no pipeline). + +#[test] +fn a_responsive_flusher_is_confirmed_promptly() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let (calls, f) = counter(); + let cf = CrashFlush::install(f).unwrap(); + let t = Instant::now(); + assert!(flush_in_flight(Duration::from_secs(30))); + assert!(t.elapsed() < Duration::from_secs(5)); + assert_eq!(calls.load(Ordering::SeqCst), 1); + drop(cf); + assert!(flush_in_flight(Duration::from_secs(30))); + assert_eq!(calls.load(Ordering::SeqCst), 1, "deregistered: not called"); +} + +#[test] +fn the_last_registered_target_wins() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let (a_calls, a) = counter(); + let (b_calls, b) = counter(); + let a = CrashFlush::install(a).unwrap(); + let b = CrashFlush::install(b).unwrap(); + assert!(flush_in_flight(Duration::from_secs(30))); + assert_eq!( + ( + a_calls.load(Ordering::SeqCst), + b_calls.load(Ordering::SeqCst) + ), + (0, 1) + ); + // Dropping the one that is not current must not deregister the current one. + drop(a); + assert!(flush_in_flight(Duration::from_secs(30))); + assert_eq!(b_calls.load(Ordering::SeqCst), 2); + drop(b); +} + +#[test] +fn a_stuck_flusher_costs_the_budget_and_no_more() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let release = Arc::new(AtomicBool::new(false)); + let r = Arc::clone(&release); + let cf = CrashFlush::install(move || { + while !r.load(Ordering::SeqCst) { + std::thread::sleep(Duration::from_millis(5)); + } + true + }) + .unwrap(); + let t = Instant::now(); + assert!(!flush_in_flight(Duration::from_millis(50))); + let took = t.elapsed(); + assert!(took >= Duration::from_millis(45), "{took:?}"); + assert!(took < Duration::from_secs(5), "{took:?}"); + release.store(true, Ordering::SeqCst); + drop(cf); +} + +#[test] +fn a_panicking_flusher_does_not_kill_the_helper() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let n = Arc::new(AtomicUsize::new(0)); + let c = Arc::clone(&n); + let cf = CrashFlush::install(move || { + if c.fetch_add(1, Ordering::SeqCst) == 0 { + panic!("boom"); + } + true + }) + .unwrap(); + assert!(!flush_in_flight(Duration::from_millis(50))); + // Let the helper finish the first request: two wakeups that land before it + // reads are served by ONE flush call (which here panics). + while n.load(Ordering::SeqCst) == 0 { + std::thread::sleep(Duration::from_millis(5)); + } + assert!(flush_in_flight(Duration::from_secs(30))); + drop(cf); +} + +#[test] +fn a_pipeline_thread_does_not_wait_for_itself() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let (calls, f) = counter(); + let cf = CrashFlush::install(f).unwrap(); + let t = Instant::now(); + let r = std::thread::spawn(|| { + let _p = bugsee_core::crash_flush::register_pipeline_thread(); + flush_in_flight(Duration::from_secs(30)) + }) + .join() + .unwrap(); + assert!(!r, "skipped"); + assert!(t.elapsed() < Duration::from_secs(10)); + assert_eq!(calls.load(Ordering::SeqCst), 0); + drop(cf); +} + +#[test] +fn a_crash_flush_raises_the_dying_flag_and_blocks_new_claims() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let dir = TempDir::new(); + let claim = dir.0.join("x.claim"); + bugsee_core::crash_flush::reset_dying_for_test(); + assert!(bugsee_core::claim::try_claim(&claim).is_some()); + let (_calls, f) = counter(); + let cf = CrashFlush::install(f).unwrap(); + // The plain flush (a panic hook's) does not declare the process dying. + flush_in_flight(Duration::from_millis(100)); + assert!(!bugsee_core::crash_flush::is_dying()); + bugsee_core::crash_flush::flush_for_crash(Duration::from_millis(100)); + assert!(bugsee_core::crash_flush::is_dying()); + assert!( + bugsee_core::claim::try_claim(&claim).is_none(), + "a dying process starts no claim a kill would leave half-made" + ); + assert!(bugsee_core::claim::try_claim_lenient(&claim).is_none()); + bugsee_core::crash_flush::reset_dying_for_test(); + assert!(bugsee_core::claim::try_claim(&claim).is_some()); + drop(cf); +} + +/// A crash while the launch pass / worker recovery runs: a dying process must +/// recover nothing (a kill would strand the claim), and the session stays pending +/// for the next launch. +#[test] +fn a_dying_process_recovers_nothing_and_leaves_the_session_pending() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let dir = TempDir::new(); + std::fs::write(dir.0.join("gen"), "1").unwrap(); + let part = dir.0.join("parts/1/0"); + std::fs::create_dir_all(&part).unwrap(); + let payload = br#"{"timestamp":100,"level":1,"source":2,"message":"x"}"#; + let mut bytes = Vec::new(); + bugsee_core::capture::record::frame(100, payload, &mut bytes); + std::fs::write(part.join("log.part"), &bytes).unwrap(); + std::fs::create_dir_all(dir.0.join("sessions")).unwrap(); + std::fs::write(dir.0.join("sessions/1.alive"), "0").unwrap(); + + bugsee_core::crash_flush::flush_for_crash(Duration::from_millis(10)); + assert!(bugsee_core::crash_flush::is_dying()); + let transport = Arc::new(MockTransport::default()); + let r1 = Recorder::launch(RecorderConfig::new(&dir.0, "TOKEN"), transport.clone()).unwrap(); + r1.recover_on_launch_path(); + assert!(r1.flush(Duration::from_secs(5))); + assert!(transport.uploaded_bundles.lock().unwrap().is_empty()); + assert!(dir.0.join("sessions/1.alive").exists(), "still pending"); + drop(r1); + + // The worker reached its loop at once (no scan/GC ahead of it), so a flush + // request is answered promptly and what was queued is on disk. + let r1 = Recorder::launch(RecorderConfig::new(&dir.0, "TOKEN"), transport.clone()).unwrap(); + for i in 0..200 { + r1.capture(event(i)); + } + let t = Instant::now(); + assert!(flush_in_flight(Duration::from_secs(30))); + assert!(t.elapsed() < Duration::from_secs(30)); + assert_eq!(records_on_disk(&dir.0), 200); + drop(r1); + + bugsee_core::crash_flush::reset_dying_for_test(); + let r2 = Recorder::launch(RecorderConfig::new(&dir.0, "TOKEN"), transport.clone()).unwrap(); + r2.recover_on_launch_path(); + assert!(r2.flush(Duration::from_secs(5))); + assert_eq!(transport.uploaded_bundles.lock().unwrap().len(), 1); + drop(r2); +} + +/// A `fork()` child has none of the parent's threads: it must not wait on the +/// parent's helper, and must get its own once it registers. +#[cfg(unix)] +#[test] +fn a_fork_child_gets_its_own_helper() { + let _g = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let (_calls, f) = counter(); + let cf = CrashFlush::install(f).unwrap(); + assert!(flush_in_flight(Duration::from_secs(30))); + // SAFETY: the child only calls the code under test and `_exit`s. + let pid = unsafe { libc::fork() }; + if pid == 0 { + let before = flush_in_flight(Duration::from_secs(30)); + let (_c, g) = counter(); + let mine = CrashFlush::install(g).unwrap(); + let after = flush_in_flight(Duration::from_secs(30)); + std::mem::forget(mine); + let code = if !before && after { 0 } else { 1 }; + // SAFETY: leaving a forked child. + unsafe { libc::_exit(code) }; + } + let mut status = 0; + // SAFETY: waiting for our own child. + unsafe { libc::waitpid(pid, &mut status, 0) }; + assert!( + libc::WIFEXITED(status) && libc::WEXITSTATUS(status) == 0, + "child status {status}" + ); + assert!( + flush_in_flight(Duration::from_secs(30)), + "parent unaffected" + ); + drop(cf); +} diff --git a/crates/bugsee-core/tests/recovery_e2e.rs b/crates/bugsee-core/tests/recovery_e2e.rs index 4e88f14..456d219 100644 --- a/crates/bugsee-core/tests/recovery_e2e.rs +++ b/crates/bugsee-core/tests/recovery_e2e.rs @@ -363,3 +363,69 @@ fn clean_shutdown_is_not_recovered() { "a cleanly-closed session is never recovered" ); } + +/// A process that dies mid-`write` (or whose in-flight flush was cut short by the +/// crash handler's budget) leaves a torn tail on the part it was appending to. +/// Recovery must deliver every complete record before it and ignore whatever +/// follows — never panic, never loop, never read past the buffer. +#[test] +fn a_torn_or_garbage_part_tail_is_ignored_on_recovery() { + let header = |len: u32| { + let mut t = Vec::new(); + t.extend_from_slice(&300u64.to_le_bytes()); + t.extend_from_slice(&len.to_le_bytes()); + t + }; + let mut cut = header(64); + cut.extend_from_slice(br#"{"timestamp":300,"lev"#); + let mut absurd = header(u32::MAX); + absurd.extend_from_slice(b"junk"); + let tails: Vec<(&str, Vec)> = vec![ + ("partial header", vec![0xAB; 5]), + ("header only, payload missing", header(64)), + ("payload cut mid-way", cut), + ("absurd length prefix", absurd), + ("zeroed tail of a grown file", vec![0u8; 4096]), + ( + "random garbage", + (0..777u32).map(|i| (i * 31 + 7) as u8).collect(), + ), + ]; + + for (what, tail) in tails { + let dir = TempDir::new(); + seed_crashed_session(&dir.path, 1); + let log = dir.path.join("parts/1/0/log.part"); + let mut bytes = std::fs::read(&log).unwrap(); + bytes.extend_from_slice(&tail); + std::fs::write(&log, &bytes).unwrap(); + + let transport = Arc::new(MockTransport::default()); + let recorder = + Recorder::launch(RecorderConfig::new(&dir.path, "TOKEN"), transport.clone()).unwrap(); + assert!(recorder.flush(Duration::from_secs(5)), "{what}"); + + let bundles = transport.uploaded_bundles.lock().unwrap(); + assert_eq!(bundles.len(), 1, "{what}: the session was still delivered"); + let mut zip = ZipArchive::new(Cursor::new(bundles[0].clone())).unwrap(); + let log_name = (0..zip.len()) + .map(|i| zip.by_index(i).unwrap().name().to_string()) + .find(|n| n.ends_with(".log.json")) + .unwrap_or_else(|| panic!("{what}: no log file")); + let mut raw = Vec::new(); + zip.by_name(&log_name) + .unwrap() + .read_to_end(&mut raw) + .unwrap(); + let log: Value = serde_json::from_slice(&raw).unwrap(); + let msgs: Vec<&str> = log["events"] + .as_array() + .unwrap() + .iter() + .map(|e| e["message"].as_str().unwrap()) + .collect(); + assert_eq!(msgs, vec!["pre-crash line A", "pre-crash line B"], "{what}"); + drop(bundles); + drop(recorder); + } +} diff --git a/crates/bugsee-native/src/lib.rs b/crates/bugsee-native/src/lib.rs index 273a5e4..907e3ae 100644 --- a/crates/bugsee-native/src/lib.rs +++ b/crates/bugsee-native/src/lib.rs @@ -225,6 +225,40 @@ mod imp { #[cfg(any(unix, windows))] const HANDLER_BUDGET_SECS: u32 = 5; + /// How long a dying process waits for the capture worker to write out the + /// entries still queued when it faulted (see `bugsee_core::crash_flush`), in + /// milliseconds. Small against [`HANDLER_BUDGET_SECS`] on purpose: it is a + /// courtesy, never something the process may be held for, and it runs only + /// AFTER the marker is complete, so giving up costs the last few events and + /// nothing else. + #[cfg(any(unix, windows))] + static INFLIGHT_FLUSH_BUDGET_MS: core::sync::atomic::AtomicU64 = + core::sync::atomic::AtomicU64::new( + bugsee_core::crash_flush::DEFAULT_BUDGET.as_millis() as u64 + ); + + /// Override the in-flight flush budget. Tests only: they use a long one to + /// prove the mechanism independent of how fast the machine drains. + #[cfg(any(unix, windows))] + #[doc(hidden)] + pub fn set_inflight_flush_budget_for_test(budget: std::time::Duration) { + INFLIGHT_FLUSH_BUDGET_MS.store( + budget.as_millis() as u64, + core::sync::atomic::Ordering::SeqCst, + ); + } + + /// Best-effort drain of in-flight capture entries. Async-signal-safe (atomics, + /// one non-blocking write, a bounded poll loop); the heavy lifting happens on + /// the capture worker. Call only once the marker is on disk. Also raises the + /// dying flag, so the runtime starts no new claims in the window. + #[cfg(any(unix, windows))] + #[inline(never)] + fn flush_inflight_events() { + let ms = INFLIGHT_FLUSH_BUDGET_MS.load(core::sync::atomic::Ordering::Relaxed); + let _ = bugsee_core::crash_flush::flush_for_crash(std::time::Duration::from_millis(ms)); + } + /// A `SIGALRM`-based deadline for the crash handler. Only async-signal-safe /// calls (`sigaction`, `pthread_sigmask`, `sigpending`, `sigwait`, `alarm`) are /// used. @@ -878,6 +912,9 @@ mod imp { if n > 0 { unsafe { append_frames(path_cbytes, &frames[..n]) }; } + // The marker is complete; now give the worker a moment to write out what + // was still queued. + flush_inflight_events(); } /// Open the marker for writing, creating/truncating it. Returns the raw handle @@ -2409,6 +2446,8 @@ mod imp { report_unrecorded(tid, &*path, sig, info, regs); } } + // Only now, with the marker complete and the latch released. + flush_inflight_events(); } /// The interrupted registers, to tell whether a host changed anything. @@ -2660,6 +2699,7 @@ mod imp { unsafe { if generation != 0 { complete_report(tid, &*path, regs, generation); + flush_inflight_events(); return; } // The original fault was not the first in flight, so it wrote no @@ -2681,6 +2721,7 @@ mod imp { watchdog.disarm(); } WRITING.store(0, Ordering::Release); + flush_inflight_events(); } } diff --git a/crates/bugsee-panic/src/lib.rs b/crates/bugsee-panic/src/lib.rs index e0e285d..728c277 100644 --- a/crates/bugsee-panic/src/lib.rs +++ b/crates/bugsee-panic/src/lib.rs @@ -215,6 +215,11 @@ pub fn install(reporter: Arc) { // emit the same panic twice. let worker_thread = worker_thread_name(); let will_be_caught = guard_active(); + // A main-thread panic nothing will catch is the process's last act under + // `panic = "abort"` (and usually under unwind too): drain the capture + // worker so the events emitted just before it are on disk. Bounded and + // best effort; a no-op when the SDK has no disk pipeline. + let likely_fatal = !will_be_caught && worker_thread.is_none(); if !will_be_caught && (worker_thread.is_some() || REPORT_FROM_HOOK.load(Ordering::SeqCst)) { @@ -247,6 +252,16 @@ pub fn install(reporter: Arc) { frames, }); }); + if likely_fatal { + // Runs AFTER the inline report above (when `report_panics_from_hook` + // is on), so that report's processing is ahead of the flush marker + // in the worker's queue and is part of what is waited for. A panic + // the host goes on to catch outside our guards pays up to the + // budget once. The same budget the native handler uses. + let _ = bugsee_core::crash_flush::flush_in_flight( + bugsee_core::crash_flush::DEFAULT_BUDGET, + ); + } // Preserve prior hook behavior (default abort message, other SDKs). previous(info); })); diff --git a/crates/bugsee/Cargo.toml b/crates/bugsee/Cargo.toml index 068b3f2..814f254 100644 --- a/crates/bugsee/Cargo.toml +++ b/crates/bugsee/Cargo.toml @@ -48,5 +48,11 @@ name = "crash_e2e_matrix" harness = false required-features = ["native"] +# Dies for real after a burst of events; needs its own `main`, like the above. +[[test]] +name = "crash_inflight" +harness = false +required-features = ["native"] + [target.'cfg(unix)'.dev-dependencies] libc = "0.2" diff --git a/crates/bugsee/tests/crash_inflight.rs b/crates/bugsee/tests/crash_inflight.rs new file mode 100644 index 0000000..18c206f --- /dev/null +++ b/crates/bugsee/tests/crash_inflight.rs @@ -0,0 +1,305 @@ +// +// crash_inflight.rs +// bugsee +// +// Copyright © 2026 Bugsee. All rights reserved. +// +//! Best-effort flush of in-flight events at crash time (issue #13). +//! +//! A producer only queues an entry; the capture worker appends it to disk a +//! little later. These tests die for real, in a child process, immediately after +//! emitting a burst of events, and read what the next launch recovers: +//! +//! * `burst_then_segv` — with a long flush budget (so the result does not depend +//! on how fast this machine drains), every event emitted before the fault, +//! including the ones still queued when it hit, is in the recovered report. +//! Without the flush only a few dozen of the 1000 survive. +//! * `small_burst_real_budget` — the same under the real 50 ms budget, with a +//! burst small enough to drain inside it. +//! * `wedged_worker_then_segv` — the worker is stuck in a host callback, so the +//! flush cannot succeed. The process must wait the (test-shortened) budget and +//! no longer — proving the wait happens and is bounded; the 5 s watchdog must +//! not be what ends it — and the crash is still reported, with the events that +//! had already reached disk. +//! +//! `harness = false` (see `Cargo.toml`): the child must fault from the real main +//! thread. + +#![cfg(all(feature = "native", any(unix, windows)))] + +use std::io::{BufRead, BufReader, Cursor, Read, Write}; +use std::process::{Command, Stdio}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use bugsee::core::MockTransport; +use bugsee::{Bugsee, LaunchOptions, LogLevel}; +use serde_json::{Map, Value}; +use zip::ZipArchive; + +const KIND_ENV: &str = "BUGSEE_INFLIGHT_KIND"; +const DIR_ENV: &str = "BUGSEE_INFLIGHT_DIR"; +/// Events in the burst. Sized so the producer (which only queues) finishes well +/// before the worker (which serializes and writes) has drained, yet the worker +/// can drain within the 50 ms handler budget. +const BURST: usize = 1000; +/// The burst for the real-budget case: drains well inside 50 ms even on a slow +/// core, yet is far more than the ~20-40 the no-flush baseline recovers. +const SMALL_BURST: usize = 250; +/// Budget used by the cases that prove the mechanism rather than the speed. +const LONG_BUDGET: Duration = Duration::from_secs(5); +/// Budget used by the wedged case: long enough to measure, short enough to run. +const WEDGE_BUDGET: Duration = Duration::from_millis(300); + +#[inline(never)] +fn fault_null() { + // SAFETY: deliberately faulting. + unsafe { std::ptr::write_volatile(std::hint::black_box(std::ptr::null_mut::()), 1) } +} + +fn emit_burst(prefix: &str, count: usize) { + let pad = "x".repeat(400); + for i in 0..count { + let mut params = Map::new(); + params.insert("pad".into(), Value::String(pad.clone())); + Bugsee::event_with(format!("{prefix}-{i}"), params); + } +} + +fn run_child(kind: &str, dir: &str) -> ! { + let wedge = kind == "wedged_worker_then_segv"; + let _guard = Bugsee::launch_with( + LaunchOptions::new("APP_TOKEN") + .data_dir(dir) + .native_crash_capture(true) + // Runs on the capture worker: wedging it stands in for a worker blocked + // on a dead disk or a host callback that never returns. + .before_send(move |_| { + if wedge { + loop { + std::thread::sleep(Duration::from_secs(60)); + } + } + #[allow(unreachable_code)] + true + }) + .with_transport(Arc::new(MockTransport::default())), + ) + .expect("launch"); + std::thread::sleep(Duration::from_millis(300)); + + match kind { + "burst_then_segv" => { + bugsee_native::set_inflight_flush_budget_for_test(LONG_BUDGET); + emit_burst("burst", BURST); + } + "small_burst_real_budget" => emit_burst("small", SMALL_BURST), + "wedged_worker_then_segv" => { + // These reach disk before the worker is wedged... + for i in 0..20 { + Bugsee::event(format!("early-{i}")); + } + Bugsee::flush(Duration::from_secs(2)); + // ...then a report wedges it in `before_send`, and everything after + // is stuck in the queue. + Bugsee::capture_message(LogLevel::Error, "wedge the worker"); + std::thread::sleep(Duration::from_millis(300)); + bugsee_native::set_inflight_flush_budget_for_test(WEDGE_BUDGET); + emit_burst("stuck", 100); + } + other => panic!("unknown kind {other}"), + } + println!("FAULTING"); + let _ = std::io::stdout().flush(); + fault_null(); + std::process::exit(0); +} + +fn event_names(bundle: &[u8]) -> Vec { + let mut zip = ZipArchive::new(Cursor::new(bundle)).expect("bundle is a zip"); + let mut names = Vec::new(); + for i in 0..zip.len() { + let mut entry = zip.by_index(i).expect("zip entry"); + if !entry.name().contains("events.user") { + continue; + } + let mut raw = Vec::new(); + entry.read_to_end(&mut raw).expect("read events"); + let v: Value = serde_json::from_slice(&raw).expect("events parse"); + for e in v["events"].as_array().cloned().unwrap_or_default() { + if let Some(n) = e["name"].as_str() { + names.push(n.to_string()); + } + } + } + names +} + +fn crash_json(bundle: &[u8]) -> Value { + let mut zip = ZipArchive::new(Cursor::new(bundle)).expect("bundle is a zip"); + let mut entry = zip.by_name("crash.json").expect("crash.json"); + let mut raw = Vec::new(); + entry.read_to_end(&mut raw).unwrap(); + serde_json::from_slice(&raw).unwrap() +} + +fn recover(dir: &std::path::Path) -> Vec> { + let mock = Arc::new(MockTransport::default()); + let guard = Bugsee::launch_with( + LaunchOptions::new("APP_TOKEN") + .data_dir(dir) + .native_crash_capture(false) + .with_transport(mock.clone()), + ) + .expect("relaunch"); + std::thread::sleep(Duration::from_millis(1500)); + Bugsee::flush(Duration::from_secs(10)); + let bundles = mock.uploaded_bundles.lock().unwrap().clone(); + drop(guard); + bundles +} + +/// Run the child to its death; returns the time from its "about to fault" line to +/// its exit. +fn crash_child(kind: &str, dir: &std::path::Path) -> Result { + let mut child = Command::new(std::env::current_exe().unwrap()) + .env(KIND_ENV, kind) + .env(DIR_ENV, dir) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .map_err(|e| format!("spawn: {e}"))?; + let mut line = String::new(); + BufReader::new(child.stdout.take().unwrap()) + .read_line(&mut line) + .map_err(|e| e.to_string())?; + let faulted = Instant::now(); + if line.trim() != "FAULTING" { + let _ = child.kill(); + return Err(format!("child never reached the fault: {line:?}")); + } + let status = child.wait().map_err(|e| e.to_string())?; + let took = faulted.elapsed(); + if status.success() { + return Err("child should have died".into()); + } + Ok(took) +} + +fn with_dir( + kind: &str, + f: impl FnOnce(&std::path::Path) -> Result, +) -> Result { + let dir = std::env::temp_dir().join(format!( + "bugsee-inflight-{kind}-{}", + bugsee::core::util::random_hex(6) + )); + std::fs::create_dir_all(&dir).map_err(|e| e.to_string())?; + let r = f(&dir); + let _ = std::fs::remove_dir_all(&dir); + r +} + +fn burst_then_segv() -> Result<(), String> { + with_dir("burst", |dir| { + crash_child("burst_then_segv", dir)?; + let bundles = recover(dir); + if bundles.len() != 1 { + return Err(format!("expected one report, got {}", bundles.len())); + } + let crash = crash_json(&bundles[0]); + if crash["signal"]["name"] != "SIGSEGV" { + return Err(format!("not a SIGSEGV report: {crash}")); + } + let names = event_names(&bundles[0]); + let last = format!("burst-{}", BURST - 1); + let have = names.iter().filter(|n| n.starts_with("burst-")).count(); + if !names.contains(&last) { + return Err(format!( + "the last event before the crash was lost ({have}/{BURST} recovered)" + )); + } + if have != BURST { + return Err(format!("only {have}/{BURST} events recovered")); + } + Ok(()) + }) +} + +fn small_burst_real_budget() -> Result<(), String> { + with_dir("small", |dir| { + crash_child("small_burst_real_budget", dir)?; + let bundles = recover(dir); + if bundles.len() != 1 { + return Err(format!("expected one report, got {}", bundles.len())); + } + let names = event_names(&bundles[0]); + let have = names.iter().filter(|n| n.starts_with("small-")).count(); + let last = format!("small-{}", SMALL_BURST - 1); + if !names.contains(&last) || have != SMALL_BURST { + return Err(format!( + "events lost under the real budget ({have}/{SMALL_BURST}, last present: {})", + names.contains(&last) + )); + } + Ok(()) + }) +} + +fn wedged_worker_then_segv() -> Result<(), String> { + with_dir("wedged", |dir| { + let took = crash_child("wedged_worker_then_segv", dir)?; + // Waited the budget (without the feature the process dies at once) and no + // longer (the 5 s watchdog must not be what ended it). + if took < WEDGE_BUDGET - Duration::from_millis(50) { + return Err(format!("died in {took:?}: the flush did not wait")); + } + if took > Duration::from_secs(1) { + return Err(format!("the dying process hung for {took:?}")); + } + let bundles = recover(dir); + if bundles.len() != 1 { + return Err(format!("expected one report, got {}", bundles.len())); + } + let crash = crash_json(&bundles[0]); + let has_frames = crash["frames"].as_array().is_some_and(|f| !f.is_empty()); + if crash["signal"]["name"] != "SIGSEGV" || !has_frames { + return Err(format!("crash report incomplete: {crash}")); + } + let names = event_names(&bundles[0]); + if !names.contains(&"early-19".to_string()) { + return Err(format!("events already on disk were lost: {names:?}")); + } + Ok(()) + }) +} + +type Case = fn() -> Result<(), String>; + +fn main() { + if let Ok(kind) = std::env::var(KIND_ENV) { + run_child(&kind, &std::env::var(DIR_ENV).expect("dir")); + } + let filter = std::env::args().skip(1).find(|a| !a.starts_with('-')); + let cases: [(&str, Case); 3] = [ + ("burst_then_segv", burst_then_segv), + ("small_burst_real_budget", small_burst_real_budget), + ("wedged_worker_then_segv", wedged_worker_then_segv), + ]; + let mut failed = Vec::new(); + for (name, case) in cases { + if filter.as_deref().is_some_and(|f| !name.contains(f)) { + continue; + } + match case() { + Ok(()) => println!("test crash_inflight::{name} ... ok"), + Err(e) => { + println!("test crash_inflight::{name} ... FAILED\n {e}"); + failed.push(name); + } + } + } + if !failed.is_empty() { + std::process::exit(1); + } +}