feat: bounded launch-path recovery so crash loops can't outrun the report (#10)

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.

crates/bugsee-core/src/queue.rs

@@ -102,6 +102,63 @@ pub fn list_pending(data_dir: &Path) -> Vec<QueuedReport> {
     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<String> = std::fs::read(report.req_path())
+            .ok()
+            .and_then(|b| serde_json::from_slice::<serde_json::Value>(&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<u8>, Vec<u8>, 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<String> = 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)));

crates/bugsee-core/src/recovery.rs

@@ -61,41 +61,99 @@ 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<PendingSession> {
+    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<PendingSession> {
     let mut pending = Vec::new();
     let sessions = data_dir.join("sessions");
     let read = match std::fs::read_dir(&sessions) {
         Ok(r) => r,
         Err(_) => return pending,
     };
     let self_pid = std::process::id();
+    let mut candidates: Vec<u64> = 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;
         };
         let Ok(generation) = gen_str.parse::<u64>() 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;
         }
         let parts_dir = data_dir.join("parts").join(generation.to_string());
         let minidump = exists_opt(parts_dir.join(MINIDUMP_NAME));
         let crash_info = exists_opt(parts_dir.join(CRASH_INFO_NAME));
         let panic_info = exists_opt(parts_dir.join(PANIC_INFO_NAME));
-        pending.push(PendingSession {
+        let session = PendingSession {
             generation,
             parts_dir,
             minidump,
             crash_info,
             panic_info,
-        });
+        };
+        if retrying {
+            retries += 1;
+            retried.push(session);
+        } else {
+            fresh += 1;
+            pending.push(session);
+        }
+    }
+    // Sessions already tried come FIRST. If recovering the newest keeps killing
+    // the process, each death leaves another newer session behind and the older
+    // one would never get its next attempt counted (and never be queued, or
+    // abandoned at the cap).
+    retried.extend(pending);
+    retried
+}
+
+/// Give back the attempt [`bump_attempts`] just counted, for a session that was
+/// refused for a reason that says nothing about its evidence (the launch path had
+/// no queue room), so such refusals cannot use up `MAX_RECOVERY_ATTEMPTS`.
+pub fn undo_attempt(data_dir: &Path, pending: &PendingSession) {
+    let path = attempts_path(data_dir, pending.generation);
+    let current = std::fs::read_to_string(&path)
+        .ok()
+        .and_then(|s| s.trim().parse::<u32>().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<PathBuf> {
@@ -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<u64> {
+            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<u64> = 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();