From adbc9c8eec303658b1c8dd668c19c78f33ab8d38 Mon Sep 17 00:00:00 2001 From: Chris Czub Date: Wed, 22 Jul 2026 13:30:36 -0400 Subject: [PATCH 1/3] Surface OAuth consent on the run row and consume the persisted scan resume token The scanner now installs a custom InstalledFlow delegate that writes state='awaiting_auth' plus the consent URL onto its ingest_runs row (via a new AuthState writer-channel message, so single-writer discipline holds) and returns the row to 'running' once tokens arrive. The delegate holds only a weak sender: the authenticator outlives the scan loop, and a strong clone would keep the writer channel open forever and deadlock the end-of-scan drain. Token acquisition happens eagerly, bounded by GMAIL_STATS_AUTH_TIMEOUT (default 600s); an unattended consent becomes failed/auth_required (policy_enforced keeps its dedicated kind). Terminal UX is unchanged: the URL is still printed unless --quiet, and no schema migration was needed (the auth_url column shipped with migration 1). scan --resume (issue #26 Q4, best effort) starts listing from the run's persisted page token. A token Gmail rejects falls back to page one silently (logged in verbose mode); seen_mails dedupe keeps the counts correct either way. --resume on a bare scan is therefore no longer an error. --- Cargo.toml | 9 +- src/ingest.rs | 56 +++++ src/main.rs | 391 ++++++++++++++++++++++++++++++++-- src/{bin/web.rs => webapp.rs} | 0 4 files changed, 434 insertions(+), 22 deletions(-) rename src/{bin/web.rs => webapp.rs} (100%) diff --git a/Cargo.toml b/Cargo.toml index caea790..b29eb09 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,13 +15,14 @@ getrandom = "0.4" google-gmail1 = "7.0" lazy_static = "1.5" regex = "1.13" -# Kernel flock() for the cross-process ingest lock: safe wrappers (no unsafe in -# this crate), tiny with only the fs feature enabled. Unix-only, by design. -rustix = { version = "1.0", features = ["fs"] } +# Kernel flock() for the cross-process ingest lock plus kill() for the +# viewer's owned-child cancel path: safe wrappers (no unsafe in this crate). +# Unix-only, by design. +rustix = { version = "1.0", features = ["fs", "process"] } serde = "1.0" serde_json = "1.0" sqlx = { version = "0.9", features = ["runtime-tokio", "sqlite"] } -tokio = { version = "1.53", features = ["macros", "rt-multi-thread", "signal", "sync", "time"] } +tokio = { version = "1.53", features = ["io-util", "macros", "process", "rt-multi-thread", "signal", "sync", "time"] } yup-oauth2 = "12" [dev-dependencies] diff --git a/src/ingest.rs b/src/ingest.rs index 04feccb..ba76381 100644 --- a/src/ingest.rs +++ b/src/ingest.rs @@ -249,6 +249,15 @@ pub enum WriteMsg { /// Scan only: the mailbox's total message count, once known. total_estimate: Option, }, + /// Scan only: OAuth consent progress. The InstalledFlow delegate sets + /// state='awaiting_auth' plus the consent URL when Google asks for the + /// user; once tokens arrive the scanner sets state back to 'running' and + /// clears the URL. Rides the writer channel like every other durable + /// write, so single-writer discipline holds. + AuthState { + state: &'static str, + auth_url: Option, + }, } /// The single DB writer: receives results over the channel and commits them @@ -304,6 +313,18 @@ pub async fn db_writer( .execute(&mut conn) .await?; } + Some(WriteMsg::AuthState { state, auth_url }) => { + sqlx::query( + "UPDATE ingest_runs SET state = ?, auth_url = ?, \ + updated_at_unix = ? WHERE run_id = ?", + ) + .bind(state) + .bind(auth_url) + .bind(now_unix()) + .bind(run_id) + .execute(&mut conn) + .await?; + } }, _ = heartbeat.tick() => { sqlx::query("UPDATE ingest_runs SET updated_at_unix = ? WHERE run_id = ?") @@ -658,6 +679,41 @@ mod tests { assert_eq!(mails, 2); } + #[tokio::test] + async fn db_writer_applies_auth_state_transitions() { + let dir = tempfile::tempdir().unwrap(); + let options = file_options(&dir.path().join("stats.db")); + let mut conn = SqliteConnection::connect_with(&options).await.unwrap(); + migrate(&mut conn).await.unwrap(); + let run_id = create_run(&mut conn, "gmail_api", None).await.unwrap(); + + let (tx, rx) = mpsc::channel(4); + let writer = tokio::spawn(db_writer(conn, rx, run_id)); + tx.send(WriteMsg::AuthState { + state: "awaiting_auth", + auth_url: Some("https://accounts.google.com/o/oauth2/auth?x=1".into()), + }) + .await + .unwrap(); + tx.send(WriteMsg::AuthState { + state: "running", + auth_url: None, + }) + .await + .unwrap(); + drop(tx); + writer.await.unwrap().unwrap(); + + let mut conn = SqliteConnection::connect_with(&options).await.unwrap(); + let row = sqlx::query("SELECT state, auth_url FROM ingest_runs WHERE run_id = ?") + .bind(run_id) + .fetch_one(&mut conn) + .await + .unwrap(); + assert_eq!(row.try_get::("state").unwrap(), "running"); + assert_eq!(row.try_get::, _>("auth_url").unwrap(), None); + } + #[test] fn fingerprint_changes_when_the_file_changes() { let dir = tempfile::tempdir().unwrap(); diff --git a/src/main.rs b/src/main.rs index ec20601..687db5d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -51,10 +51,13 @@ OPTIONS: [env: GMAIL_STATS_CREDENTIALS] [default: credentials.json] --tokens OAuth token cache file (scan only) [env: GMAIL_STATS_TOKENS] [default: tokencache.json] - --resume import only: resume a cancelled/failed/abandoned - import from its recorded byte offset; falls back to - a full re-parse if the file changed (dedupe keeps - the counts correct either way) + --resume resume an earlier run of the same kind. + import: continue from the recorded byte offset, + falling back to a full re-parse if the file changed; + scan: start listing from the run's persisted page + token, silently restarting from the beginning when + the token has expired (seen-mail dedupe keeps the + counts correct either way) --quiet only errors and the final summary --verbose per-message detail (prints every sender) -h, --help show this help @@ -88,7 +91,7 @@ fn chatty() -> bool { #[derive(Debug, Clone, PartialEq, Eq)] enum Mode { - Scan, + Scan { resume: Option }, Import { path: PathBuf, resume: Option }, } @@ -159,10 +162,7 @@ fn parse_args( if positionals.len() > 1 { return Err(format!("unexpected argument {:?}", positionals[1])); } - if resume.is_some() { - return Err("--resume is only valid with the import subcommand".to_string()); - } - Mode::Scan + Mode::Scan { resume } } Some("import") => { let path = positionals @@ -284,7 +284,7 @@ async fn main() -> anyhow::Result<()> { let cancel = spawn_signal_handler(); match &cfg.mode { - Mode::Scan => run_scan(&cfg, &options, writer_conn, cancel).await, + Mode::Scan { resume } => run_scan(&cfg, &options, writer_conn, cancel, *resume).await, Mode::Import { path, resume } => { let summary = run_import(&options, writer_conn, path, *resume, cancel).await?; report_import_summary(&summary, path); @@ -614,6 +614,7 @@ async fn run_scan( options: &SqliteConnectOptions, mut writer_conn: SqliteConnection, cancel: Arc, + resume: Option, ) -> anyhow::Result<()> { let fetch_concurrency: usize = env_or("GMAIL_STATS_FETCH_CONCURRENCY", DEFAULT_FETCH_CONCURRENCY)?; @@ -635,6 +636,14 @@ async fn run_scan( .connect_with(options.clone()) .await?; + // Best-effort resume (issue #26 Q4): start listing from the page token the + // resumed run persisted. If Gmail rejects the token as expired, work() + // silently restarts from page one — seen-mail dedupe keeps that correct. + let resumed_token = match resume { + None => None, + Some(run_id) => resolve_scan_resume(&mut writer_conn, run_id).await?, + }; + let run_id = ingest::create_run(&mut writer_conn, "gmail_api", None).await?; let (mut write_tx, write_rx) = mpsc::channel::(100); let mut writer_handle = task::spawn(ingest::db_writer(writer_conn, write_rx, run_id)); @@ -673,12 +682,17 @@ async fn run_scan( // Create an authenticator that uses an InstalledFlow to authenticate. The // authentication tokens are persisted to the token cache file, so they are - // cached to disk and refreshed once they've expired. + // cached to disk and refreshed once they've expired. The custom delegate + // surfaces the consent URL on the run row (state='awaiting_auth') so the + // web viewer can offer an "Authorize in Google" link for spawned runs. let auth = match yup_oauth2::InstalledFlowAuthenticator::builder( secret, yup_oauth2::InstalledFlowReturnMethod::HTTPRedirect, ) .persist_tokens_to_disk(cfg.tokens.clone()) + .flow_delegate(Box::new(RunRowAuthDelegate { + write_tx: write_tx.downgrade(), + })) .build() .await { @@ -699,6 +713,40 @@ async fn run_scan( } }; + // Acquire a token up front instead of lazily on the first API call, so an + // unattended consent is bounded by GMAIL_STATS_AUTH_TIMEOUT (the run turns + // failed/auth_required instead of hanging forever) and the run row can + // transition awaiting_auth -> running the moment tokens arrive. When the + // token cache is already valid this never prompts. + let auth_timeout = Duration::from_secs(env_or( + "GMAIL_STATS_AUTH_TIMEOUT", + DEFAULT_AUTH_TIMEOUT_SECS, + )?); + let scopes = ["https://www.googleapis.com/auth/gmail.readonly"]; + if let Err(e) = consent_within(auth.token(&scopes), auth_timeout).await { + drop(write_tx); + let _ = writer_handle.await; + let msg = format!("{e:#}"); + ingest::finish_run( + options, + run_id, + "failed", + Some(consent_error_kind(&msg)), + Some(&msg), + ) + .await + .ok(); + return Err(e); + } + // Tokens are in hand: leave awaiting_auth (a no-op when consent was never + // needed — the row is already 'running'). + let _ = write_tx + .send(WriteMsg::AuthState { + state: "running", + auth_url: None, + }) + .await; + let hub = Gmail::new( hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()).build( hyper_rustls::HttpsConnectorBuilder::new() @@ -742,7 +790,8 @@ async fn run_scan( // progress in between exhaust the budget. let mut retries = 0; let mut progress = ScanProgress { - resume_token: None, + resume_unverified: resumed_token.is_some(), + resume_token: resumed_token, pages_done: 0, messages_listed: 0, }; @@ -934,6 +983,145 @@ where } } +/// Default bound on how long an interactive OAuth consent may stay pending +/// before the run is marked failed/auth_required. Override with the +/// GMAIL_STATS_AUTH_TIMEOUT env var (seconds). +const DEFAULT_AUTH_TIMEOUT_SECS: u64 = 600; + +/// Look up the persisted page token for `scan --resume `. Missing or +/// wrong-source runs are hard errors (the user named a specific run); a run +/// without a saved token just means a from-scratch scan. +async fn resolve_scan_resume( + conn: &mut SqliteConnection, + run_id: i64, +) -> anyhow::Result> { + let row = sqlx::query("SELECT source, resume_token FROM ingest_runs WHERE run_id = ?") + .bind(run_id) + .fetch_optional(&mut *conn) + .await?; + let Some(row) = row else { + anyhow::bail!("no ingest run {run_id} to resume"); + }; + let source: String = row.try_get("source")?; + anyhow::ensure!( + source == "gmail_api", + "run {run_id} is a {source} run, not a Gmail API scan" + ); + let token: Option = row.try_get("resume_token")?; + match token { + Some(token) if !token.is_empty() => { + if chatty() { + println!("resuming scan run {run_id} from its saved page token (best effort)"); + } + Ok(Some(token)) + } + _ => { + if chatty() { + println!("run {run_id} saved no page token; scanning from the beginning"); + } + Ok(None) + } + } +} + +/// InstalledFlow delegate that mirrors consent progress onto the run row: +/// state='awaiting_auth' plus the URL while Google waits for the user, so the +/// web viewer can render a validated "Authorize in Google" link. Terminal UX +/// is unchanged: the URL is still printed unless --quiet. +/// +/// Holds only a *weak* sender: the authenticator (and thus this delegate) +/// lives as long as the whole scan, and a strong clone here would keep the +/// writer channel open forever, deadlocking the end-of-scan drain. +struct RunRowAuthDelegate { + write_tx: mpsc::WeakSender, +} + +impl yup_oauth2::authenticator_delegate::InstalledFlowDelegate for RunRowAuthDelegate { + fn present_user_url<'a>( + &'a self, + url: &'a str, + need_code: bool, + ) -> std::pin::Pin> + Send + 'a>> + { + Box::pin(present_auth_url(self.write_tx.clone(), url, need_code)) + } +} + +async fn present_auth_url( + write_tx: mpsc::WeakSender, + url: &str, + need_code: bool, +) -> Result { + // Best-effort: consent must not die just because the writer is gone. + if let Some(write_tx) = write_tx.upgrade() { + let _ = write_tx + .send(WriteMsg::AuthState { + state: "awaiting_auth", + auth_url: Some(url.to_string()), + }) + .await; + } + if need_code { + // Parity with yup-oauth2's default delegate: interactive fallback + // reads the code from stdin. Spawned runs use HTTPRedirect + // (need_code=false) and have stdin null, so this path is + // terminal-only. + if chatty() { + println!( + "Please direct your browser to {url}, follow the instructions and enter \ + the code displayed here: " + ); + } + use tokio::io::AsyncBufReadExt; + let mut user_input = String::new(); + tokio::io::BufReader::new(tokio::io::stdin()) + .read_line(&mut user_input) + .await + .map_err(|e| format!("couldn't read code: {e}"))?; + user_input.truncate(user_input.trim_end().len()); + Ok(user_input) + } else { + if chatty() { + println!( + "Please direct your browser to {url} and follow the instructions \ + displayed there." + ); + } + Ok(String::new()) + } +} + +/// Bound an OAuth token acquisition so an unattended consent (nobody ever +/// opens the URL) becomes a clean failure instead of an eternal hang. +async fn consent_within( + fut: impl std::future::Future>, + timeout: Duration, +) -> anyhow::Result +where + E: std::error::Error + Send + Sync + 'static, +{ + match tokio::time::timeout(timeout, fut).await { + Ok(Ok(value)) => Ok(value), + Ok(Err(e)) => Err(anyhow::Error::new(e).context("acquiring an OAuth token")), + Err(_) => anyhow::bail!( + "OAuth consent was not completed within {}s \ + (set GMAIL_STATS_AUTH_TIMEOUT to adjust)", + timeout.as_secs() + ), + } +} + +/// error_kind for a failed consent: Advanced Protection rejections keep their +/// dedicated kind so the UI can steer to Takeout; everything else (timeout, +/// declined consent, broken client secret) is auth_required. +fn consent_error_kind(message: &str) -> &'static str { + if message.contains("policy_enforced") { + "policy_enforced" + } else { + "auth_required" + } +} + struct ScanCtx<'a> { pool: &'a Pool, hub: &'a GmailHub, @@ -947,9 +1135,13 @@ struct ScanCtx<'a> { /// token to resume from, advanced only once a page has been fully processed: /// a retry re-lists the page that failed instead of restarting the whole /// mailbox scan from page one. It is also persisted to the run row after -/// every page, so a future scan could pick it up best-effort. +/// every page, so `scan --resume ` can pick it up best-effort. struct ScanProgress { resume_token: Option, + /// True while resume_token came from a previous run's row and has not yet + /// produced a successful list: Gmail rejecting it (expired/invalid) means + /// a silent from-scratch restart, not a failure. + resume_unverified: bool, pages_done: u64, messages_listed: u64, } @@ -961,8 +1153,31 @@ async fn work(ctx: &ScanCtx<'_>, progress: &mut ScanProgress) -> anyhow::Result< if ctx.cancel.load(Ordering::Relaxed) { return Ok(()); } - let result = - list_messages(ctx.hub, progress.resume_token.as_deref(), ctx.rate_limiter).await?; + let result = match list_messages( + ctx.hub, + progress.resume_token.as_deref(), + ctx.rate_limiter, + ) + .await + { + Ok(result) => result, + Err(e) if progress.resume_unverified => { + // The persisted token from a previous run was rejected + // (typically expired). Fall back to page one silently — + // seen-mail dedupe makes the re-listing idempotent. + progress.resume_unverified = false; + progress.resume_token = None; + if verbose() { + println!( + "persisted resume token was rejected ({e:#}); \ + restarting the scan from the beginning" + ); + } + continue; + } + Err(e) => return Err(e), + }; + progress.resume_unverified = false; let next_page_token = result.next_page_token; let messages = result.messages.unwrap_or_default(); @@ -1497,7 +1712,7 @@ mod tests { #[test] fn bare_invocation_is_the_scan_with_default_paths() { let cfg = run_cfg(&[], no_env); - assert_eq!(cfg.mode, Mode::Scan); + assert_eq!(cfg.mode, Mode::Scan { resume: None }); assert_eq!(cfg.db, PathBuf::from("stats.db")); assert_eq!(cfg.credentials, PathBuf::from("credentials.json")); assert_eq!(cfg.tokens, PathBuf::from("tokencache.json")); @@ -1506,7 +1721,19 @@ mod tests { #[test] fn explicit_scan_subcommand_is_equivalent() { - assert_eq!(run_cfg(&["scan"], no_env).mode, Mode::Scan); + assert_eq!(run_cfg(&["scan"], no_env).mode, Mode::Scan { resume: None }); + } + + #[test] + fn scan_accepts_a_resume_run_id() { + assert_eq!( + run_cfg(&["scan", "--resume", "5"], no_env).mode, + Mode::Scan { resume: Some(5) } + ); + assert_eq!( + run_cfg(&["--resume", "5"], no_env).mode, + Mode::Scan { resume: Some(5) } + ); } #[test] @@ -1554,7 +1781,6 @@ mod tests { for args in [ &["--quiet", "--verbose"][..], &["import"][..], - &["--resume", "3"][..], &["--resume", "x", "import", "f.mbox"][..], &["frobnicate"][..], &["--frobnicate"][..], @@ -1789,6 +2015,135 @@ mod tests { assert_eq!(state, "cancelled"); } + // --- scan resume + OAuth consent surfacing --- + + #[tokio::test] + async fn resolve_scan_resume_returns_the_persisted_token() { + let dir = tempfile::tempdir().unwrap(); + let options = db_connect_options(&dir.path().join("stats.db")); + let mut conn = migrated_conn(&options).await; + let run_id = ingest::create_run(&mut conn, "gmail_api", None) + .await + .unwrap(); + sqlx::query("UPDATE ingest_runs SET resume_token = 'tok-123' WHERE run_id = ?") + .bind(run_id) + .execute(&mut conn) + .await + .unwrap(); + + let token = resolve_scan_resume(&mut conn, run_id).await.unwrap(); + assert_eq!(token.as_deref(), Some("tok-123")); + } + + #[tokio::test] + async fn resolve_scan_resume_without_a_token_scans_from_scratch() { + let dir = tempfile::tempdir().unwrap(); + let options = db_connect_options(&dir.path().join("stats.db")); + let mut conn = migrated_conn(&options).await; + let run_id = ingest::create_run(&mut conn, "gmail_api", None) + .await + .unwrap(); + assert_eq!(resolve_scan_resume(&mut conn, run_id).await.unwrap(), None); + } + + #[tokio::test] + async fn resolve_scan_resume_rejects_missing_and_wrong_source_runs() { + let dir = tempfile::tempdir().unwrap(); + let options = db_connect_options(&dir.path().join("stats.db")); + let mut conn = migrated_conn(&options).await; + + let err = resolve_scan_resume(&mut conn, 999).await.unwrap_err(); + assert!(err.to_string().contains("no ingest run")); + + let run_id = ingest::create_run(&mut conn, "mbox", None).await.unwrap(); + let err = resolve_scan_resume(&mut conn, run_id).await.unwrap_err(); + assert!(err.to_string().contains("not a Gmail API scan")); + } + + /// The consent delegate writes state='awaiting_auth' + auth_url onto the + /// run row via the writer channel, and the post-consent AuthState message + /// returns it to 'running'. No real OAuth is involved: the delegate is + /// called directly, exactly as yup-oauth2 would. + #[tokio::test] + async fn auth_delegate_surfaces_the_consent_url_on_the_run_row() { + use google_gmail1::yup_oauth2::authenticator_delegate::InstalledFlowDelegate; + + let dir = tempfile::tempdir().unwrap(); + let options = db_connect_options(&dir.path().join("stats.db")); + let mut conn = migrated_conn(&options).await; + let run_id = ingest::create_run(&mut conn, "gmail_api", None) + .await + .unwrap(); + + let (tx, rx) = mpsc::channel::(8); + let writer = task::spawn(ingest::db_writer(conn, rx, run_id)); + + let delegate = RunRowAuthDelegate { + write_tx: tx.downgrade(), + }; + let url = "https://accounts.google.com/o/oauth2/auth?client_id=x&redirect_uri=y"; + let code = delegate.present_user_url(url, false).await.unwrap(); + assert_eq!(code, "", "HTTPRedirect mode returns no code"); + + // Let the writer drain the awaiting_auth message, then check the row. + tx.send(WriteMsg::Progress { + messages_seen: 0, + bytes_done: None, + resume_token: None, + total_estimate: None, + }) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(100)).await; + let mut check = SqliteConnection::connect_with(&options).await.unwrap(); + let row = sqlx::query("SELECT state, auth_url FROM ingest_runs WHERE run_id = ?") + .bind(run_id) + .fetch_one(&mut check) + .await + .unwrap(); + assert_eq!(row.try_get::("state").unwrap(), "awaiting_auth"); + assert_eq!(row.try_get::("auth_url").unwrap(), url); + + // Tokens arrived: the scanner sends the running transition. + tx.send(WriteMsg::AuthState { + state: "running", + auth_url: None, + }) + .await + .unwrap(); + // The delegate holds a Sender clone; both must drop for the writer + // to drain and exit. + drop(delegate); + drop(tx); + writer.await.unwrap().unwrap(); + let row = sqlx::query("SELECT state, auth_url FROM ingest_runs WHERE run_id = ?") + .bind(run_id) + .fetch_one(&mut check) + .await + .unwrap(); + assert_eq!(row.try_get::("state").unwrap(), "running"); + assert_eq!(row.try_get::, _>("auth_url").unwrap(), None); + } + + /// An unattended consent (token future never resolves) must convert into + /// an error after the timeout, and its kind must map to auth_required. + #[tokio::test] + async fn consent_timeout_becomes_auth_required() { + let err = consent_within( + std::future::pending::>(), + Duration::from_millis(50), + ) + .await + .unwrap_err(); + let msg = format!("{err:#}"); + assert!(msg.contains("consent was not completed"), "got: {msg}"); + assert_eq!(consent_error_kind(&msg), "auth_required"); + assert_eq!( + consent_error_kind("Error 400: policy_enforced"), + "policy_enforced" + ); + } + #[tokio::test] async fn import_refuses_a_non_mbox_file() { let dir = tempfile::tempdir().unwrap(); diff --git a/src/bin/web.rs b/src/webapp.rs similarity index 100% rename from src/bin/web.rs rename to src/webapp.rs From 01478d8db9e00af936efdfe512eb9742863d84e0 Mon Sep 17 00:00:00 2001 From: Chris Czub Date: Wed, 22 Jul 2026 13:31:03 -0400 Subject: [PATCH 2/3] Web viewer: launch, cancel, and log endpoints behind the full CSRF stack The viewer application moves from src/bin/web.rs into a library module (gmail_stats::webapp) so integration tests can drive the exact router the binary serves; src/bin/web.rs is now a thin main. Phase B behavior is unchanged. New surface (issue #30, per the #26 design): - POST /api/runs spawns the ingester as a supervised child: fixed argv (never a shell), stdin null, stdout null, stderr piped into a 200-line in-memory ring buffer, kill_on_drop(false) so runs survive viewer restarts. The binary is located as a sibling of current_exe() (GMAIL_STATS_INGEST_BIN override); spawned runs get --quiet and the viewer's --db path. Responses: 202 {run_id} (the child mints the id by inserting its run row; the viewer waits for it, matched by child pid), 409 run_active (friendly flock pre-check, plus mapping the child's own authoritative flock refusal), 422 bad_path/bad_resume/bad_source (absolute + regular + readable + mbox magic sniff), 500 spawn_failed with a cargo build hint. - POST /api/runs/{id}/cancel: owned runs only. SIGTERM through the supervisor that owns the child handle (never a pid read from the database), SIGKILL after a 10s grace; a SIGKILLed child's open row is abandoned by the next ingester's janitor - the viewer still never writes. 409 not_owned otherwise. - GET /api/runs/{id}/log: the owned child's stderr ring buffer, memory-only, gone on viewer restart. - CSRF stack on every mutating route, all three layers required: strict Content-Type application/json (415), X-Gmail-Stats-Csrf equal to the per-process token from /api/status (403, constant-time compare), and Origin / Sec-Fetch-Site validation when present (absent passes - curl is the same local user). No CORS headers anywhere. Host guard and the existing security headers stay on every route. - /api/status now reports owns_active_run truthfully and exposes auth_url on run rows (the client refuses to hyperlink anything that is not https://accounts.google.com/...). src/bin/fake_ingester.rs is a test-only stand-in driven by tests: it takes the real flock and writes real run rows, with modes for completing, crashing mid-run, honoring SIGTERM, ignoring SIGTERM, emitting stderr, and simulating a lost flock race. tests/web_mutating.rs covers the full 415/403/409/422/202 matrix via tower::ServiceExt::oneshot (asserting no Access-Control-* headers) and the supervision paths: normal completion, crash then janitor abandonment, cancel honored, SIGTERM-ignored SIGKILL escalation, viewer restart losing ownership while the run continues, log ring capping, and both flock-conflict shapes. --- src/bin/fake_ingester.rs | 162 ++++++++ src/bin/web.rs | 12 + src/lib.rs | 1 + src/webapp.rs | 787 ++++++++++++++++++++++++++++++++++++-- tests/web_mutating.rs | 790 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 1718 insertions(+), 34 deletions(-) create mode 100644 src/bin/fake_ingester.rs create mode 100644 src/bin/web.rs create mode 100644 tests/web_mutating.rs diff --git a/src/bin/fake_ingester.rs b/src/bin/fake_ingester.rs new file mode 100644 index 0000000..c1b909a --- /dev/null +++ b/src/bin/fake_ingester.rs @@ -0,0 +1,162 @@ +//! Test-only stand-in for the real ingester, used by the web viewer's +//! supervision tests (tests/web_mutating.rs) to drive child-process paths that +//! are impractical with the real binary: crashing mid-run, ignoring SIGTERM, +//! emitting controlled stderr. It is never spawned in production — the viewer +//! only launches it when GMAIL_STATS_INGEST_BIN points at it explicitly. +//! +//! It speaks just enough of the real CLI (`scan`/`import `, `--db`, +//! `--quiet`, `--resume`) to be spawned by the viewer, takes the real ingest +//! flock, and writes real `ingest_runs` rows, so the viewer's observations are +//! exercised against the genuine coordination primitives. +//! +//! Behavior is selected with FAKE_INGESTER_MODE: +//! - `complete` (default): run row -> brief work -> done, exit 0 +//! - `crash`: run row, then exit(1) leaving the row open (flock freed by the +//! kernel; the next real ingester's janitor marks it abandoned) +//! - `hang_trap`: run row, then wait; on SIGTERM mark the row cancelled and +//! exit 0 (the real ingester's clean-shutdown contract) +//! - `hang_ignore`: run row, then wait forever, ignoring SIGTERM (forces the +//! viewer's SIGKILL escalation) +//! - `stderr`: emit FAKE_INGESTER_STDERR_LINES lines (default 250) to stderr, +//! then behave like `hang_trap` +//! - `norow`: print a failure to stderr and exit(1) before any row exists + +use std::path::PathBuf; +use std::time::Duration; + +use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqliteSynchronous}; +use sqlx::{Connection, SqliteConnection}; + +use gmail_stats::ingest; + +fn parse_args() -> (String, PathBuf) { + let mut source = "gmail_api".to_string(); + let mut db = PathBuf::from("stats.db"); + let mut args = std::env::args().skip(1); + while let Some(arg) = args.next() { + match arg.as_str() { + "scan" => source = "gmail_api".to_string(), + "import" => { + source = "mbox".to_string(); + let _path = args.next(); + } + "--db" => { + if let Some(path) = args.next() { + db = PathBuf::from(path); + } + } + "--resume" => { + let _ = args.next(); + } + "--quiet" | "--verbose" => {} + other => { + eprintln!("fake ingester: ignoring argument {other:?}"); + } + } + } + (source, db) +} + +/// Tests run in parallel inside one process, so a global env var cannot pick +/// the mode per spawn; a `fake_mode` file next to the target database wins, +/// falling back to FAKE_INGESTER_MODE, then `complete`. +fn resolve_mode(db: &std::path::Path) -> String { + if let Some(dir) = db.parent() { + if let Ok(mode) = std::fs::read_to_string(dir.join("fake_mode")) { + let mode = mode.trim().to_string(); + if !mode.is_empty() { + return mode; + } + } + } + std::env::var("FAKE_INGESTER_MODE").unwrap_or_else(|_| "complete".to_string()) +} + +/// Hard lifetime cap so a test that forgets to cancel can never leak an +/// eternally sleeping process. +const MAX_LIFE: Duration = Duration::from_secs(30); + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let (source, db) = parse_args(); + let mode = resolve_mode(&db); + + if mode == "norow" { + eprintln!("fake ingester: refusing to start (norow mode)"); + std::process::exit(1); + } + if mode == "flockfail" { + // Simulate losing the flock race *after* the viewer's friendly + // pre-check passed (the TOCTOU window): same message and exit as the + // real acquire_ingest_lock failure. + eprintln!( + "Error: another ingester is already running against {} \ + (could not acquire exclusive lock); wait for it to finish or stop it first", + db.display() + ); + std::process::exit(1); + } + + // The real flock, so the viewer's pre-check/probe sees the truth. + let _lock = ingest::acquire_ingest_lock(&db)?; + + let options = SqliteConnectOptions::new() + .filename(&db) + .create_if_missing(true) + .journal_mode(SqliteJournalMode::Wal) + .synchronous(SqliteSynchronous::Normal) + .busy_timeout(Duration::from_secs(5)); + let mut conn = SqliteConnection::connect_with(&options).await?; + ingest::migrate(&mut conn).await?; + ingest::abandon_stale_runs(&mut conn).await?; + let run_id = ingest::create_run(&mut conn, &source, None).await?; + conn.close().await.ok(); + + if mode == "stderr" { + let lines: usize = std::env::var("FAKE_INGESTER_STDERR_LINES") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(250); + for i in 1..=lines { + eprintln!("fake stderr line {i}"); + } + } + + match mode.as_str() { + "complete" => { + tokio::time::sleep(Duration::from_millis(200)).await; + ingest::finish_run(&options, run_id, "done", None, None).await?; + } + "crash" => { + eprintln!("fake ingester: crashing mid-run"); + std::process::exit(1); + } + "hang_ignore" => { + // Swallow SIGTERM (up to the lifetime cap); only SIGKILL ends + // this mode early. + let mut sigterm = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?; + let deadline = tokio::time::sleep(MAX_LIFE); + tokio::pin!(deadline); + loop { + tokio::select! { + _ = sigterm.recv() => eprintln!("fake ingester: ignoring SIGTERM"), + _ = &mut deadline => break, + } + } + } + // hang_trap and stderr: wait for SIGTERM, then the real ingester's + // clean-shutdown contract (cancelled row, exit 0). + _ => { + let mut sigterm = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?; + tokio::select! { + _ = sigterm.recv() => { + ingest::finish_run(&options, run_id, "cancelled", None, None).await?; + } + _ = tokio::time::sleep(MAX_LIFE) => {} + } + } + } + Ok(()) +} diff --git a/src/bin/web.rs b/src/bin/web.rs new file mode 100644 index 0000000..f154331 --- /dev/null +++ b/src/bin/web.rs @@ -0,0 +1,12 @@ +//! Thin entry point for the local web viewer. The whole application lives in +//! `gmail_stats::webapp` so integration tests can drive the exact router the +//! binary serves. +//! +//! Usage: `cargo run --bin web [port]` (default port 7878), then open the +//! printed URL. Run from the repo root so `./stats.db` resolves, or point +//! GMAIL_STATS_DB at the database. + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + gmail_stats::webapp::serve().await +} diff --git a/src/lib.rs b/src/lib.rs index 2975970..ebb1048 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,3 +8,4 @@ pub mod ingest; pub mod mbox; +pub mod webapp; diff --git a/src/webapp.rs b/src/webapp.rs index 204a289..a4cbe9f 100644 --- a/src/webapp.rs +++ b/src/webapp.rs @@ -1,41 +1,58 @@ -//! Local web viewer for gmail-stats (issue #11 Phase 1, issue #28 Phase B). +//! Local web viewer for gmail-stats (issue #11 Phase 1, issue #28 Phase B, +//! issue #30 Phase C). //! -//! Serves the embedded UI plus read-only JSON endpoints over `./stats.db`, -//! bound to 127.0.0.1 only. Phase B adds observe-only ingestion state: -//! `GET /api/status` (db readiness, active/last run, flock probe, viewer-side -//! rate/ETA, CSRF token slot) and `GET /api/runs` (run history). The viewer -//! performs no database writes of any kind; the only file it touches is the -//! ingest lockfile, probed without truncation via open + try-lock + release. +//! Serves the embedded UI plus JSON endpoints over `./stats.db`, bound to +//! 127.0.0.1 only. Phase B added observe-only ingestion state: `GET +//! /api/status` (db readiness, active/last run, flock probe, viewer-side +//! rate/ETA, CSRF token) and `GET /api/runs` (run history). Phase C adds the +//! state-changing surface: `POST /api/runs` (spawn the ingester as a +//! supervised child), `POST /api/runs/{id}/cancel` (SIGTERM the owned child, +//! SIGKILL escalation), and `GET /api/runs/{id}/log` (in-memory stderr ring +//! buffer of owned children). //! -//! Usage: `cargo run --bin web [port]` (default port 7878), then open the -//! printed URL. Run from the repo root so `./stats.db` resolves, or point -//! GMAIL_STATS_DB at the database. +//! The viewer still performs no database writes of any kind: every durable +//! state change happens inside the spawned `gmail_stats` ingester process. +//! The only files the viewer touches are the ingest lockfile (probed without +//! truncation via open + try-lock + release) — its new powers are exactly +//! spawn, signal-own-child, and hold a stderr ring buffer in memory. +//! +//! Every mutating route sits behind the full CSRF stack from issue #26: +//! strict `Content-Type: application/json` (415 otherwise), the per-process +//! `X-Gmail-Stats-Csrf` token served by `/api/status`, and Origin / +//! Sec-Fetch-Site validation when those headers are present. No CORS headers +//! are ever emitted. +//! +//! This module is the whole application; `src/bin/web.rs` is a thin `main`. -use std::collections::VecDeque; +use std::collections::{HashMap, VecDeque}; use std::path::{Path, PathBuf}; +use std::process::Stdio; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use axum::{ - extract::{Query, Request, State}, - http::{header, HeaderValue, StatusCode}, + extract::{Path as UrlPath, Query, Request, State}, + http::{header, HeaderMap, HeaderValue, StatusCode}, middleware::{self, Next}, response::{Html, IntoResponse, Response}, - routing::get, + routing::{get, post}, Json, Router, }; use serde_json::json; use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions, SqliteRow}; use sqlx::{Row, SqlitePool}; +use tokio::io::AsyncBufReadExt; +use tokio::sync::mpsc; -use gmail_stats::ingest; +use crate::ingest; // CSS/JS are separate same-origin assets, not inline blocks: under the // `default-src 'self'` CSP below, `'self'` only allowlists same-origin URLs — // inline