feat: defined shutdown/flush contract and opt-in SIGTERM/SIGINT flush (#8)

DESIGN.md

@@ -85,7 +85,7 @@ A Cargo workspace. Integrations are **separate crates** re-exported by an umbrel
 
 ## 6. Public API surface
 
-Global singleton (mobile parity); `launch` returns a guard whose `Drop` flushes.
+Global singleton (mobile parity); `launch` returns a guard whose `Drop` persists and stops cleanly (see §8; it does not upload).
 
 ```rust
 // Lifecycle
@@ -218,6 +218,21 @@ 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.
 
+**Shutdown and flush contract (issue #8).** `flush(timeout)` persists everything captured before the call and tries to upload every queued report; it returns `true` only if the outbound queue is empty, leaves the SDK running, and is idempotent. `stop()` (and the launch guard's `Drop`) persists, then ends the session by removing the liveness marker; it never uploads from the durable queue, so an undelivered report stays there and goes out next launch (disk mode). `flush` drains every queue it can see: the data directory's `queue/`, the legacy directory adopted by #9, and the in-memory backlog. In memory mode `flush` is the only guarantee; `stop` then gives the backlog one last best-effort attempt (the uploader's stop arm, bounded to 1.5 s, which deliberately ignores the `stopping` flag that aborts a flush), and a report still unsent at exit is lost. `shutdown(timeout)` is `flush` then `stop`. A `stop` arriving during a flush aborts the flush between deliveries instead of waiting out its deadline. The uploader's acknowledgement does not gate ending the session (it writes no session parts and may sit in an HTTP call for tens of seconds), so a stop during a slow upload still ends cleanly and the abandoned report is re-sent next launch: delivery is **at-least-once**, and a graceful stop plus restart reports nothing twice only in the absence of a kill mid-upload. Worst case a `shutdown(t)` takes `t` plus about 4 s. The release-health record (#12) is finished on any non-panicking stop, whatever the threads acknowledged; only the capture worker's acknowledgement gates removing the marker (and releasing the session lock early, #14). The marker is intentionally left when the capture worker does not acknowledge in time, so a wedged shutdown is recovered, not lost. None of it is async-signal-safe; in a `fork` child it acts on the child's own revived recorder.
+
+| Exit path | Marker | Reports | Next launch |
+|---|---|---|---|
+| return from `main` / `stop` / `shutdown` | removed | persisted, queue intact | no abnormal-exit; queue delivered once |
+| `process::exit` / `exit()` | left (default) | as already written | one abnormal-exit report; queue delivered once |
+| same, with `LaunchOptions::flush_at_exit` (unix only; ignored on Windows, call `shutdown`) | removed by an `atexit` handler (not run in `fork` children) | persisted | as the first row |
+| `SIGTERM`/`SIGINT` + `Bugsee::install_termination_flush` | removed if the signal would have terminated the process | flushed, bounded | as the first row; host handler still called |
+| Windows Ctrl+C/Break/close/logoff/shutdown + hook | removed (`shutdown`) | flushed, bounded | as the first row |
+| `SIGTERM`/`SIGINT` without the hook, `SIGKILL`, OOM, power loss | left | records appended as drained survive | one abnormal-exit report |
+
+*Decision: `exit` without `stop` stays an abnormal exit.* From inside the process an `exit` is indistinguishable from a runtime tearing down after a fault the host swallowed, so the default errs towards reporting; hosts that exit on purpose opt in with `flush_at_exit` (or call `shutdown`). *Decision: the signal hook is opt-in and chaining.* The handler only writes the signal number to a non-blocking self-pipe and then calls the previously installed handler; a watcher thread does the flush. With a default previous disposition it runs `shutdown`, restores `SIG_DFL` and re-raises, so the exit status is unchanged; a second signal during the shutdown terminates at once. With a host handler it flushes only and the host decides; a host handler installed AFTER ours (tokio, `signal-hook`) is detected by asking the kernel what is in force, and is treated the same, so we never kill a host mid-graceful-shutdown. A signal ignored at install is left alone. A second signal while the first is served, or a dead watcher, terminates at once. The chained call keeps the host's `sa_mask`, `SA_ONSTACK`, `SA_NODEFER`, `SA_SIGINFO`, and `SA_RESETHAND` (emulated). Do not call `exit` from a host handler with `flush_at_exit` on: the exit-time `shutdown` would run in signal context. A `fork` child inherits the handler but not the watcher: it chains/terminates as if the hook were absent until `install_termination_flush` is called again there. Windows: a console control handler that runs `shutdown` for every event (a host that swallows Ctrl+C and keeps running must register its handler AFTER the SDK's, so it runs first) that returns `FALSE` to pass the event on; service-control stop is not a console event and needs a `shutdown` call from the host's stop handler. **Not compiled or run on Windows** (a macOS cross-check stops at the C dependencies of `zstd-sys`/`ring`); the Windows half was reviewed by reading only.
+
+Tests: `bugsee/tests/shutdown.rs` (a real child process per exit path, then a relaunch asserting exactly-once delivery and the abnormal-exit count).
+
 **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`.
@@ -320,7 +335,7 @@ Documented limits. Records of legacy and extra recovery directories are not surf
 | Report/upload worker | 1 | `reports/`, `queue/` | export → bundle (zstd-93) → 3-step upload → retry |
 | Crash-time context | any thread | — | panic observer / native handler: touch only preallocated `PanicSnapshot` + minidump |
 
-**Crash-time isolation rule:** the fault path never enqueues/allocates/locks — it writes only to fixed preallocated storage + a file-based crash marker (worker may be dead/frozen). **Back-pressure:** the depth-capped channel drops the newest entry when full (never blocks a producer); **reports** are bounded by a separate in-flight counter (heavier per message, so a report storm drops-newest **non-crash** reports over the cap while never dropping a crash); the worker holds no large in-RAM buffer (it frames one record at a time and `write_all`s it), so on-disk growth is bounded by `max_bytes`/`max_window` eviction rather than RSS. **Async-optional:** capture worker is always a `std` thread; only the uploader can become a task on the host runtime (`async` feature). **Shutdown:** guard `Drop` → `flush(timeout)`.
+**Crash-time isolation rule:** the fault path never enqueues/allocates/locks — it writes only to fixed preallocated storage + a file-based crash marker (worker may be dead/frozen). **Back-pressure:** the depth-capped channel drops the newest entry when full (never blocks a producer); **reports** are bounded by a separate in-flight counter (heavier per message, so a report storm drops-newest **non-crash** reports over the cap while never dropping a crash); the worker holds no large in-RAM buffer (it frames one record at a time and `write_all`s it), so on-disk growth is bounded by `max_bytes`/`max_window` eviction rather than RSS. **Async-optional:** capture worker is always a `std` thread; only the uploader can become a task on the host runtime (`async` feature). **Shutdown:** guard `Drop` → persist + clean stop (see §8, "Shutdown and flush contract"); `Bugsee::shutdown(timeout)` = flush then stop.
 
 ## 12. Crash-time safety, best-effort flush & edge cases
 

crates/bugsee-core/src/runtime.rs

@@ -270,6 +270,10 @@ struct Shared {
     max_report_queued: usize,
     /// Whether telemetry sampling is paused (mirrors the facade's pause state).
     paused: AtomicBool,
+    /// Set when the recorder is being stopped: a flush in progress gives up at
+    /// once, so the stop is acknowledged inside its bound (and the session can
+    /// end cleanly) instead of waiting out the flush's whole deadline.
+    stopping: AtomicBool,
     /// Reports that could not be queued on disk, delivered from memory. An entry
     /// stays here until its delivery is settled, so the length is the real backlog.
     mem_queue: ForkLock<Vec<MemReport>>,
@@ -438,6 +442,7 @@ impl Recorder {
             report_queued: AtomicUsize::new(0),
             max_report_queued: DEFAULT_MAX_REPORT_QUEUED,
             paused: AtomicBool::new(false),
+            stopping: AtomicBool::new(false),
             mem_queue: ForkLock::new(Vec::new()),
             next_mem_id: AtomicU64::new(0),
             mem_blacklist: ForkLock::new(BTreeSet::new()),
@@ -830,6 +835,7 @@ impl Drop for Recorder {
         // A thread wedged in a host callback with no timeout is DETACHED (its
         // JoinHandle dropped un-joined) rather than hanging the stopping thread
         // forever. (Called only after the facade released its global lock.)
+        self.shared.stopping.store(true, Ordering::SeqCst);
         let (tx, rx) = channel();
         let worker_acked =
             self.tx.send(Msg::Stop(tx)).is_ok() && rx.recv_timeout(Duration::from_secs(2)).is_ok();
@@ -849,7 +855,7 @@ impl Drop for Recorder {
                 let _ = handle.join();
             }
         }
-        // Only mark a genuinely clean shutdown. If either thread merely DETACHED
+        // Only mark a genuinely clean shutdown. If the capture worker merely DETACHED
         // (didn't ack in time — e.g. a slow flush), it may still be writing this
         // session's data, so removing the marker now would make next-launch
         // recovery miss the generation entirely if the process then dies mid-tail.
@@ -867,6 +873,12 @@ impl Drop for Recorder {
         // A panic that was merely *contained* never reaches here while
         // unwinding, and its snapshot is deleted by `report_caught`, so this
         // cannot turn a handled panic into a phantom crash.
+        // The uploader is deliberately NOT part of this condition: it never
+        // writes session parts, and it can be inside an HTTP call that no signal
+        // can interrupt (connect/read timeouts are tens of seconds). Waiting on
+        // it would turn every stop during a slow upload into a false
+        // abnormal-exit report. The cost is at-least-once delivery: a report
+        // whose upload was cut short is re-sent next launch.
         //
         // The health record is the exception: it is finished whenever the stop
         // is not a death by panic, whatever the threads acked. A detached
@@ -887,14 +899,15 @@ impl Drop for Recorder {
                 health.finish_clean(self.shared.health_errors.load(Ordering::SeqCst));
             }
         }
-        if worker_acked && uploader_acked && !std::thread::panicking() {
+        if worker_acked && !std::thread::panicking() {
             if let Some(session) = &self.session {
                 // The finished health record first, then the marker: a death in
                 // between leaves a marker that recovery settles idempotently.
                 session.end();
             }
-        } else if !(worker_acked && uploader_acked) {
-            // A detached thread may still be writing this session's data: keep
+        } else if !worker_acked {
+            // A detached capture worker may still be writing this session's data
+            // (the uploader writes none, so its detaching does not matter): keep
             // the liveness lock until the process exits so no peer recovers (and
             // deletes) the session underneath it.
             if let Some(session) = &mut self.session {
@@ -2042,6 +2055,8 @@ fn uploader_loop(
                         true,
                         false,
                         Some(Instant::now() + Duration::from_millis(1500)),
+                        // `stopping` is set by now; this IS the stop-time drain.
+                        false,
                     )
                 }));
                 let _ = ack.send(());
@@ -2112,7 +2127,9 @@ fn drain_all(
 /// is set so backing-off reports are retried immediately. Returns whether any
 /// report was attempted (used to pace `drain_until_empty`).
 fn drain(shared: &Shared, data_dir: Option<&Path>, backoff_base: Duration, force: bool) -> bool {
-    let attempted = drain_memory(shared, backoff_base, force, true, None);
+    // Not `honor_stop`: the scheduled pass behaves as before #8, and the
+    // stop-time drain relies on it having had its one attempt.
+    let attempted = drain_memory(shared, backoff_base, force, true, None, false);
     match data_dir {
         Some(dir) => drain_disk(shared, dir, backoff_base, force) | attempted,
         None => attempted,
@@ -2261,19 +2278,23 @@ fn notify_dropped(shared: &Shared, reason: DropReason) {
 /// An entry STAYS in the queue while it is being delivered (the lock is held only
 /// to pick it and to settle it, never across the network), so the queue length is
 /// the true backlog and [`MAX_MEMORY_REPORTS`] holds. Stops at `deadline`, like
-/// the disk loop in [`drain_until_empty`].
+/// the disk loop in [`drain_until_empty`]. With `honor_stop`, also stops as soon
+/// as the recorder is being stopped (only an explicit flush passes `true`).
 fn drain_memory(
     shared: &Shared,
     backoff_base: Duration,
     force: bool,
     count_failure: bool,
     deadline: Option<Instant>,
+    honor_stop: bool,
 ) -> bool {
     let mut attempted = false;
     // Each entry is tried once per call, whatever its outcome.
     let mut tried: Vec<u64> = Vec::new();
     loop {
-        if deadline.is_some_and(|d| Instant::now() >= d) {
+        if deadline.is_some_and(|d| Instant::now() >= d)
+            || (honor_stop && shared.stopping.load(Ordering::SeqCst))
+        {
             break;
         }
         // A dying process starts no delivery it cannot finish (crash_flush).
@@ -2376,13 +2397,32 @@ fn drain_until_empty(
 ) -> bool {
     let start = Instant::now();
     loop {
+        if shared.stopping.load(Ordering::SeqCst) {
+            return false; // stop wins; what is queued stays queued
+        }
         let now = epoch_ms();
-        drain_memory(shared, backoff_base, true, false, Some(start + deadline));
-        for &data_dir in dirs {
+        drain_memory(
+            shared,
+            backoff_base,
+            true,
+            false,
+            Some(start + deadline),
+            true,
+        );
+        // A stop that arrived during the memory pass must not be followed by a
+        // blocking disk upload: the uploader has to reach `UploadMsg::Stop` and
+        // run the stop-time memory drain, which a hung HTTP call would starve.
+        if shared.stopping.load(Ordering::SeqCst) {
+            return false;
+        }
+        'dirs: for &data_dir in dirs {
             for report in pending_reports(shared, data_dir) {
                 let Some(_claim) = queue::try_claim_lenient(&report) else {
                     continue; // a peer is delivering it; the loop below waits it out
                 };
+                if shared.stopping.load(Ordering::SeqCst) {
+                    return false; // checked after the claim so no upload starts
+                }
                 let Ok((zip, request_json, retry, _next)) = queue::load(&report) else {
                     continue;
                 };
@@ -2397,8 +2437,8 @@ fn drain_until_empty(
                     now,
                     false,
                 );
-                if start.elapsed() >= deadline {
-                    break;
+                if start.elapsed() >= deadline || shared.stopping.load(Ordering::SeqCst) {
+                    break 'dirs;
                 }
             }
         }