From 5026ce22de39c257d07ebf28e7aea2867526d8c0 Mon Sep 17 00:00:00 2001 From: Dennis Nemec Date: Wed, 2 Sep 2026 22:58:16 +0200 Subject: [PATCH] Implement Trivy adapter, contain job panics, fail interrupted runs on startup The Trivy scanner still had unimplemented stubs, which panicked the scan task in the deployed test instance and left the run in 'running' forever. JobRunner now runs handlers in their own task and marks a panic as a failed run; on startup runs left 'running' by a previous process are marked failed. Co-Authored-By: Claude Fable 5.1 --- backend/crates/api/src/lib.rs | 4 + backend/crates/application/src/jobs.rs | 29 +++- backend/crates/application/src/test_fakes.rs | 10 ++ .../application/src/tests/jobs_tests.rs | 46 +++++++ backend/crates/domain/src/ports.rs | 2 + backend/crates/infrastructure/src/sqlite.rs | 9 ++ backend/crates/infrastructure/src/trivy.rs | 127 ++++++++++++++++-- 7 files changed, 212 insertions(+), 15 deletions(-) diff --git a/backend/crates/api/src/lib.rs b/backend/crates/api/src/lib.rs index 700cf01..ac8d0a5 100644 --- a/backend/crates/api/src/lib.rs +++ b/backend/crates/api/src/lib.rs @@ -161,6 +161,10 @@ impl AppState { } pub async fn bootstrap(&self) -> anyhow::Result<()> { + let stale = self.jobs.recover().await?; + if stale > 0 { + tracing::warn!("marked {stale} interrupted job run(s) as failed"); + } if let Some((email, password)) = &self.cfg.bootstrap_admin { if self.users.bootstrap_admin(email, password).await? { tracing::info!("created bootstrap admin {email}"); diff --git a/backend/crates/application/src/jobs.rs b/backend/crates/application/src/jobs.rs index e5a0bc1..3da736b 100644 --- a/backend/crates/application/src/jobs.rs +++ b/backend/crates/application/src/jobs.rs @@ -94,22 +94,41 @@ impl JobRunner { } async fn execute(runs: Arc, handler: Arc, run: &JobRun) { - let log = RepoLog { + let log = Arc::new(RepoLog { runs: runs.clone(), id: run.id, - }; - let status = match handler.run(run.params.clone(), &log).await { - Ok(()) => JobStatus::Success, - Err(e) => { + }); + // Run the handler in its own task so a panic is contained and reported. + let (params, task_log) = (run.params.clone(), log.clone()); + let joined = tokio::spawn(async move { handler.run(params, &*task_log).await }).await; + let status = match joined { + Ok(Ok(())) => JobStatus::Success, + Ok(Err(e)) => { log.line(&format!("ERROR: {e}")).await; JobStatus::Failed } + Err(e) => { + log.line(&format!("ERROR: job panicked: {e}")).await; + JobStatus::Failed + } }; if let Err(e) = runs.finish(run.id, status).await { tracing_line(&format!("failed to finish job: {e}")); } } + /// Mark runs left in `running` state by a previous process as failed. Returns the count. + pub async fn recover(&self) -> Result { + let stale = self.runs.running().await?; + for r in &stale { + self.runs + .append_log(r.id, "ERROR: interrupted by a restart of the service") + .await?; + self.runs.finish(r.id, JobStatus::Failed).await?; + } + Ok(stale.len()) + } + /// Start a job in the background. Fails with `Conflict` if the kind is already running. pub async fn start( &self, diff --git a/backend/crates/application/src/test_fakes.rs b/backend/crates/application/src/test_fakes.rs index f7193fa..d8e2c4e 100644 --- a/backend/crates/application/src/test_fakes.rs +++ b/backend/crates/application/src/test_fakes.rs @@ -292,6 +292,16 @@ impl JobRunRepository for MemJobRuns { .find(|r| r.kind == kind && r.status != JobStatus::Running) .cloned()) } + async fn running(&self) -> Result, DomainError> { + Ok(self + .0 + .lock() + .unwrap() + .iter() + .filter(|r| r.status == JobStatus::Running) + .cloned() + .collect()) + } } pub fn smtp() -> SmtpSettings { diff --git a/backend/crates/application/src/tests/jobs_tests.rs b/backend/crates/application/src/tests/jobs_tests.rs index 8015b62..e0b1e24 100644 --- a/backend/crates/application/src/tests/jobs_tests.rs +++ b/backend/crates/application/src/tests/jobs_tests.rs @@ -7,6 +7,7 @@ use uuid::Uuid; use crate::jobs::{JobHandler, JobLog, JobRunner}; use crate::test_fakes::MemJobRuns; +use domain::ports::JobRunRepository; struct Echo; #[async_trait] @@ -102,3 +103,48 @@ async fn list_returns_newest_first_with_limit() { assert_eq!(list.len(), 2); assert!(list[0].started_at >= list[1].started_at); } + +struct Panics; +#[async_trait] +impl JobHandler for Panics { + async fn run(&self, _: Option, log: &dyn JobLog) -> Result<(), String> { + log.line("about to panic").await; + panic!("handler bug"); + } +} + +#[tokio::test] +async fn panicking_handler_marks_run_failed() { + let runs = Arc::new(MemJobRuns::default()); + let r = JobRunner::new(runs.clone()).register(JobKind::Backup, Arc::new(Panics)); + let run = r.start(JobKind::Backup, None, "test").await.unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let run = r.get(run.id).await.unwrap(); + assert_eq!(run.status, JobStatus::Failed); + assert!(run.log.contains("panicked"), "{}", run.log); + // the kind is free again + assert!(r.start(JobKind::Backup, None, "test").await.is_ok()); +} + +#[tokio::test] +async fn stale_running_runs_are_failed_on_recovery() { + let runs = Arc::new(MemJobRuns::default()); + runs.insert(&domain::jobs::JobRun { + id: Uuid::new_v4(), + kind: JobKind::PackageRefresh, + params: None, + status: JobStatus::Running, + started_at: chrono::Utc::now(), + finished_at: None, + log: String::new(), + triggered_by: "old process".into(), + }) + .await + .unwrap(); + let r = JobRunner::new(runs.clone()).register(JobKind::PackageRefresh, Arc::new(Echo)); + assert_eq!(r.recover().await.unwrap(), 1); + let list = r.list(10).await.unwrap(); + assert_eq!(list[0].status, JobStatus::Failed); + assert!(list[0].log.contains("interrupted")); + assert!(r.start(JobKind::PackageRefresh, None, "test").await.is_ok()); +} diff --git a/backend/crates/domain/src/ports.rs b/backend/crates/domain/src/ports.rs index 9ed16b1..842984a 100644 --- a/backend/crates/domain/src/ports.rs +++ b/backend/crates/domain/src/ports.rs @@ -78,6 +78,8 @@ pub trait JobRunRepository: Send + Sync { async fn list(&self, limit: u32) -> Result, DomainError>; async fn find_running(&self, kind: JobKind) -> Result, DomainError>; async fn last_finished(&self, kind: JobKind) -> Result, DomainError>; + /// All runs still marked running (e.g. after a crash). + async fn running(&self) -> Result, DomainError>; } /// Reads the state of the Debian host. diff --git a/backend/crates/infrastructure/src/sqlite.rs b/backend/crates/infrastructure/src/sqlite.rs index 27ec044..a867f91 100644 --- a/backend/crates/infrastructure/src/sqlite.rs +++ b/backend/crates/infrastructure/src/sqlite.rs @@ -393,6 +393,15 @@ impl domain::ports::JobRunRepository for SqliteJobRuns { .map(|r| r.as_ref().map(job_from_row)) .map_err(storage) } + async fn running(&self) -> Result, DomainError> { + sqlx::query(&format!( + "SELECT {JOB_COLS} FROM job_runs WHERE status = 'running'" + )) + .fetch_all(&self.0) + .await + .map(|rows| rows.iter().map(job_from_row).collect()) + .map_err(storage) + } async fn last_finished(&self, kind: JobKind) -> Result, DomainError> { sqlx::query(&format!("SELECT {JOB_COLS} FROM job_runs WHERE kind = ? AND status != 'running' ORDER BY started_at DESC LIMIT 1")) .bind(kind.as_str()) diff --git a/backend/crates/infrastructure/src/trivy.rs b/backend/crates/infrastructure/src/trivy.rs index c40ac4b..08eba6a 100644 --- a/backend/crates/infrastructure/src/trivy.rs +++ b/backend/crates/infrastructure/src/trivy.rs @@ -18,26 +18,133 @@ impl TrivyScanner { } } -/// Parse `trivy ... --format json` output into findings (all results merged, deduplicated). -pub fn parse_trivy_json(_json: &str) -> Result, DomainError> { - todo!() +#[derive(serde::Deserialize)] +struct Report { + #[serde(rename = "Results", default)] + results: Option>, } +#[derive(serde::Deserialize)] +struct ResultEntry { + #[serde(rename = "Vulnerabilities", default)] + vulnerabilities: Option>, +} + +#[derive(serde::Deserialize)] +struct Vuln { + #[serde(rename = "VulnerabilityID")] + id: String, + #[serde(rename = "PkgName", default)] + pkg: String, + #[serde(rename = "InstalledVersion", default)] + installed: String, + #[serde(rename = "FixedVersion", default)] + fixed: Option, + #[serde(rename = "Severity", default)] + severity: String, + #[serde(rename = "Title", default)] + title: Option, + #[serde(rename = "PrimaryURL", default)] + url: Option, +} + +/// Parse `trivy ... --format json` output into findings (all results merged, deduplicated). +pub fn parse_trivy_json(json: &str) -> Result, DomainError> { + let report: Report = serde_json::from_str(json) + .map_err(|e| DomainError::Unavailable(format!("trivy output: {e}")))?; + let mut seen = std::collections::HashSet::new(); + let mut out = Vec::new(); + let vulns = report + .results + .unwrap_or_default() + .into_iter() + .flat_map(|r| r.vulnerabilities.unwrap_or_default()); + for v in vulns { + let f = RawFinding { + title: v + .title + .filter(|t| !t.is_empty()) + .unwrap_or_else(|| format!("{} in {}", v.id, v.pkg)), + url: v + .url + .unwrap_or_else(|| format!("https://nvd.nist.gov/vuln/detail/{}", v.id)), + cve_id: v.id, + severity: Severity::parse(&v.severity), + package: v.pkg, + installed_version: v.installed, + fixed_version: v.fixed.filter(|f| !f.is_empty()), + }; + if seen.insert(f.key()) { + out.push(f); + } + } + Ok(out) +} + +impl TrivyScanner { + async fn scan( + &self, + args: &[&str], + out: &dyn LineSink, + ) -> Result, DomainError> { + out.line(&format!("$ trivy {}", args.join(" "))); + let res = self.runner.run("trivy", args).await?; + if !res.success { + let tail: Vec<&str> = res.stderr.lines().rev().take(5).collect(); + for l in tail.into_iter().rev() { + out.line(l); + } + return Err(DomainError::Unavailable( + "trivy exited with a non-zero status".into(), + )); + } + let findings = parse_trivy_json(&res.stdout)?; + out.line(&format!("{} finding(s)", findings.len())); + Ok(findings) + } +} + +const COMMON: [&str; 7] = [ + "--format", + "json", + "--quiet", + "--scanners", + "vuln", + "--timeout", + "15m", +]; + #[async_trait] impl VulnerabilityScanner for TrivyScanner { async fn version(&self) -> Result { - let _ = &self.runner; - todo!() + let res = self.runner.run("trivy", &["--version"]).await?; + if !res.success { + return Err(DomainError::Unavailable("trivy is not installed".into())); + } + Ok(res + .stdout + .lines() + .next() + .unwrap_or_default() + .trim_start_matches("Version:") + .trim() + .to_string()) } - async fn scan_os(&self, _out: &dyn LineSink) -> Result, DomainError> { - todo!() + async fn scan_os(&self, out: &dyn LineSink) -> Result, DomainError> { + let mut args = vec!["rootfs"]; + args.extend(COMMON); + args.push("/"); + self.scan(&args, out).await } async fn scan_image( &self, - _image: &str, - _out: &dyn LineSink, + image: &str, + out: &dyn LineSink, ) -> Result, DomainError> { - todo!() + let mut args = vec!["image"]; + args.extend(COMMON); + args.push(image); + self.scan(&args, out).await } }