diff --git a/DESIGN.md b/DESIGN.md index 869f99b..e35adab 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -217,6 +217,19 @@ Tests: `bugsee-core/tests/multiprocess.rs` (real child processes: racing launche 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. +**Early-crash recovery on the launch path (#10).** A host that dies again within milliseconds of launch (a crash loop under systemd/Kubernetes) must not outrun the report of the crash before it. `Bugsee::launch_with` therefore ends with `Recorder::recover_on_launch_path()`: a bounded pass that turns the newest crashed sessions into queue entries (build, `before_send`/blacklist, assemble, enqueue, then discard the evidence) before launch returns. Upload stays asynchronous and resumes from the queue. A bundle is queued with plain file writes and a rename, no `fsync`, so it survives process death (the case here) but not power loss, as everywhere in §8. + +- **Threading contract.** The pass runs on a helper thread (`bugsee-launch-recovery`) and `launch` waits at most `launch_recovery_budget` (500 ms default, `LaunchOptions::launch_recovery_budget`; zero disables the pass) by polling a flag (never parking: a parker used before a `fork` is inherited by the child, and on macOS the child then traps in libdispatch when it parks). That is a hard upper bound whatever a `before_send` callback does: on timeout `launch` returns and the pass finishes in the background, so a host that holds its own lock while calling `launch` cannot deadlock against a callback that takes it. The callback can run on an SDK thread before `launch` returns; this is documented on `LaunchOptions::before_send`. +- **Ordering with the crash handlers.** The pass is the last thing `launch_with` does, after the native handler and the panic hook are installed and outside every lock, so a fault inside it is recorded with a stack against THIS session (its marker and `crash.info` path are already this generation's) and reported by the next launch. The previous session's evidence is untouched until its report is queued, and `bump_attempts` runs before processing, so a session that crashes the assemble path is abandoned after `MAX_RECOVERY_ATTEMPTS` launches instead of looping. On a re-launch within one process the native handler is replaced with one for the new generation before the pass runs; a fault in the pass is therefore the new generation's. +- **Worker coordination.** With the pass armed, the worker holds its own recovery back until the pass is done (bounded by budget plus 5 s, so a wedged callback cannot stall capture for good), then skips generations the pass claimed (`launch_attempted`, recorded only after winning the #14 claim). Claims keep the two exclusive in any interleaving. A host using `Recorder::launch` directly with a non-zero budget must call `recover_on_launch_path` or the worker waits that grace out; the core default budget is zero, i.e. no pass. +- **Bounds.** At most 8 sessions (`launch_recovery_max_sessions`), no new session after the budget (checked between sessions, so one huge session can overrun it, and the newest is always attempted), and at most 16 times that many candidates probed. Every session marker NAME is read (`readdir` order is not generation order, so a capped scan could miss the newest), sorted, and only the newest are probed; up to as many older sessions that already have an `.attempts` sidecar are included too and are processed FIRST (the worker's pass uses the same order), and a queue-cap refusal gives its attempt back (`undo_attempt`), so a session that faults during recovery itself, whose failure leaves a newer session behind each time, still reaches `MAX_RECOVERY_ATTEMPTS` and is abandoned. **Newest first**: the latest crash is the one most likely to belong to the build now running, and a crash loop makes only recent evidence reachable. What the pass leaves is recovered by the worker's unbounded pass, so nothing is lost; recovered reports may therefore upload newest first, which is harmless because the server deduplicates by signature. +- **Crash-loop escalation.** A fast loop never lets the uploader finish, so the server's `12004` verdict and the local blacklist are never reached. In the pass, a recovered crash whose signature set equals one already queued is dropped (the first copy is kept; once it has been delivered a recurrence is queued again), and the launch pass stops queuing recovered crashes when the queue holds 64 reports or 64 MiB, counting every queued report (`queue::admit_recovered`). A unique crash the cap refuses is NOT discarded: its evidence stays and it is left to the worker's pass, which does not cap (the cap bounds launch-path work, not total retention). Not done: a repeat counter on the kept report. The worker pass does not dedup or cap. +- **Scope.** A recovered crash carries no scope (email, labels, attributes) from the new process, on either path: the scope describes the new run, and merging it made the two paths disagree. Persisting scope per session is possible, not done. +- **Not covered.** Sessions of the legacy data directory are recovered by the worker only. A fork child (prefork worker boot loop) recovers on its worker, not on a launch path, so a prefork worker that dies before its worker thread finishes recovery leaves the evidence for a later launch (nothing is lost, it is reported late); the master's pass covers only the master's own launch. In memory mode with a recovery directory (#9), a crash that cannot be queued on disk is delivered from memory and its evidence stays, so a later launch reports it again (the backend deduplicates). +- **Measured** (`launch_recovery.rs::measure_launch_latency`, Apple silicon, synthetic log-only windows at the default 8 MiB / 60 s caps, a pass that waits for completion): empty directory about 0.1 to 0.3 ms; one crashed session about 7 ms release, 21 ms debug; eight identical sessions about 17 ms release, 43 ms debug. The eight-session figure undercounts assembly, since seven of them are dropped as duplicates before being assembled; scrubbing (#11) will add per-session cost. `Recorder::launch` itself is about 0.5 to 0.8 ms of that. + +Tests: `bugsee-core/tests/launch_recovery.rs` (a real abort-after-launch crash loop, time and count caps, a blocked callback, zero budget), `queue.rs` and `recovery.rs` unit tests. + **Generations:** each launch bumps `gen`; next-launch recovery reads the **previous** gen's parts before advancing. **Snapshot (no seal, no copy):** pin `[start,end]` (`end`=snapshot instant, `start`=`end−window`) + **hard-link current parts including the active one** into the report dir (`PartStore::snapshot_into`, after a `flush`). At export, read linked parts and include only entries with `start ≤ timestamp ≤ end` (monotonic order → early-stop at the tail; the self-terminating `RecordIter` cleanly ignores any torn trailing write on the active part). Bounds become `manifest.time.start/end`. Cost: 2 timestamps + O(#parts) `link` syscalls. Fallback to copy on FAT/exFAT/cross-volume (capability-probed once). Reflink (APFS `clonefile` / Linux `FICLONE`) a noted future tier. diff --git a/crates/bugsee-core/src/queue.rs b/crates/bugsee-core/src/queue.rs index e6298ff..b01b7a9 100644 --- a/crates/bugsee-core/src/queue.rs +++ b/crates/bugsee-core/src/queue.rs @@ -102,6 +102,63 @@ pub fn list_pending(data_dir: &Path) -> Vec { out } +/// Most reports the launch-path recovery will leave queued before it stops adding +/// recovered crashes (see [`admit_recovered`]). +pub const MAX_QUEUED_REPORTS: usize = 64; +/// Most queued bundle bytes before the same. +pub const MAX_QUEUED_BYTES: u64 = 64 << 20; + +/// Whether the launch-path recovery should queue a recovered crash. +#[derive(Debug, PartialEq, Eq)] +pub enum Admission { + Admit, + /// A queued report already carries exactly this signature set. + Duplicate, + /// The queue is at its entry or byte cap. + Full, +} + +/// Decide whether a recovered crash with `signatures` may be queued. +/// +/// A host in a crash loop re-reports the same crash on every start, and the loop +/// is faster than the uploader, so neither the server's duplicate verdict nor the +/// local blacklist is ever reached. Instead the first copy stays queued and +/// later copies with the identical, non-empty signature set are dropped; once it +/// has been delivered a recurrence is queued again. The queue is also capped +/// (entries and bytes): beyond it the caller keeps the crash's evidence for the +/// background pass rather than queue more on the launch path. Reports without signatures are never +/// treated as duplicates. +pub fn admit_recovered(data_dir: &Path, signatures: &[String]) -> Admission { + let wanted: BTreeSet<&String> = signatures.iter().collect(); + let mut count = 0usize; + let mut bytes = 0u64; + for report in list_pending(data_dir) { + count += 1; + bytes += std::fs::metadata(&report.bundle).map_or(0, |m| m.len()); + if count >= MAX_QUEUED_REPORTS || bytes >= MAX_QUEUED_BYTES { + return Admission::Full; + } + if wanted.is_empty() { + continue; + } + let queued: BTreeSet = std::fs::read(report.req_path()) + .ok() + .and_then(|b| serde_json::from_slice::(&b).ok()) + .and_then(|v| { + v.get("signatures").and_then(|s| s.as_array()).map(|a| { + a.iter() + .filter_map(|x| x.as_str().map(String::from)) + .collect() + }) + }) + .unwrap_or_default(); + if queued.len() == wanted.len() && wanted.iter().all(|s| queued.contains(*s)) { + return Admission::Duplicate; + } + } + Admission::Admit +} + /// Load a queued report's bundle bytes, request body, retry count, and the /// earliest next-attempt time (epoch ms). pub fn load(report: &QueuedReport) -> std::io::Result<(Vec, Vec, u32, i64)> { @@ -303,6 +360,40 @@ pub fn any_blacklisted(data_dir: &Path, sigs: &[String]) -> bool { mod tests { use super::*; + fn queue_report(data: &Path, sigs: &[&str]) { + let list: Vec = sigs.iter().map(|s| format!("\"{s}\"")).collect(); + enqueue( + data, + &AssembledReport { + bundle_name: format!("{}.bundle.zip", crate::util::random_hex(10)), + zip: b"z".to_vec(), + request_json: format!("{{\"signatures\":[{}]}}", list.join(",")).into_bytes(), + }, + ) + .unwrap(); + } + + #[test] + fn recovered_crashes_are_deduplicated_against_the_queue_and_capped() { + let data = std::env::temp_dir().join(format!("bugsee-qadm-{}", crate::util::random_hex(8))); + let sig = |s: &str| vec![s.to_string()]; + assert_eq!(admit_recovered(&data, &sig("a")), Admission::Admit); + queue_report(&data, &["a"]); + assert_eq!(admit_recovered(&data, &sig("a")), Admission::Duplicate); + assert_eq!(admit_recovered(&data, &sig("b")), Admission::Admit); + // A different signature SET is not a duplicate; neither is no signature. + assert_eq!( + admit_recovered(&data, &["a".to_string(), "b".to_string()]), + Admission::Admit + ); + assert_eq!(admit_recovered(&data, &[]), Admission::Admit); + for i in 1..MAX_QUEUED_REPORTS { + queue_report(&data, &[&format!("s{i}")]); + } + assert_eq!(admit_recovered(&data, &sig("new")), Admission::Full); + let _ = std::fs::remove_dir_all(&data); + } + #[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))); diff --git a/crates/bugsee-core/src/recovery.rs b/crates/bugsee-core/src/recovery.rs index fe50447..6bec2c8 100644 --- a/crates/bugsee-core/src/recovery.rs +++ b/crates/bugsee-core/src/recovery.rs @@ -61,6 +61,23 @@ pub struct RecoveredReport { /// because it launched earlier) has a live marker that is *not* a crash. Its /// marker is left in place for a future launch to recover once the peer is gone. pub fn find_pending(data_dir: &Path, current_generation: u64) -> Vec { + find_pending_bounded(data_dir, current_generation, usize::MAX) +} + +/// Like [`find_pending`], for the launch path: newest generation first, and +/// bounded in what it probes. Every session marker NAME is read (cheap, and the +/// newest cannot be known without reading them all: `readdir` order is not +/// generation order), sorted, and only then are files touched: the newest `limit` +/// pending sessions, plus up to `limit` older ones that already have an +/// `.attempts` sidecar. The latter keep a session that faults during recovery +/// itself moving toward `MAX_RECOVERY_ATTEMPTS`, instead of being starved by the +/// newer sessions each failed recovery leaves behind. At most `limit * 16` +/// candidates are probed. +pub fn find_pending_bounded( + data_dir: &Path, + current_generation: u64, + limit: usize, +) -> Vec { let mut pending = Vec::new(); let sessions = data_dir.join("sessions"); let read = match std::fs::read_dir(&sessions) { @@ -68,8 +85,8 @@ pub fn find_pending(data_dir: &Path, current_generation: u64) -> Vec return pending, }; let self_pid = std::process::id(); + let mut candidates: Vec = Vec::new(); for entry in read.flatten() { - let marker = entry.path(); let name = entry.file_name().to_string_lossy().into_owned(); let Some(gen_str) = name.strip_suffix(".alive") else { continue; @@ -77,9 +94,22 @@ pub fn find_pending(data_dir: &Path, current_generation: u64) -> Vec() else { continue; }; - if generation >= current_generation { + if generation < current_generation { + candidates.push(generation); + } + } + candidates.sort_unstable_by(|a, b| b.cmp(a)); + let mut retried = Vec::new(); + let (mut fresh, mut retries) = (0usize, 0usize); + for generation in candidates.into_iter().take(limit.saturating_mul(16)) { + if fresh >= limit && retries >= limit { + break; + } + let retrying = attempts_path(data_dir, generation).exists(); + if (retrying && retries >= limit) || (!retrying && fresh >= limit) { continue; } + let marker = sessions.join(format!("{generation}.alive")); if !session_is_crashed(data_dir, generation, &marker, self_pid) { continue; } @@ -87,15 +117,43 @@ pub fn find_pending(data_dir: &Path, current_generation: u64) -> Vec().ok()) + .unwrap_or(0); + if current <= 1 { + let _ = std::fs::remove_file(&path); + } else { + let _ = std::fs::write(&path, (current - 1).to_string()); } - pending } fn exists_opt(p: PathBuf) -> Option { @@ -818,6 +876,56 @@ mod tests { .collect() } + #[test] + fn the_bounded_scan_is_newest_first_and_capped() { + let dir = tmp(); + for g in [3u64, 9, 1, 7, 5] { + marker(&dir, g, "0"); + } + let got = |current, limit| -> Vec { + find_pending_bounded(&dir, current, limit) + .iter() + .map(|p| p.generation) + .collect() + }; + assert_eq!(got(u64::MAX, 100), vec![9, 7, 5, 3, 1]); + assert_eq!(got(u64::MAX, 2), vec![9, 7]); + // Only older generations than the caller's own are candidates. + assert_eq!(got(7, 100), vec![5, 3, 1]); + assert!(got(u64::MAX, 0).is_empty()); + } + + #[test] + fn an_undone_attempt_is_not_counted() { + let dir = tmp(); + marker(&dir, 1, "0"); + let pending = find_pending(&dir, u64::MAX).pop().unwrap(); + assert_eq!(bump_attempts(&dir, &pending), 1); + assert_eq!(bump_attempts(&dir, &pending), 2); + undo_attempt(&dir, &pending); + assert_eq!(bump_attempts(&dir, &pending), 2); + undo_attempt(&dir, &pending); + undo_attempt(&dir, &pending); + assert_eq!(bump_attempts(&dir, &pending), 1); + } + + #[test] + fn a_session_with_failed_attempts_is_not_starved_by_newer_ones() { + let dir = tmp(); + for g in 1..=6u64 { + marker(&dir, g, "0"); + } + // Session 2 was tried before (and the process died doing it). + std::fs::write(attempts_path(&dir, 2), "1").unwrap(); + let got: Vec = find_pending_bounded(&dir, u64::MAX, 2) + .iter() + .map(|p| p.generation) + .collect(); + // The one already tried comes first, so it advances even if the newest + // keeps killing the process. + assert_eq!(got, vec![2, 6, 5]); + } + #[test] fn a_free_lock_means_the_session_died_whatever_pid_its_marker_names() { let dir = tmp(); diff --git a/crates/bugsee-core/src/runtime.rs b/crates/bugsee-core/src/runtime.rs index 02d241a..1edafae 100644 --- a/crates/bugsee-core/src/runtime.rs +++ b/crates/bugsee-core/src/runtime.rs @@ -53,6 +53,10 @@ const UPLOAD_RETRY_CAP: u32 = 60; /// Maximum next-launch recovery attempts for one pending session before it is /// abandoned (guards against a corrupt-parts poison pill re-running every launch). const MAX_RECOVERY_ATTEMPTS: u32 = 3; +/// Default time `launch` may spend queueing the previous run's crashes. +pub const DEFAULT_LAUNCH_RECOVERY_BUDGET: Duration = Duration::from_millis(500); +/// Default cap on crashed sessions queued on the launch path. +pub const DEFAULT_LAUNCH_RECOVERY_SESSIONS: usize = 8; /// Default in-flight capture-entry cap (drop-newest beyond this). const DEFAULT_MAX_QUEUED: usize = 8192; /// Default in-flight *report* cap. Reports are far heavier than capture entries @@ -179,6 +183,16 @@ pub struct RecorderConfig { /// crashed sessions and deliver its queue. For a host layer that moved its /// default location and must not strand reports left at the old one. pub legacy_data_dir: Option, + /// Time [`Recorder::recover_on_launch_path`] waits for the launch-path pass + /// (a hard bound; the pass then finishes in the background) and after which + /// the pass starts no further session. Zero, the default here, disables the + /// pass: all recovery happens on the worker thread. The facade sets + /// [`DEFAULT_LAUNCH_RECOVERY_BUDGET`]. With a non-zero budget the worker holds + /// its own recovery back until the pass has run, so the host layer must call + /// [`Recorder::recover_on_launch_path`] (or the worker waits out a few seconds). + pub launch_recovery_budget: Duration, + /// Most crashed sessions the launch-path pass recovers. + pub launch_recovery_max_sessions: usize, /// More existing directories to recover from and drain once, like /// `legacy_data_dir`: trusted trees that could not take new files (a full or /// read-only default) even though a later default was writable. @@ -205,6 +219,8 @@ impl RecorderConfig { max_queued_entries: DEFAULT_MAX_QUEUED, memory_only: None, legacy_data_dir: None, + launch_recovery_budget: Duration::ZERO, + launch_recovery_max_sessions: DEFAULT_LAUNCH_RECOVERY_SESSIONS, extra_data_dirs: Vec::new(), } } @@ -262,6 +278,31 @@ struct Shared { /// Set the first time a disk write fails while running in disk mode: the /// recorder is no longer fully durable (see [`Recorder::is_degraded`]). degraded: AtomicBool, + /// The launch-path recovery pass to run, taken by + /// [`Recorder::recover_on_launch_path`]; `None` when not armed. + launch_plan: ForkLock>, + /// Generations the launch-path pass claimed, which the worker's pass skips: + /// one it could only hold in memory still has its evidence on disk, and doing + /// it again would report it twice. + launch_attempted: ForkLock>, + /// Whether the worker holds its own recovery until the launch-path pass is + /// over (it is armed and the pass has not finished). + launch_armed: AtomicBool, + launch_done: AtomicBool, + /// Set when the launch pass left a session for the worker (queue cap), so a + /// worker that already did its one pass runs another. + launch_deferred: AtomicBool, + launch_wait_ms: AtomicU64, +} + +/// What the launch-path recovery pass needs, fixed at launch. +struct LaunchPlan { + dir: PathBuf, + generation: u64, + budget: Duration, + max_sessions: usize, + /// This session's marker, whose heartbeat the pass keeps fresh. + heartbeat_of: Option, } /// The recorder's own locks join the `fork()` lock protocol, so a forked child @@ -274,8 +315,12 @@ impl ForkLockable for Shared { self.mem_queue.fork_lock(); self.mem_blacklist.fork_lock(); self.delivered_stuck.fork_lock(); + self.launch_plan.fork_lock(); + self.launch_attempted.fork_lock(); } fn fork_unlock(&self) { + self.launch_attempted.fork_unlock(); + self.launch_plan.fork_unlock(); self.delivered_stuck.fork_unlock(); self.mem_blacklist.fork_unlock(); self.mem_queue.fork_unlock(); @@ -343,6 +388,11 @@ struct Pipeline { impl Recorder { /// Start the workers and begin capturing. Recovers any prior session that /// ended abnormally, queueing it for delivery. + /// + /// Recovery of the newest crashed sessions can be made to finish on the + /// launch path with [`Recorder::recover_on_launch_path`], which the host layer + /// calls after installing its crash handlers (see there). Whatever it does not + /// cover is recovered by the worker thread. pub fn launch(config: RecorderConfig, transport: Arc) -> std::io::Result { let env = Environment::detect_with( &config.sdk_version, @@ -371,6 +421,12 @@ impl Recorder { mem_blacklist: ForkLock::new(BTreeSet::new()), delivered_stuck: ForkLock::new(BTreeSet::new()), degraded: AtomicBool::new(false), + launch_plan: ForkLock::new(None), + launch_attempted: ForkLock::new(BTreeSet::new()), + launch_armed: AtomicBool::new(false), + launch_done: AtomicBool::new(false), + launch_deferred: AtomicBool::new(false), + launch_wait_ms: AtomicU64::new(0), }); crate::fork::install_fork_hook(); @@ -408,6 +464,24 @@ impl Recorder { // recovering them only reads and deletes. So an existing directory is // still drained and recovered from in memory mode. let drain_dir = (session.is_some() || data_dir.is_dir()).then_some(data_dir.as_path()); + // Arm the launch-path recovery pass (issue #10). It does not run here: + // the host layer runs it with `recover_on_launch_path` once its crash + // handlers are installed, so a fault inside the pass is captured with a + // stack. Until then the worker holds its own recovery back. + if let (Some(dir), false) = (drain_dir, config.launch_recovery_budget.is_zero()) { + *shared.launch_plan.lock() = Some(LaunchPlan { + dir: dir.to_path_buf(), + generation: session.as_ref().map_or(u64::MAX, Session::generation), + budget: config.launch_recovery_budget, + max_sessions: config.launch_recovery_max_sessions, + heartbeat_of: session + .as_ref() + .map(|s| Session::marker_path(dir, s.generation())), + }); + let wait_ms = config.launch_recovery_budget.as_millis() as u64; + shared.launch_wait_ms.store(wait_ms, Ordering::SeqCst); + shared.launch_armed.store(true, Ordering::SeqCst); + } let pipeline = spawn_pipeline( &shared, store.is_some().then_some(data_dir.as_path()), @@ -536,6 +610,24 @@ impl Recorder { Ok(()) } + /// Run the launch-path recovery pass (issue #10): turn the newest crashed + /// sessions of earlier runs into queued reports before returning. + /// + /// Call it once, after the crash handlers are installed (so a fault inside the + /// pass is captured with a stack, attributed to this session) and without + /// holding any lock a `before_send` hook might take. The pass runs on a helper + /// thread and this waits at most [`RecorderConfig::launch_recovery_budget`] + /// for it: on timeout it returns and the pass finishes in the background, so + /// `launch` has a hard upper bound whatever the host's hook does. `before_send` + /// can therefore run on an SDK thread before this returns. + /// + /// Does nothing when the budget is zero or it already ran. A recorder whose + /// budget is non-zero holds its worker's own recovery back until this has + /// finished, for at most the budget plus a few seconds. + pub fn recover_on_launch_path(&self) { + self.handle().recover_on_launch_path(); + } + /// Enqueue an entry. Never blocks; drops (newest) under back-pressure or if /// the worker is gone. pub fn capture(&self, entry: CaptureEntry) { @@ -659,6 +751,9 @@ impl Recorder { impl Drop for Recorder { fn drop(&mut self) { + // A worker still holding its recovery for the launch-path pass must not + // make the stop below time out. + self.shared.launch_done.store(true, Ordering::SeqCst); // Signal both threads; JOIN only if they acknowledge within the timeout. // A thread wedged in a host callback with no timeout is DETACHED (its // JoinHandle dropped un-joined) rather than hanging the stopping thread @@ -724,6 +819,36 @@ pub struct RecorderHandle { } impl RecorderHandle { + /// See [`Recorder::recover_on_launch_path`]. + pub fn recover_on_launch_path(&self) { + let Some(plan) = self.shared.launch_plan.lock().take() else { + return; + }; + let budget = plan.budget; + let shared = Arc::clone(&self.shared); + let spawned = std::thread::Builder::new() + .name("bugsee-launch-recovery".into()) + .spawn(move || { + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + run_launch_recovery(&shared, &plan); + })); + shared.launch_done.store(true, Ordering::SeqCst); + }); + match spawned { + // Poll, never park: this thread's parker (a libdispatch semaphore on + // macOS) would then be inherited in whatever state it was left by a + // later `fork`, whose child traps in libdispatch when it parks. + Ok(_) => { + let deadline = Instant::now() + budget; + while !self.shared.launch_done.load(Ordering::SeqCst) && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(1)); + } + } + // No thread: the worker's own pass recovers everything. + Err(_) => self.shared.launch_done.store(true, Ordering::SeqCst), + } + } + /// Enqueue a capture entry (bounded, drop-newest under back-pressure). pub fn capture(&self, entry: CaptureEntry) { capture_into(&self.shared, &self.tx, entry); @@ -919,6 +1044,7 @@ fn spawn_pipeline( let worker_shared = Arc::clone(shared); let worker_data_dir = data_dir.map(Path::to_path_buf); let worker_drain_dir = drain_dir.map(Path::to_path_buf); + let worker_recover_dir = worker_drain_dir.clone(); 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. @@ -953,7 +1079,16 @@ 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, beat_marker.as_deref()); + if fresh_launch { + wait_for_launch_pass(&worker_shared); + } + run_recovery( + &worker_shared, + dir, + generation, + beat_marker.as_deref(), + fresh_launch, + ); } // Best effort, once: sessions a previous default location left // behind. No generation of ours lives there, so none is skipped. @@ -963,7 +1098,7 @@ fn spawn_pipeline( worker_shared.mem_blacklist.lock().extend(known); } queue::gc_orphans(dir); - run_recovery(&worker_shared, dir, u64::MAX, beat_marker.as_deref()); + run_recovery(&worker_shared, dir, u64::MAX, beat_marker.as_deref(), false); } })); legacy_ready.store(true, Ordering::SeqCst); @@ -978,6 +1113,8 @@ fn spawn_pipeline( caps, sampler, worker_upload_tx, + generation, + worker_recover_dir, ); })?; @@ -992,11 +1129,16 @@ 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). +/// +/// `heartbeat_of` is this session's marker, touched as the pass makes progress so a +/// peer does not take a long recovery for a dead session. With `skip_launch_pass`, +/// generations the launch-path pass already claimed are left alone. fn run_recovery( shared: &Shared, data_dir: &Path, current_generation: u64, heartbeat_of: Option<&Path>, + skip_launch_pass: bool, ) { let beat = || { if let Some(marker) = heartbeat_of { @@ -1006,32 +1148,109 @@ fn run_recovery( 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 { - continue; - }; - // Count the attempt *before* processing: a corrupt session that panics - // the build/assemble path is a poison pill that would otherwise re-run — - // and abort the whole loop — on every launch. After the cap, abandon it. - let attempts = recovery::bump_attempts(data_dir, &pending); - if attempts > MAX_RECOVERY_ATTEMPTS { - recovery::discard(data_dir, &pending); + if skip_launch_pass && shared.launch_attempted.lock().contains(&pending.generation) { continue; } - // Isolate each pending so one corrupt session cannot block recovering the - // rest (a panic here would otherwise unwind the entire loop). - let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - recover_one(shared, data_dir, &pending); - })); + 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); } +/// Claim one pending session and recover it. Returns whether this process won the +/// claim: a session another process holds is skipped. `launch_pass` applies the +/// launch path's dedup and queue cap to the recovered report. +fn recover_claimed( + shared: &Shared, + data_dir: &Path, + pending: &recovery::PendingSession, + launch_pass: bool, +) -> bool { + // 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 { + return false; + }; + if launch_pass { + shared.launch_attempted.lock().insert(pending.generation); + } + // Count the attempt *before* processing: a corrupt session that panics or + // faults in the build/assemble path is a poison pill that would otherwise + // re-run on every launch. After the cap, abandon it. + let attempts = recovery::bump_attempts(data_dir, pending); + if attempts > MAX_RECOVERY_ATTEMPTS { + recovery::discard(data_dir, pending); + return true; + } + // Isolate each pending so one corrupt session cannot block recovering the + // rest (a panic here would otherwise unwind the entire loop). + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + recover_one(shared, data_dir, pending, launch_pass); + })); + true +} + +/// How long the worker holds its recovery back for the launch-path pass beyond +/// the pass's own budget. The pass runs host code (`before_send`) that can block; +/// the worker must not stall capture behind it for good. +const LAUNCH_PASS_GRACE: Duration = Duration::from_secs(5); + +/// Hold the worker's recovery until the launch-path pass has finished, so the two +/// do not interleave (claims would keep them exclusive, but the pass's contract is +/// that it handles the newest sessions first). A no-op when the pass is not armed. +/// Bounded: if the host layer never runs the pass, or the pass wedges, the worker +/// goes ahead. +fn wait_for_launch_pass(shared: &Shared) { + if !shared.launch_armed.load(Ordering::SeqCst) { + return; + } + 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 { + std::thread::sleep(Duration::from_millis(2)); + } +} + +/// The launch-path recovery pass (issue #10): turn the newest crashed sessions +/// into durable queue entries BEFORE `launch` returns, so a host that dies again +/// within milliseconds (a crash loop under a supervisor) cannot outrun the report +/// of the crash before it. Only the queueing is synchronous; upload stays on the +/// uploader thread. +/// +/// Bounded: at most `max_sessions` (plus as many already-attempted ones, so a +/// session that faults in recovery still reaches `MAX_RECOVERY_ATTEMPTS`) are +/// recovered, and no new session is started once `budget` has elapsed. The budget is checked between sessions, so one session +/// can overrun it; the newest session is always attempted whatever the budget, +/// since it is the one a crash loop is about to make unreachable. Newest first: +/// the latest crash is the one most likely to be on the build now running, and +/// what is left is not lost, since the worker's unbounded pass recovers the rest. +/// +/// Records the generations it claimed in `launch_attempted`, for the worker's pass +/// to skip. +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) { + if let Some(marker) = &plan.heartbeat_of { + touch_heartbeat(marker); + } + if claimed > 0 && started.elapsed() >= plan.budget { + break; + } + if recover_claimed(shared, &plan.dir, &pending, true) { + claimed += 1; + } + } +} + /// Build, filter, assemble and enqueue one pending session's crash report, /// discarding its on-disk evidence only once it is safely queued. -fn recover_one(shared: &Shared, data_dir: &Path, pending: &recovery::PendingSession) { +fn recover_one( + shared: &Shared, + data_dir: &Path, + pending: &recovery::PendingSession, + launch_pass: bool, +) { let span = export::report_span(&pending.parts_dir).ok().flatten(); let now = epoch_ms(); let (start, end) = span.unwrap_or((now, now)); @@ -1049,7 +1268,31 @@ fn recover_one(shared: &Shared, data_dir: &Path, pending: &recovery::PendingSess recovery::discard(data_dir, pending); return; } - merge_scope(shared, &mut built.meta); + // A recovered crash carries no scope (email/labels/attributes) from THIS + // process: they describe the new run, not the one that crashed, and the + // launch-path and worker passes must agree on it. + if launch_pass { + match queue::admit_recovered(data_dir, &built.meta.signatures) { + queue::Admission::Admit => {} + // A crash loop reports the same crash on every start: the first one + // is already queued, and another copy adds nothing the server keeps. + queue::Admission::Duplicate => { + recovery::discard(data_dir, pending); + return; + } + // A unique crash the queue has no room for on the launch path: + // keep its evidence and leave it to the worker's pass, which does not + // cap, rather than destroy it. + queue::Admission::Full => { + shared.launch_attempted.lock().remove(&pending.generation); + // Not a failed attempt: a supervisor crash loop would otherwise + // burn the session's attempts on queue refusals and abandon it. + recovery::undo_attempt(data_dir, pending); + shared.launch_deferred.store(true, Ordering::SeqCst); + return; + } + } + } let window = TimeWindow { start, end }; let assembled = reporting::assemble_with_extras( @@ -1072,6 +1315,13 @@ fn recover_one(shared: &Shared, data_dir: &Path, pending: &recovery::PendingSess true } else { hold_in_memory(shared, assembled); + // A failed WRITE says nothing about the evidence (the volume is + // full or read-only), unlike an assemble failure, so it must not + // use up the session's attempts: a supervisor crash loop would + // kill the process before the memory copy is sent, and three such + // launches would discard a crash that was never delivered. The + // launch pass keeps its claim so the worker does not hold it twice. + recovery::undo_attempt(data_dir, pending); false } } @@ -1111,6 +1361,8 @@ fn worker_loop( caps: WindowCaps, mut sampler: Option>, upload_tx: Sender, + generation: u64, + recover_dir: Option, ) { // Rotation/eviction/telemetry must run on a wall-clock cadence, NOT only when // `recv_timeout` times out. Under sustained capture traffic every `recv` @@ -1206,6 +1458,20 @@ fn worker_loop( rotate_tick(store.as_mut(), &shared, &mut sampler); next_rotate = Instant::now() + rotate_interval; } + // The launch-path pass left a session for us (queue cap) after our one + // pass had already run, because it outlived the wait: take it now. + // `recover_dir` is the tree startup recovery used: it exists in memory + // mode too, where `data_dir` (the capture tree) does not. + if let Some(dir) = &recover_dir { + if shared.launch_done.load(Ordering::SeqCst) + && shared.launch_deferred.swap(false, Ordering::SeqCst) + { + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + run_recovery(&shared, dir, generation, marker.as_deref(), true); + })); + let _ = upload_tx.send(UploadMsg::Wake); + } + } // Independent of the rotation interval (which a host may set large). if Instant::now() >= next_heartbeat { if let Some(marker) = &marker { diff --git a/crates/bugsee-core/tests/launch_recovery.rs b/crates/bugsee-core/tests/launch_recovery.rs new file mode 100644 index 0000000..387c672 --- /dev/null +++ b/crates/bugsee-core/tests/launch_recovery.rs @@ -0,0 +1,391 @@ +// +// launch_recovery.rs +// bugsee-core +// +// Copyright © 2026 Bugsee. All rights reserved. +// +//! Early-crash recovery finishes on the launch path (issue #10): by the time +//! `recover_on_launch_path` returns, the previous run's crash is a queue entry on +//! disk, so a host that crashes again right away (a crash loop) cannot outrun its +//! report. +//! +//! The crash loop is real: this test binary re-executes itself as children that +//! launch the SDK and `abort()` immediately. + +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use bugsee_core::capture::record; +use bugsee_core::transport::{Transport, TransportError}; +use bugsee_core::{queue, MockTransport, Recorder, RecorderConfig}; + +const ROLE: &str = "BUGSEE_LR_ROLE"; +const DIR: &str = "BUGSEE_LR_DIR"; +const OUT: &str = "BUGSEE_LR_OUT"; +const LAUNCH_THREAD: &str = "bugsee-launch-recovery"; + +/// A transport that never delivers anything, so the queue keeps its entries. +struct Offline; +impl Transport for Offline { + fn register_session(&self, _: &str, _: &[u8]) -> Result { + Err(TransportError::Transient("offline".into())) + } + fn create_issue(&self, _: &str, _: Option<&str>, _: &[u8]) -> Result { + Err(TransportError::Transient("offline".into())) + } + fn upload_bundle(&self, _: &str, _: &[u8]) -> Result<(), TransportError> { + Err(TransportError::Transient("offline".into())) + } +} + +fn config(dir: &Path, budget: Duration) -> RecorderConfig { + let mut config = RecorderConfig::new(dir, "tok"); + config.launch_recovery_budget = budget; + config +} + +/// Child entry point. A no-op in a normal test run. +#[test] +fn child_entry() { + if std::env::var(ROLE).as_deref() != Ok("crash_after_launch") { + return; + } + let dir = PathBuf::from(std::env::var(DIR).unwrap()); + let out = PathBuf::from(std::env::var(OUT).unwrap()); + let rec = + Recorder::launch(config(&dir, Duration::from_millis(500)), Arc::new(Offline)).unwrap(); + rec.recover_on_launch_path(); + // The instant the launch path is done: record what is queued, then die the + // way a crash does, with no destructor and no flush. + std::fs::write(out, queue::list_pending(&dir).len().to_string()).unwrap(); + std::process::abort(); +} + +struct TempDir(PathBuf); +impl TempDir { + fn new() -> Self { + let p = + std::env::temp_dir().join(format!("bugsee-lr-{}", 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 seed_crashed(dir: &Path, generations: std::ops::RangeInclusive) { + let sessions = dir.join("sessions"); + std::fs::create_dir_all(&sessions).unwrap(); + for g in generations.clone() { + // "0" is the portable "owner gone" sentinel. + std::fs::write(sessions.join(format!("{g}.alive")), "0").unwrap(); + std::fs::create_dir_all(dir.join("parts").join(g.to_string())).unwrap(); + } + std::fs::write(dir.join("gen"), generations.end().to_string()).unwrap(); +} + +/// A `before_send` that records the name of the thread it runs on and sleeps. +fn recording_hook( + threads: &Arc>>, + sleep: Duration, +) -> bugsee_core::runtime::BeforeSend { + let seen = Arc::clone(threads); + Box::new(move |_meta| { + let t = std::thread::current(); + seen.lock() + .unwrap() + .push(t.name().unwrap_or("launch").to_string()); + std::thread::sleep(sleep); + true + }) +} + +fn on_launch_thread(threads: &Arc>>) -> usize { + threads + .lock() + .unwrap() + .iter() + .filter(|n| *n == LAUNCH_THREAD) + .count() +} + +#[test] +fn the_previous_crash_is_queued_before_the_launch_path_returns() { + let tmp = TempDir::new(); + let dir = tmp.0.join("data"); + seed_crashed(&dir, 1..=1); + + let rec = + Recorder::launch(config(&dir, Duration::from_millis(500)), Arc::new(Offline)).unwrap(); + rec.recover_on_launch_path(); + // No flush, no waiting: it is on disk the moment the launch path returns. + assert_eq!(queue::list_pending(&dir).len(), 1); + drop(rec); +} + +/// Time cap: a budget shorter than one session lets exactly the newest session +/// through on the launch path (it is always attempted); the rest follow on the +/// worker. And the launch path returns at the budget, not when the hook does. +#[test] +fn a_short_budget_recovers_one_session_on_the_launch_path() { + let tmp = TempDir::new(); + let dir = tmp.0.join("data"); + seed_crashed(&dir, 1..=5); + + let threads: Arc>> = Arc::default(); + let mut config = config(&dir, Duration::from_millis(50)); + config.before_send = Some(recording_hook(&threads, Duration::from_millis(400))); + let transport = Arc::new(MockTransport::default()); + let started = Instant::now(); + let rec = Recorder::launch(config, transport.clone()).unwrap(); + rec.recover_on_launch_path(); + assert!( + started.elapsed() < Duration::from_millis(300), + "the launch path waited for the hook: {:?}", + started.elapsed() + ); + + assert!(rec.flush(Duration::from_secs(20))); + assert_eq!( + on_launch_thread(&threads), + 1, + "{:?}", + threads.lock().unwrap() + ); + assert_eq!(threads.lock().unwrap().len(), 5, "each session once"); + assert_eq!(transport.uploaded_bundles.lock().unwrap().len(), 5); +} + +/// Count cap: with time to spare, `launch_recovery_max_sessions` limits the pass. +#[test] +fn the_session_cap_limits_the_launch_path() { + let tmp = TempDir::new(); + let dir = tmp.0.join("data"); + seed_crashed(&dir, 1..=5); + + let threads: Arc>> = Arc::default(); + let mut config = config(&dir, Duration::from_secs(10)); + config.launch_recovery_max_sessions = 2; + config.before_send = Some(recording_hook(&threads, Duration::ZERO)); + let transport = Arc::new(MockTransport::default()); + let rec = Recorder::launch(config, transport.clone()).unwrap(); + rec.recover_on_launch_path(); + // The pass was quick, so it is over: exactly the cap ran on it. + assert_eq!(on_launch_thread(&threads), 2); + + assert!(rec.flush(Duration::from_secs(20))); + assert_eq!(threads.lock().unwrap().len(), 5, "each session once"); + // The pass's second session repeats the first one's crash and is dropped. + assert_eq!(transport.uploaded_bundles.lock().unwrap().len(), 4); +} + +/// A unique crash the launch pass has no queue room for is not destroyed: the +/// worker's (uncapped) pass queues it. +#[test] +fn a_full_queue_defers_a_unique_crash_to_the_worker_instead_of_discarding_it() { + let tmp = TempDir::new(); + let dir = tmp.0.join("data"); + seed_crashed(&dir, 1..=1); + for i in 0..queue::MAX_QUEUED_REPORTS { + queue::enqueue( + &dir, + &bugsee_core::AssembledReport { + bundle_name: format!("{}.bundle.zip", bugsee_core::util::random_hex(10)), + zip: b"z".to_vec(), + request_json: format!("{{\"signatures\":[\"other-{i}\"]}}").into_bytes(), + }, + ) + .unwrap(); + } + + let rec = + Recorder::launch(config(&dir, Duration::from_millis(500)), Arc::new(Offline)).unwrap(); + rec.recover_on_launch_path(); + // The launch pass refused it (the worker may already be queueing it, so + // only the outcome is asserted): it ends up queued, not discarded, and + // without having used up its recovery attempts. + // Observable condition, generous timeout: a loaded CI machine can delay the + // helper thread, the worker's wait (budget plus a 5 s grace) and its pass. + let deadline = Instant::now() + Duration::from_secs(60); + while queue::list_pending(&dir).len() <= queue::MAX_QUEUED_REPORTS { + assert!( + Instant::now() < deadline, + "the worker never recovered it: {} queued, session marker present: {}", + queue::list_pending(&dir).len(), + dir.join("sessions").join("1.alive").exists() + ); + std::thread::sleep(Duration::from_millis(20)); + } + drop(rec); +} + +/// A hook that never returns must not hold `launch` past the budget. +#[test] +fn a_blocked_before_send_cannot_stall_the_launch_path() { + let tmp = TempDir::new(); + let dir = tmp.0.join("data"); + seed_crashed(&dir, 1..=1); + + let release = Arc::new(AtomicBool::new(false)); + let gate = Arc::clone(&release); + let mut config = config(&dir, Duration::from_millis(100)); + config.before_send = Some(Box::new(move |_meta| { + while !gate.load(Ordering::SeqCst) { + std::thread::sleep(Duration::from_millis(5)); + } + true + })); + let transport = Arc::new(MockTransport::default()); + let started = Instant::now(); + let rec = Recorder::launch(config, transport.clone()).unwrap(); + rec.recover_on_launch_path(); + assert!( + started.elapsed() < Duration::from_secs(2), + "launch waited on a blocked hook: {:?}", + started.elapsed() + ); + + // Unblocked, the pass finishes in the background and the report goes out. + release.store(true, Ordering::SeqCst); + assert!(rec.flush(Duration::from_secs(20))); + assert_eq!(transport.uploaded_bundles.lock().unwrap().len(), 1); +} + +/// Without a launch-path pass (the core default), the worker does it all. +#[test] +fn a_zero_budget_leaves_recovery_to_the_worker() { + let tmp = TempDir::new(); + let dir = tmp.0.join("data"); + seed_crashed(&dir, 1..=2); + + let threads: Arc>> = Arc::default(); + let mut config = config(&dir, Duration::ZERO); + config.before_send = Some(recording_hook(&threads, Duration::ZERO)); + let transport = Arc::new(MockTransport::default()); + let rec = Recorder::launch(config, transport.clone()).unwrap(); + rec.recover_on_launch_path(); + assert!(rec.flush(Duration::from_secs(10))); + assert_eq!(transport.uploaded_bundles.lock().unwrap().len(), 2); + assert!(threads + .lock() + .unwrap() + .iter() + .all(|n| n == "bugsee-capture")); +} + +/// The product scenario: the host crashes right after every launch. A crash loop +/// reports the same crash each time, so the first copy is queued and later copies +/// are not stacked behind it; and each launch already holds the previous crash in +/// the queue on return, with no async worker involved. +#[test] +fn a_crash_loop_queues_the_first_crash_before_each_relaunch_and_collapses_repeats() { + const CRASHES: usize = 5; + let tmp = TempDir::new(); + let dir = tmp.0.join("data"); + std::fs::create_dir_all(&dir).unwrap(); + + for i in 0..CRASHES { + let out = tmp.0.join(format!("queued-{i}")); + let status = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "child_entry", "--nocapture"]) + .env(ROLE, "crash_after_launch") + .env(DIR, &dir) + .env(OUT, &out) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .unwrap(); + assert!(!status.success(), "child {i} should have crashed"); + let queued: usize = std::fs::read_to_string(&out) + .unwrap_or_else(|_| panic!("child {i} never got past launch")) + .parse() + .unwrap(); + // Child 0 has nothing to recover. Every later child found the crash + // before it already queued, and identical repeats are not queued again. + assert_eq!(queued, usize::from(i > 0), "child {i} saw {queued} queued"); + } + + // Another launch, offline so nothing is delivered behind the test's back, + // recovers the last child's crash: a repeat of the queued one, so dropped. + let rec = + Recorder::launch(config(&dir, Duration::from_millis(500)), Arc::new(Offline)).unwrap(); + rec.recover_on_launch_path(); + assert_eq!(queue::list_pending(&dir).len(), 1); + drop(rec); + + // A healthy run delivers the queued crash, once. + let transport = Arc::new(MockTransport::default()); + let rec = + Recorder::launch(config(&dir, Duration::from_millis(500)), transport.clone()).unwrap(); + rec.recover_on_launch_path(); + assert!(rec.flush(Duration::from_secs(20))); + assert_eq!(transport.uploaded_bundles.lock().unwrap().len(), 1); + assert!(queue::list_pending(&dir).is_empty()); + drop(rec); + + // And not again. + let again = Arc::new(MockTransport::default()); + let rec = Recorder::launch(config(&dir, Duration::from_millis(500)), again.clone()).unwrap(); + rec.recover_on_launch_path(); + assert!(rec.flush(Duration::from_secs(5))); + assert!(again.uploaded_bundles.lock().unwrap().is_empty()); +} + +/// Launch latency of the pass, for DESIGN.md: run with +/// `cargo test -p bugsee-core --release --test launch_recovery -- --ignored --nocapture`. +#[test] +#[ignore] +fn measure_launch_latency() { + fn seed_with_window(dir: &Path, generations: std::ops::RangeInclusive, bytes: usize) { + seed_crashed(dir, generations.clone()); + for g in generations { + let part = dir.join("parts").join(g.to_string()).join("0"); + std::fs::create_dir_all(&part).unwrap(); + let mut buf = Vec::new(); + let mut ts = 1_000i64; + while buf.len() < bytes { + let payload = format!( + r#"{{"timestamp":{ts},"level":1,"source":2,"message":"{}"}}"#, + "x".repeat(200) + ); + record::frame(ts, payload.as_bytes(), &mut buf); + ts += 1; + } + std::fs::write(part.join("log.part"), &buf).unwrap(); + } + } + for (label, sessions, bytes) in [ + ("empty directory", 0u64, 0usize), + ("1 session, 8 MiB window", 1, 8 << 20), + ("8 sessions, 8 MiB windows", 8, 8 << 20), + ] { + let tmp = TempDir::new(); + let dir = tmp.0.join("data"); + std::fs::create_dir_all(&dir).unwrap(); + if sessions > 0 { + seed_with_window(&dir, 1..=sessions, bytes); + } + let mut config = config(&dir, Duration::from_secs(30)); + config.caps = bugsee_core::capture::WindowCaps { + max_window_ms: 60_000, + max_bytes: 8 << 20, + max_events: 100_000, + }; + let t0 = Instant::now(); + let rec = Recorder::launch(config, Arc::new(Offline)).unwrap(); + let launched = t0.elapsed(); + let t1 = Instant::now(); + rec.recover_on_launch_path(); + println!( + "{label}: Recorder::launch {launched:?}, recover_on_launch_path {:?}, queued {}", + t1.elapsed(), + queue::list_pending(&dir).len() + ); + } +} diff --git a/crates/bugsee/src/api.rs b/crates/bugsee/src/api.rs index ccf1f48..66e67d9 100644 --- a/crates/bugsee/src/api.rs +++ b/crates/bugsee/src/api.rs @@ -141,6 +141,7 @@ impl Bugsee { config.on_report_dropped = options.on_report_dropped; config.before_breadcrumb = options.before_breadcrumb; config.sample_rate = options.sample_rate; + config.launch_recovery_budget = options.launch_recovery_budget; config.app_identity = bugsee_core::model::environment::AppIdentity { package_id: options.app_package_id.clone(), version: options.app_version.clone(), @@ -178,6 +179,7 @@ impl Bugsee { #[cfg(feature = "panic")] let panic_info_path = recorder.panic_info_path(); + let recovery_handle = recorder.handle(); // Install the native crash handler pointing at this generation's marker. // Replace under a brief lock, then drop the OLD handler outside it. @@ -230,6 +232,13 @@ impl Bugsee { bugsee_panic::set_report_from_hook(options.report_panics_from_hook); } + // Last, once the native handler and the panic hook are in place: a fault + // inside the recovery pass is then recorded, with a stack, against THIS + // session (the previous crash's evidence stays on disk and is retried, + // attempt-counted). Run outside every lock: it can call the host's + // `before_send` on a helper thread, and waits at most the launch budget. + recovery_handle.recover_on_launch_path(); + Ok(LaunchGuard { _private: () }) } diff --git a/crates/bugsee/src/options.rs b/crates/bugsee/src/options.rs index 914f50d..c3d69f6 100644 --- a/crates/bugsee/src/options.rs +++ b/crates/bugsee/src/options.rs @@ -39,6 +39,7 @@ pub struct LaunchOptions { pub(crate) before_send: Option, pub(crate) before_breadcrumb: Option, pub(crate) sample_rate: f64, + pub(crate) launch_recovery_budget: Duration, } impl LaunchOptions { @@ -63,6 +64,7 @@ impl LaunchOptions { before_send: None, before_breadcrumb: None, sample_rate: 1.0, + launch_recovery_budget: bugsee_core::runtime::DEFAULT_LAUNCH_RECOVERY_BUDGET, } } @@ -76,6 +78,15 @@ impl LaunchOptions { /// Register a callback run on every report before delivery. Mutate the /// metadata (summary/severity/labels/attributes/…) in place; return `false` /// to drop the report. + /// + /// Threading: the callback runs on an SDK thread (the capture worker, or the + /// launch-recovery helper thread for a crash recovered during `launch`), and + /// that can be before `launch` has returned. Do not assume it runs on the + /// thread that launched, and do not make `launch` wait on a lock the callback + /// needs: `launch` itself never waits for the callback longer than + /// [`launch_recovery_budget`](Self::launch_recovery_budget), but a callback + /// that blocks forever stalls report processing. A recovered crash carries no + /// scope (email, labels, attributes) from the new process. pub fn before_send( mut self, f: impl Fn(&mut ReportMeta) -> bool + Send + Sync + 'static, @@ -212,6 +223,36 @@ impl LaunchOptions { self } + /// How long `launch` waits for the previous run's crash reports to be + /// queued. Default 500 ms + /// ([`DEFAULT_LAUNCH_RECOVERY_BUDGET`](bugsee_core::runtime::DEFAULT_LAUNCH_RECOVERY_BUDGET)). + /// + /// A host that crashes again right after launch (a crash loop under a + /// supervisor) still gets the previous crash reported, because it is put in + /// the on-disk queue on this path (it survives process death, not power + /// loss); only the upload is asynchronous. The work runs on a helper thread + /// and `launch` waits at most this long, so this is a hard upper bound on + /// what recovery adds to `launch`, whatever a `before_send` callback does; if + /// it runs out the pass finishes in the background. The newest crashed + /// session is attempted first, at most + /// [`DEFAULT_LAUNCH_RECOVERY_SESSIONS`](bugsee_core::runtime::DEFAULT_LAUNCH_RECOVERY_SESSIONS) + /// are taken, newest first, and no further session starts once the budget has + /// elapsed (the background worker recovers the rest, so nothing is lost). A + /// session's cost grows with its capture window (`max_window`/`max_bytes`). + /// Repeated identical crashes are queued once while the first is undelivered, + /// and at + /// [`MAX_QUEUED_REPORTS`](bugsee_core::queue::MAX_QUEUED_REPORTS) reports or + /// [`MAX_QUEUED_BYTES`](bugsee_core::queue::MAX_QUEUED_BYTES) bytes queued + /// the launch pass stops adding more (a crash it leaves is kept on disk and + /// recovered in the background). + /// `Duration::ZERO` disables the pass: everything is recovered in the + /// background. Recovered crashes are queued newest first, so they may upload + /// in that order; the server deduplicates by signature whatever the order. + pub fn launch_recovery_budget(mut self, budget: Duration) -> Self { + self.launch_recovery_budget = budget; + self + } + pub fn endpoint(mut self, endpoint: impl Into) -> Self { self.endpoint = Some(endpoint.into()); self diff --git a/crates/bugsee/tests/launch_recovery.rs b/crates/bugsee/tests/launch_recovery.rs new file mode 100644 index 0000000..805ad56 --- /dev/null +++ b/crates/bugsee/tests/launch_recovery.rs @@ -0,0 +1,56 @@ +// +// launch_recovery.rs +// bugsee +// +// Copyright © 2026 Bugsee. All rights reserved. +// +//! Issue #10 through the facade: the previous run's crash is on disk in the queue +//! by the time `launch_with` returns, even if the transport never delivers. + +use std::path::PathBuf; +use std::sync::Arc; + +use bugsee::core::transport::{Transport, TransportError}; +use bugsee::{Bugsee, LaunchOptions}; + +struct Offline; +impl Transport for Offline { + fn register_session(&self, _: &str, _: &[u8]) -> Result { + Err(TransportError::Transient("offline".into())) + } + fn create_issue(&self, _: &str, _: Option<&str>, _: &[u8]) -> Result { + Err(TransportError::Transient("offline".into())) + } + fn upload_bundle(&self, _: &str, _: &[u8]) -> Result<(), TransportError> { + Err(TransportError::Transient("offline".into())) + } +} + +#[test] +fn launch_with_queues_the_previous_crash_before_returning() { + let dir: PathBuf = + std::env::temp_dir().join(format!("bugsee-fl-{}", bugsee::core::util::random_hex(8))); + let sessions = dir.join("sessions"); + std::fs::create_dir_all(&sessions).unwrap(); + // A crashed earlier session: marker naming no live owner, and its parts. + std::fs::write(sessions.join("1.alive"), "0").unwrap(); + std::fs::create_dir_all(dir.join("parts").join("1")).unwrap(); + std::fs::write(dir.join("gen"), "1").unwrap(); + + let guard = Bugsee::launch_with( + LaunchOptions::new("T") + .data_dir(&dir) + .with_transport(Arc::new(Offline)), + ) + .unwrap(); + let queued = std::fs::read_dir(dir.join("queue")) + .map(|d| { + d.flatten() + .filter(|e| e.path().to_string_lossy().ends_with(".bundle.zip")) + .count() + }) + .unwrap_or(0); + assert_eq!(queued, 1, "no flush, no waiting"); + drop(guard); + let _ = std::fs::remove_dir_all(&dir); +}