diff --git a/Cargo.lock b/Cargo.lock index d003239..26c1c1e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -155,6 +155,7 @@ dependencies = [ "chumsky", "clap", "env_logger", + "futures", "jenkins_api", "log", "maud", @@ -166,6 +167,7 @@ dependencies = [ "serde_json", "time", "tokio", + "tokio-stream", "toml", ] @@ -392,6 +394,21 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.31" @@ -399,6 +416,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -407,6 +425,40 @@ version = "0.3.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" +[[package]] +name = "futures-executor" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" + +[[package]] +name = "futures-macro" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "futures-sink" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7" + [[package]] name = "futures-task" version = "0.3.31" @@ -419,10 +471,16 @@ version = "0.3.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" dependencies = [ + "futures-channel", "futures-core", + "futures-io", + "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "pin-utils", + "slab", ] [[package]] @@ -1557,6 +1615,17 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-stream" +version = "0.1.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eca58d7bba4a75707817a2c44174253f9236b2d5fbd055602e9d5c07c139a047" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "toml" version = "0.8.23" diff --git a/Cargo.toml b/Cargo.toml index 9874866..ff39884 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,3 +23,5 @@ chumsky = { version = "0.10.1", features = ["pratt"] } rusqlite_regex = "0.6.0" arcstr = "1.2.0" tokio = { version = "1.47.1", features = ["fs", "macros", "process", "rt-multi-thread"] } +tokio-stream = "0.1.17" +futures = "0.3.31" diff --git a/src/api.rs b/src/api.rs index 6b68f6a..f9da757 100644 --- a/src/api.rs +++ b/src/api.rs @@ -109,9 +109,9 @@ impl Job for SparseJob { impl AsJob for SparseJob { fn as_job(&self, last_n: usize) -> crate::db::Job { crate::db::Job { - name: self.name.clone(), + name: self.name.as_str().into(), last_build: self.builds.iter().take(last_n).last().map(|b| b.number), - url: self.url.clone(), + url: self.url.as_str().into(), } } } @@ -136,7 +136,7 @@ where let display_name = self.full_display_name_or_default(); let status = self.build_status(); Run { - url: self.url().to_string(), + url: self.url().into(), status, display_name: display_name.into(), log: match status { diff --git a/src/config.rs b/src/config.rs index 0052da0..5f478ec 100644 --- a/src/config.rs +++ b/src/config.rs @@ -66,7 +66,7 @@ pub struct TagView { pub expr: String, } -/// Represents one tag to be loaded as [crate::parse::Tag] +/// Represents one tag to be loaded as [crate::db::TagInfo] #[derive(Deserialize)] pub struct ConfigTag { /// Unique name of the tag diff --git a/src/db/artifact.rs b/src/db/artifact.rs index 58e6fff..ef81e90 100644 --- a/src/db/artifact.rs +++ b/src/db/artifact.rs @@ -2,7 +2,7 @@ use crate::{db::Queryable, schema}; /// [Artifact] stored in [super::Database] pub struct Artifact { - /// Byte contents of [Artifact] + /// Path of the [Artifact] pub path: String, /// Byte contents of [Artifact] @@ -31,39 +31,39 @@ schema! { } impl Queryable for Artifact { - fn map_row(_: ()) -> impl FnMut(&rusqlite::Row) -> rusqlite::Result> { - |row| { - Ok(super::InDatabase::new( - row.get(0)?, - Artifact { - path: row.get(1)?, - contents: row.get(2)?, - run_id: row.get(3)?, - }, - )) - } + fn map_row(row: &rusqlite::Row) -> rusqlite::Result> { + Ok(super::InDatabase::new( + row.get(0)?, + Artifact { + path: row.get(1)?, + contents: row.get(2)?, + run_id: row.get(3)?, + }, + )) } - fn as_params(&self, _: ()) -> rusqlite::Result { + fn as_params(&self) -> rusqlite::Result { Ok((&self.path, &self.contents, self.run_id)) } } impl Artifact { /// Get all [Artifact] from [super::Database] by [super::Run] - pub fn select_all_by_run( + pub async fn select_all_by_run( db: &super::Database, run_id: i64, - params: (), ) -> rusqlite::Result>> { - db.prepare_cached( - " + db.call(move |conn| { + conn.prepare_cached( + " SELECT * FROM artifacts WHERE run_id = ? ", - )? - .query_map((run_id,), Self::map_row(params))? - .collect() + )? + .query_map((run_id,), Self::map_row)? + .collect() + }) + .await } /// Gets the [BlobFormat] of the [Artifact] diff --git a/src/db/build.rs b/src/db/build.rs index d2dd52c..a5bf9d5 100644 --- a/src/db/build.rs +++ b/src/db/build.rs @@ -6,6 +6,7 @@ use crate::{ }; /// [JobBuild] in [super::Database] +#[derive(Clone)] pub struct JobBuild { /// Build url pub url: String, @@ -35,22 +36,20 @@ schema! { } impl Queryable for JobBuild { - fn map_row(_: ()) -> impl FnMut(&rusqlite::Row) -> rusqlite::Result> { - |row| { - Ok(super::InDatabase::new( - row.get(0)?, - JobBuild { - url: row.get(1)?, - status: read_value!(row, 2), - number: row.get(3)?, - timestamp: row.get(4).map(i64::cast_unsigned)?, - job_id: row.get(5)?, - }, - )) - } + fn map_row(row: &rusqlite::Row) -> rusqlite::Result> { + Ok(super::InDatabase::new( + row.get(0)?, + JobBuild { + url: row.get(1)?, + status: read_value!(row, 2), + number: row.get(3)?, + timestamp: row.get(4).map(i64::cast_unsigned)?, + job_id: row.get(5)?, + }, + )) } - fn as_params(&self, _: ()) -> rusqlite::Result { + fn as_params(&self) -> rusqlite::Result { Ok(( &self.url, write_value!(self.status), @@ -62,105 +61,118 @@ impl Queryable for JobBuild { } impl Upsertable for JobBuild { - fn upsert(self, db: &super::Database, params: ()) -> rusqlite::Result> { - db.prepare_cached( - " - INSERT INTO builds ( - url, - status, - number, - timestamp, - job_id - ) VALUES (?, ?, ?, ?, ?) - ON CONFLICT(url) DO UPDATE SET - status = excluded.status, - number = excluded.number, - timestamp = excluded.timestamp, - job_id = excluded.job_id - ", - )? - .execute(self.as_params(params)?)?; + async fn upsert(self, db: &super::Database) -> rusqlite::Result> { + let number = self.number; + let job_id = self.job_id; + + db.call(move |conn| { + conn.prepare_cached( + " + INSERT INTO builds ( + url, + status, + number, + timestamp, + job_id + ) VALUES (?, ?, ?, ?, ?) + ON CONFLICT(url) DO UPDATE SET + status = excluded.status, + number = excluded.number, + timestamp = excluded.timestamp, + job_id = excluded.job_id + ", + )? + .execute(self.as_params()?) + }) + .await?; - Self::select_one_by_job(db, self.job_id, self.number, ()) + Self::select_one_by_job(db, job_id, number).await } } impl JobBuild { /// Get a [JobBuild] from [super::Database] by [super::Job] id and build number - pub fn select_one_by_job( + pub async fn select_one_by_job( db: &super::Database, job_id: i64, number: u32, - params: (), ) -> rusqlite::Result> { - db.prepare_cached( - " - SELECT * FROM builds - WHERE job_id = ? - AND number = ? - ORDER BY number DESC - ", - )? - .query_one((job_id, number), Self::map_row(params)) + db.call(move |conn| { + conn.prepare_cached( + " + SELECT * FROM builds + WHERE job_id = ? + AND number = ? + ORDER BY number DESC + ", + )? + .query_one((job_id, number), Self::map_row) + }) + .await } /// Get all [JobBuild] from [super::Database] by [super::Job] - pub fn select_all_by_job( + pub async fn select_all_by_job( db: &super::Database, job_id: i64, - params: (), ) -> rusqlite::Result>> { - db.prepare_cached( - " - SELECT * FROM builds - WHERE job_id = ? - ORDER BY number DESC - ", - )? - .query_map((job_id,), Self::map_row(params))? - .collect() + db.call(move |conn| { + conn.prepare_cached( + " + SELECT * FROM builds + WHERE job_id = ? + ORDER BY number DESC + ", + )? + .query_map((job_id,), Self::map_row)? + .collect() + }) + .await } /// Remove all [JobBuild]s which aren't referenced by [super::Job] from [super::Database] - pub fn delete_all_orphan(db: &super::Database) -> rusqlite::Result<()> { - db.execute_batch( - " - BEGIN; - DELETE FROM similarities WHERE similarity_hash IN ( - SELECT DISTINCT similarities.similarity_hash FROM similarities - JOIN issues ON issues.id = similarities.issue_id - JOIN runs ON runs.id = issues.run_id - JOIN builds ON builds.id = runs.build_id - JOIN jobs ON jobs.id = builds.job_id - WHERE number < last_build - ); - DELETE FROM issues WHERE id IN ( - SELECT issues.id FROM issues - JOIN runs ON runs.id = issues.run_id - JOIN builds ON builds.id = runs.build_id - JOIN jobs ON jobs.id = builds.job_id - WHERE number < last_build - ); - DELETE FROM artifacts WHERE id IN ( - SELECT artifacts.id FROM artifacts - JOIN runs ON runs.id = artifacts.run_id - JOIN builds ON builds.id = runs.build_id - JOIN jobs ON jobs.id = builds.job_id - WHERE number < last_build - ); - DELETE FROM runs WHERE id IN ( - SELECT runs.id FROM runs - JOIN builds ON builds.id = runs.build_id - JOIN jobs ON jobs.id = builds.job_id - WHERE number < last_build - ); - DELETE FROM builds WHERE id IN ( - SELECT builds.id FROM builds - JOIN jobs ON jobs.id = builds.job_id - WHERE number < last_build - ); - COMMIT; - ", - ) + pub async fn delete_all_orphan(db: &super::Database) -> rusqlite::Result<()> { + db.call(|conn| { + conn.execute_batch( + " + BEGIN; + DELETE FROM similarities WHERE similarity_hash IN ( + SELECT DISTINCT similarities.similarity_hash FROM similarities + JOIN issues ON issues.id = similarities.issue_id + JOIN runs ON runs.id = issues.run_id + JOIN builds ON builds.id = runs.build_id + JOIN jobs ON jobs.id = builds.job_id + WHERE number < last_build + ); + DELETE FROM issues WHERE id IN ( + SELECT issues.id FROM issues + JOIN runs ON runs.id = issues.run_id + JOIN builds ON builds.id = runs.build_id + JOIN jobs ON jobs.id = builds.job_id + WHERE number < last_build + ); + DELETE FROM artifacts WHERE id IN ( + SELECT artifacts.id FROM artifacts + JOIN runs ON runs.id = artifacts.run_id + JOIN builds ON builds.id = runs.build_id + JOIN jobs ON jobs.id = builds.job_id + WHERE number < last_build + ); + DELETE FROM runs WHERE id IN ( + SELECT runs.id FROM runs + JOIN builds ON builds.id = runs.build_id + JOIN jobs ON jobs.id = builds.job_id + WHERE number < last_build + ); + DELETE FROM builds WHERE id IN ( + SELECT builds.id FROM builds + JOIN jobs ON jobs.id = builds.job_id + WHERE number < last_build + ); + COMMIT; + ", + ) + }) + .await } } diff --git a/src/db/issue.rs b/src/db/issue.rs index 8c06069..29ed0f6 100644 --- a/src/db/issue.rs +++ b/src/db/issue.rs @@ -1,15 +1,22 @@ -use std::str::from_utf8; - -use arcstr::Substr; +use arcstr::{ArcStr, Substr}; use crate::{ - config::{Field, Severity}, - db::{Artifact, Queryable, Run, TagInfo}, + config::Severity, + db::{Artifact, Queryable, Run}, schema, write_value, }; /// [Issue] stored in [super::Database] -#[derive(PartialEq, Eq, Hash)] +pub struct IssueInfo { + pub snippet_start: usize, + pub snippet_end: usize, + pub run_id: i64, + pub artifact_id: Option, + pub tag_id: i64, + pub duplicates: u64, +} + +#[derive(Debug, PartialEq, Eq, Hash)] pub struct Issue { /// String snippet from [Run] pub snippet: Substr, @@ -22,7 +29,7 @@ pub struct Issue { } schema! { - issues for Issue { + issues for IssueInfo { id INTEGER PRIMARY KEY, snippet_start INTEGER NOT NULL, snippet_end INTEGER NOT NULL, @@ -33,152 +40,168 @@ schema! { } } -impl - Queryable< - (&super::Database, &super::InDatabase), - ( - &super::InDatabase, - Option<&super::InDatabase>, - ), - > for Issue -{ - fn map_row( - params: (&super::Database, &super::InDatabase), - ) -> impl FnMut(&rusqlite::Row) -> rusqlite::Result> { - let (db, run) = params; - |row| { - let tag_id = row.get(5)?; - Ok(super::InDatabase::new( - row.get(0)?, - Self { - snippet: match TagInfo::select_one(db, tag_id, ())?.field { - Field::Console => run.log.clone().ok_or(rusqlite::Error::InvalidQuery)?, - Field::RunName => run.display_name.clone(), - Field::Artifact => { - from_utf8(&Artifact::select_one(db, row.get(4)?, ())?.contents) - .map_err(|_| rusqlite::Error::InvalidQuery)? - .into() - } - } - .substr(row.get::<_, usize>(1)?..row.get::<_, usize>(2)?), - tag_id, - duplicates: row.get(6).map(i64::cast_unsigned)?, - }, - )) - } +impl Queryable for IssueInfo { + fn map_row(row: &rusqlite::Row) -> rusqlite::Result> { + Ok(super::InDatabase::new( + row.get(0)?, + Self { + snippet_start: row.get(1).map(isize::cast_unsigned)?, + snippet_end: row.get(2).map(isize::cast_unsigned)?, + run_id: row.get(3)?, + artifact_id: row.get(4)?, + tag_id: row.get(5)?, + duplicates: row.get(6).map(i64::cast_unsigned)?, + }, + )) } - fn as_params( - &self, - params: ( - &super::InDatabase, - Option<&super::InDatabase>, - ), - ) -> rusqlite::Result { - let (run, artifact) = params; - let core::ops::Range::<_> { start, end } = self.snippet.range(); + fn as_params(&self) -> rusqlite::Result { Ok(( - start, - end, - run.id, - artifact.map(|a| a.id), + self.snippet_start.cast_signed(), + self.snippet_end.cast_signed(), + self.run_id, + self.artifact_id, self.tag_id, self.duplicates.cast_signed(), )) } - - fn select_all( - db: &super::Database, - params: (&super::Database, &super::InDatabase), - ) -> rusqlite::Result>> { - let (_, run) = params; - db.prepare_cached( - " - SELECT - issues.id, - snippet_start, - snippet_end, - run_id, - artifact_id, - tag_id, - duplicates - FROM issues - JOIN tags ON tags.id = issues.tag_id - WHERE issues.run_id = ? - ", - )? - .query_map((run.id,), Self::map_row(params))? - .collect() - } } -impl Issue { - /// Get all [Issue]s from [super::Database] that aren't [Severity::Metadata] - pub fn select_all_not_metadata( +impl IssueInfo { + /// Get all [Issue]s from [super::Database] by [Run] + pub async fn select_all_by_run( db: &super::Database, - params: (&super::Database, &super::InDatabase), + run: &super::InDatabase, + include_metadata: bool, ) -> rusqlite::Result>> { - let (_, run) = params; - db.prepare_cached( - " - SELECT - issues.id, - snippet_start, - snippet_end, - run_id, - artifact_id, - tag_id, - duplicates - FROM issues - JOIN tags ON tags.id = issues.tag_id - WHERE issues.run_id = ? - AND tags.severity != ? - ", - )? - .query_map( - (run.id, write_value!(Severity::Metadata)), - Self::map_row(params), - )? - .collect() + let id = run.id; + if include_metadata { + db.call(move |conn| { + conn.prepare_cached( + " + SELECT + id, + snippet_start, + snippet_end, + run_id, + artifact_id, + tag_id, + duplicates + FROM issues + WHERE run_id = ? + ", + )? + .query_map((id,), Self::map_row)? + .collect() + }) + .await + } else { + db.call(move |conn| { + conn.prepare_cached( + " + SELECT + issues.id, + snippet_start, + snippet_end, + run_id, + artifact_id, + tag_id, + duplicates + FROM issues + JOIN tags ON tags.id = issues.tag_id + WHERE issues.run_id = ? + AND tags.severity != ? + ", + )? + .query_map((id, write_value!(Severity::Metadata)), Self::map_row)? + .collect() + }) + .await + } } /// Remove all [Issue]s with an outdated [crate::parse::TagSet] schema from [super::Database] - pub fn delete_all_invalid_by_tag_schema( - db: &mut super::Database, + pub async fn delete_all_invalid_by_tag_schema( + db: &super::Database, current_schema: u64, ) -> rusqlite::Result { - let mut tx = db.transaction()?; - tx.set_drop_behavior(rusqlite::DropBehavior::Commit); + db.call(move |conn| { + let mut tx = conn.transaction()?; + tx.set_drop_behavior(rusqlite::DropBehavior::Commit); - // delete similarities first - tx.execute( - " - DELETE FROM similarities WHERE similarity_hash IN ( - SELECT DISTINCT similarities.similarity_hash FROM similarities - JOIN issues ON issues.id = similarities.issue_id - JOIN runs ON runs.id = issues.run_id - WHERE runs.tag_schema != ? - ) - ", - (current_schema.cast_signed(),), - )?; + // delete similarities first + tx.execute( + " + DELETE FROM similarities WHERE similarity_hash IN ( + SELECT DISTINCT similarities.similarity_hash FROM similarities + JOIN issues ON issues.id = similarities.issue_id + JOIN runs ON runs.id = issues.run_id + WHERE runs.tag_schema != ? + ) + ", + (current_schema.cast_signed(),), + )?; + + // then issues + tx.execute( + " + DELETE FROM issues WHERE id IN ( + SELECT i.id FROM issues i + JOIN runs r ON i.run_id = r.id + WHERE r.tag_schema != ? + ) + ", + (current_schema.cast_signed(),), + )?; - // then issues - tx.execute( - " - DELETE FROM issues WHERE id IN ( - SELECT i.id FROM issues i - JOIN runs r ON i.run_id = r.id - WHERE r.tag_schema != ? + // also set the run tag_schema to NULL to indicate an unparsed run + tx.execute( + "UPDATE runs SET tag_schema = NULL WHERE tag_schema != ?", + (current_schema.cast_signed(),), ) - ", - (current_schema.cast_signed(),), - )?; + }) + .await + } +} - // also set the run tag_schema to NULL to indicate an unparsed run - tx.execute( - "UPDATE runs SET tag_schema = NULL WHERE tag_schema != ?", - (current_schema.cast_signed(),), +impl super::InDatabase { + pub fn into_issue(self, field: &ArcStr) -> super::InDatabase { + let super::InDatabase { + id, + item: + IssueInfo { + snippet_start, + snippet_end, + tag_id, + duplicates, + .. + }, + } = self; + super::InDatabase::new( + id, + Issue { + snippet: field.substr(snippet_start..snippet_end), + tag_id, + duplicates, + }, ) } } + +impl Issue { + pub fn into_issue_info( + self, + run: &super::InDatabase, + artifact: Option<&super::InDatabase>, + ) -> IssueInfo { + let core::ops::Range::<_> { start, end } = self.snippet.range(); + IssueInfo { + snippet_start: start, + snippet_end: end, + run_id: run.id, + artifact_id: artifact.map(|a| a.id), + tag_id: self.tag_id, + duplicates: self.duplicates, + } + } +} diff --git a/src/db/job.rs b/src/db/job.rs index f1e1586..8afa494 100644 --- a/src/db/job.rs +++ b/src/db/job.rs @@ -1,15 +1,19 @@ +use arcstr::ArcStr; +use futures::{StreamExt, TryStreamExt, stream}; + use crate::{ db::{Queryable, Upsertable}, schema, }; /// [Job] stored in [super::Database] +#[derive(Clone)] pub struct Job { /// Unique name of [Job] - pub name: String, + pub name: ArcStr, /// [Job] url - pub url: String, + pub url: ArcStr, /// Number of last [super::JobBuild] pub last_build: Option, @@ -25,28 +29,28 @@ schema! { } impl Queryable for Job { - fn map_row(_: ()) -> impl FnMut(&rusqlite::Row) -> rusqlite::Result> { - |row| { - Ok(super::InDatabase::new( - row.get(0)?, - Job { - name: row.get(1)?, - url: row.get(2)?, - last_build: row.get(3)?, - }, - )) - } + fn map_row(row: &rusqlite::Row) -> rusqlite::Result> { + Ok(super::InDatabase::new( + row.get(0)?, + Job { + name: row.get::<_, String>(1)?.into(), + url: row.get::<_, String>(2)?.into(), + last_build: row.get(3)?, + }, + )) } - fn as_params(&self, _: ()) -> rusqlite::Result { - Ok((&self.name, &self.url, self.last_build)) + fn as_params(&self) -> rusqlite::Result { + Ok((self.name.as_str(), self.url.as_str(), self.last_build)) } } impl Upsertable for Job { - fn upsert(self, db: &super::Database, params: ()) -> rusqlite::Result> { - db.prepare_cached( - " + async fn upsert(self, db: &super::Database) -> rusqlite::Result> { + let name = self.name.clone(); + db.call(move |conn| { + conn.prepare_cached( + " INSERT INTO jobs ( name, url, @@ -55,109 +59,122 @@ impl Upsertable for Job { ON CONFLICT(name) DO UPDATE SET last_build = excluded.last_build ", - )? - .execute(self.as_params(params)?)?; + )? + .execute(self.as_params()?) + }) + .await?; // get the job as a second query in-case of an insert conflict - Self::select_one_by_name(db, &self.name, ()) + Self::select_one_by_name(db, name).await } } impl Job { /// Get a [Job] from [super::Database] by name - pub fn select_one_by_name( + pub async fn select_one_by_name( db: &super::Database, - name: &str, - params: (), + name: ArcStr, ) -> rusqlite::Result> { - db.prepare_cached( - " + db.call(move |conn| { + conn.prepare_cached( + " SELECT * FROM jobs WHERE name = ? ", - )? - .query_one((name,), Self::map_row(params)) + )? + .query_one((name.as_str(),), Self::map_row) + }) + .await } /// Remove all [Job]s from [super::Database] by name - pub fn delete_all_by_blocklist( - db: &mut super::Database, - names: &[String], - ) -> rusqlite::Result { - let mut tx = db.transaction()?; - tx.set_drop_behavior(rusqlite::DropBehavior::Commit); - - names.iter().try_fold(0, |acc, name| { - // delete similarities first - tx.execute( - " - DELETE FROM similarities WHERE similarity_hash IN ( - SELECT DISTINCT similarities.similarity_hash FROM similarities - JOIN issues ON issues.id = similarities.issue_id - JOIN runs ON runs.id = issues.run_id - JOIN builds ON builds.id = runs.build_id - JOIN jobs ON jobs.id = builds.job_id - WHERE name = ? - ) - ", - (name,), - )?; - - // then issues - tx.execute( - " - DELETE FROM issues WHERE id IN ( - SELECT issues.id FROM issues - JOIN runs ON runs.id = issues.run_id - JOIN builds ON builds.id = runs.build_id - JOIN jobs ON jobs.id = builds.job_id - WHERE name = ? - ) - ", - (name,), - )?; - - // then artifacts - tx.execute( - " - DELETE FROM artifacts WHERE id IN ( - SELECT artifacts.id FROM artifacts - JOIN runs ON runs.id = artifacts.run_id - JOIN builds ON builds.id = runs.build_id - JOIN jobs ON jobs.id = builds.job_id - WHERE name = ? - ); - ", - (name,), - )?; - - // then runs - tx.execute( - " - DELETE FROM runs WHERE id IN ( - SELECT runs.id FROM runs - JOIN builds ON builds.id = runs.build_id - JOIN jobs ON jobs.id = builds.job_id - WHERE name = ? - ); - ", - (name,), - )?; - - // then builds - tx.execute( - " - DELETE FROM builds WHERE id IN ( - SELECT builds.id FROM builds - JOIN jobs ON jobs.id = builds.job_id - WHERE name = ? - ); - ", - (name,), - )?; - - // finally the job - Ok(acc + tx.execute("DELETE FROM jobs WHERE name = ?", (name,))?) - }) + pub async fn delete_all_by_blocklist( + db: &super::Database, + names: I, + ) -> rusqlite::Result + where + I: IntoIterator, + { + stream::iter(names) + .then(|name| async { + db.call(move |conn| { + let mut tx = conn.unchecked_transaction()?; + tx.set_drop_behavior(rusqlite::DropBehavior::Commit); + + // delete similarities first + tx.execute( + " + DELETE FROM similarities WHERE similarity_hash IN ( + SELECT DISTINCT similarities.similarity_hash FROM similarities + JOIN issues ON issues.id = similarities.issue_id + JOIN runs ON runs.id = issues.run_id + JOIN builds ON builds.id = runs.build_id + JOIN jobs ON jobs.id = builds.job_id + WHERE name = ? + ) + ", + (&name,), + )?; + + // then issues + tx.execute( + " + DELETE FROM issues WHERE id IN ( + SELECT issues.id FROM issues + JOIN runs ON runs.id = issues.run_id + JOIN builds ON builds.id = runs.build_id + JOIN jobs ON jobs.id = builds.job_id + WHERE name = ? + ) + ", + (&name,), + )?; + + // then artifacts + tx.execute( + " + DELETE FROM artifacts WHERE id IN ( + SELECT artifacts.id FROM artifacts + JOIN runs ON runs.id = artifacts.run_id + JOIN builds ON builds.id = runs.build_id + JOIN jobs ON jobs.id = builds.job_id + WHERE name = ? + ); + ", + (&name,), + )?; + + // then runs + tx.execute( + " + DELETE FROM runs WHERE id IN ( + SELECT runs.id FROM runs + JOIN builds ON builds.id = runs.build_id + JOIN jobs ON jobs.id = builds.job_id + WHERE name = ? + ); + ", + (&name,), + )?; + + // then builds + tx.execute( + " + DELETE FROM builds WHERE id IN ( + SELECT builds.id FROM builds + JOIN jobs ON jobs.id = builds.job_id + WHERE name = ? + ); + ", + (&name,), + )?; + + // finally the job + tx.execute("DELETE FROM jobs WHERE name = ?", (&name,)) + }) + .await + }) + .try_fold(0, |acc, x| async move { Ok(acc + x) }) + .await } } diff --git a/src/db/mod.rs b/src/db/mod.rs index 13fc25e..7088a1d 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -1,8 +1,11 @@ //! [rusqlite] based ORM to cache build results. use std::hash::Hash; use std::ops::{Deref, DerefMut}; +use std::path::Path; +use std::thread; use rusqlite::{Connection, Params, Result, Row}; +use tokio::sync::{mpsc, oneshot}; mod artifact; mod build; @@ -156,33 +159,36 @@ macro_rules! for_all { }; ($($method:tt)+) => { - for_all!([SimilarityInfo, Issue, Artifact, Run, JobBuild, Job, TagInfo] => $($method)+) + for_all!([SimilarityInfo, IssueInfo, Artifact, Run, JobBuild, Job, TagInfo] => $($method)+) }; } -/// Database object -pub struct Database { - /// Internal [rusqlite] connection - conn: Connection, +/// Messages the [Database] object passes +trait Message: Send { + /// Call closure on background thread with [Connection] and send back result + fn call(self: Box, conn: &mut Connection); } -/// Implicit deref to [Connection] from [Database] -impl Deref for Database { - type Target = Connection; - - fn deref(&self) -> &Self::Target { - &self.conn +impl Message for (F, oneshot::Sender) +where + F: FnOnce(&mut Connection) -> R + Send, + R: Send, +{ + fn call(self: Box, conn: &mut Connection) { + let (closure, tx) = *self; + tx.send(closure(conn)); } } -/// Implicit deref_mut to [Connection] from [Database] -impl DerefMut for Database { - fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.conn - } +/// Database object +#[derive(Clone)] +pub struct Database { + /// Sender to connection background thread + tx: mpsc::UnboundedSender>, } /// Represents an item `T` in [Database] +#[derive(Debug, Copy, Clone)] pub struct InDatabase { /// Row ID of `item` pub id: i64, @@ -248,29 +254,62 @@ impl Eq for InDatabase {} impl Database { /// Open or create an `sqlite3` database at `path` returning [Database] - pub fn open(path: &str) -> Result { - // Enable REGEXP - rusqlite_regex::enable_auto_extension()?; - - // try to open existing, otherwise create a new one - let db = Database { - conn: Connection::open(path)?, - }; + pub async fn open>(path: P) -> Result { + let (tx, mut rx) = mpsc::unbounded_channel::>(); + let (res_tx, res_rx) = oneshot::channel(); + + // spawn a background thread to access the database + let path = path.as_ref().to_owned(); + thread::spawn(move || { + // enable regex, then try to open existing file, otherwise create a new one + let mut conn = match rusqlite_regex::enable_auto_extension() + .and_then(|_| Connection::open(path)) + { + Ok(c) => { + res_tx.send(Ok(())).unwrap(); + c + } + Err(e) => { + res_tx.send(Err(e)).unwrap(); + return; + } + }; + + // Handle calls + while let Some(msg) = rx.blocking_recv() { + msg.call(&mut conn); + } + }); + + // ensure successful + let db = res_rx.await.unwrap().map(|_| Self { tx })?; // create the necessary tables - for_all!(create_table(&db)?); + for_all!(create_table(&db).await?); Ok(db) } + /// Get an async context to the [Database]'s [Connection] + pub async fn call(&self, f: F) -> R + where + F: FnOnce(&mut Connection) -> R + Send + 'static, + R: Send + 'static, + { + let (tx, rx) = oneshot::channel(); + self.tx.send(Box::new((f, tx))).unwrap(); + + rx.await.unwrap() + } + /// Purge all rows (but not tables) from [Database] - pub fn purge_cache(&self) -> Result<()> { - for_all!(delete_all(self)?); + pub async fn purge_cache(&self) -> Result<()> { + for_all!(delete_all(self).await?); Ok(()) } } -pub trait Schema: Sized { +pub trait Schema: Send + Sized { const CREATE_TABLE: &'static str; const INSERT: &'static str; const SELECT_ONE: &'static str; @@ -278,45 +317,57 @@ pub trait Schema: Sized { const DELETE_ALL: &'static str; /// Creates the table in [Database] - fn create_table(db: &Database) -> Result { - db.execute(Self::CREATE_TABLE, ()) + async fn create_table(db: &Database) -> Result { + db.call(|conn| conn.execute(Self::CREATE_TABLE, ())).await } } -pub trait Queryable: Schema { +pub trait Queryable: Schema +where + Self: 'static, +{ /// Convert [Row] to `Self` with params - fn map_row(params: I) -> impl FnMut(&Row) -> Result>; + fn map_row(row: &Row) -> Result>; /// Convert `self` to [Params] with `params` - fn as_params(&self, params: E) -> Result; + fn as_params(&self) -> Result; /// Insert `self` to [Database] with `params` - fn insert(self, db: &Database, params: E) -> Result> { - db.prepare_cached(Self::INSERT)? - .execute(self.as_params(params)?)?; - Ok(InDatabase::new(db.last_insert_rowid(), self)) + async fn insert(self, db: &Database) -> Result> { + db.call(|conn| { + conn.prepare_cached(Self::INSERT)? + .execute(self.as_params()?)?; + Ok(InDatabase::new(conn.last_insert_rowid(), self)) + }) + .await } - /// Select one of `Self` from [Database] by `id` with `params` - fn select_one(db: &Database, id: i64, params: I) -> Result> { - db.prepare_cached(Self::SELECT_ONE)? - .query_one((id,), Self::map_row(params)) + /// Select one of `Self` from [Database] by `id` + async fn select_one(db: &Database, id: i64) -> Result> { + db.call(move |conn| { + conn.prepare_cached(Self::SELECT_ONE)? + .query_one((id,), Self::map_row) + }) + .await } - /// Select all of `Self` from [Database] with `params` - fn select_all(db: &Database, params: I) -> Result>> { - db.prepare_cached(Self::SELECT_ALL)? - .query_map((), Self::map_row(params))? - .collect() + /// Select all of `Self` from [Database] + async fn select_all(db: &Database) -> Result>> { + db.call(|conn| { + conn.prepare_cached(Self::SELECT_ALL)? + .query_and_then((), Self::map_row)? + .collect() + }) + .await } /// Delete all of `Self` from [Database] - fn delete_all(db: &Database) -> Result { - db.execute(Self::DELETE_ALL, ()) + async fn delete_all(db: &Database) -> Result { + db.call(|conn| conn.execute(Self::DELETE_ALL, ())).await } } -pub trait Upsertable: Queryable { - /// Upsert `self` to [Database] with `params` - fn upsert(self, db: &Database, params: E) -> Result>; +pub trait Upsertable: Queryable { + /// Upsert `self` to [Database] + async fn upsert(self, db: &Database) -> Result>; } diff --git a/src/db/run.rs b/src/db/run.rs index 4a20586..2dfbf14 100644 --- a/src/db/run.rs +++ b/src/db/run.rs @@ -1,17 +1,22 @@ +use std::str::from_utf8; + use arcstr::ArcStr; +use futures::{StreamExt, TryFutureExt, TryStreamExt}; use jenkins_api::build::BuildStatus; use crate::{ - db::{JobBuild, Queryable, Upsertable}, + config::Field, + db::{Artifact, Issue, IssueInfo, JobBuild, Queryable, TagInfo, Upsertable}, read_value, schema, tag_expr::TagExpr, write_value, }; /// [Run] stored in [super::Database] +#[derive(Clone)] pub struct Run { /// Run url - pub url: String, + pub url: ArcStr, /// Build status pub status: Option, @@ -42,25 +47,23 @@ schema! { } impl Queryable for Run { - fn map_row(_: ()) -> impl FnMut(&rusqlite::Row) -> rusqlite::Result> { - |row| { - Ok(super::InDatabase::new( - row.get(0)?, - Run { - url: row.get(1)?, - status: read_value!(row, 2), - display_name: row.get::<_, String>(3)?.into(), - log: row.get::<_, Option>(4)?.map(ArcStr::from), - tag_schema: row.get::<_, Option>(5)?.map(i64::cast_unsigned), - build_id: row.get(6)?, - }, - )) - } + fn map_row(row: &rusqlite::Row) -> rusqlite::Result> { + Ok(super::InDatabase::new( + row.get(0)?, + Run { + url: row.get::<_, String>(1)?.into(), + status: read_value!(row, 2), + display_name: row.get::<_, String>(3)?.into(), + log: row.get::<_, Option>(4)?.map(ArcStr::from), + tag_schema: row.get::<_, Option>(5)?.map(i64::cast_unsigned), + build_id: row.get(6)?, + }, + )) } - fn as_params(&self, _: ()) -> rusqlite::Result { + fn as_params(&self) -> rusqlite::Result { Ok(( - &self.url, + self.url.as_str(), write_value!(self.status), self.display_name.as_str(), self.log.as_ref().map(|s| s.as_str()), @@ -71,9 +74,11 @@ impl Queryable for Run { } impl Upsertable for Run { - fn upsert(self, db: &super::Database, params: ()) -> rusqlite::Result> { - db.prepare_cached( - " + async fn upsert(self, db: &super::Database) -> rusqlite::Result> { + let url = self.url.clone(); + db.call(move |conn| { + conn.prepare_cached( + " INSERT INTO runs ( url, status, @@ -89,82 +94,147 @@ impl Upsertable for Run { tag_schema = excluded.tag_schema, build_id = excluded.build_id ", - )? - .execute(self.as_params(params)?)?; + )? + .execute(self.as_params()?) + }) + .await?; - Self::select_one_by_url(db, &self.url, ()) + Self::select_one_by_url(db, url).await } } impl Run { /// Get a [Run] from [super::Database] by url - pub fn select_one_by_url( + pub async fn select_one_by_url( db: &super::Database, - url: &str, - params: (), + url: ArcStr, ) -> rusqlite::Result> { - db.prepare_cached( - " + db.call(move |conn| { + conn.prepare_cached( + " SELECT * FROM runs WHERE url = ? ", - )? - .query_one((url,), Self::map_row(params)) + )? + .query_one((url.as_str(),), Self::map_row) + }) + .await } /// Get all [Run]s by [super::JobBuild] - pub fn select_all_by_build( + pub async fn select_all_by_build( db: &super::Database, build: &super::InDatabase, - params: (), ) -> rusqlite::Result>> { - db.prepare_cached( - " + let id = build.id; + db.call(move |conn| { + conn.prepare_cached( + " SELECT * FROM runs WHERE build_id = ? ", - )? - .query_map((build.id,), Self::map_row(params))? - .collect() + )? + .query_and_then((id,), Self::map_row)? + .collect() + }) + .await } /// Get all [Run] ID by [TagExpr] in [super::Database] - pub fn select_all_id_by_expr( + pub async fn select_all_id_by_expr( db: &super::Database, expr: &TagExpr, ) -> rusqlite::Result> { let (stmt, params) = expr.to_sql_select()?; - db.prepare(&stmt)? - .query_map(params, |row| row.get(0))? - .collect() + db.call(move |conn| { + conn.prepare(&stmt)? + .query_and_then(rusqlite::params_from_iter(params), |row| row.get(0))? + .collect() + }) + .await } /// Get a [Run]'s display name by id in [super::Database] - pub fn select_one_display_name(db: &super::Database, id: i64) -> rusqlite::Result { - db.prepare_cached("SELECT display_name FROM runs WHERE id = ?")? - .query_one((id,), |row| row.get(0)) + pub async fn select_one_display_name( + db: &super::Database, + id: i64, + ) -> rusqlite::Result { + db.call(move |conn| { + conn.prepare_cached("SELECT display_name FROM runs WHERE id = ?")? + .query_one((id,), |row| row.get(0)) + }) + .await } /// Get a [Run]'s url by id in [super::Database] - pub fn select_one_url(db: &super::Database, id: i64) -> rusqlite::Result { - db.prepare_cached("SELECT url FROM runs WHERE id = ?")? - .query_one((id,), |row| row.get(0)) + pub async fn select_one_url(db: &super::Database, id: i64) -> rusqlite::Result { + db.call(move |conn| { + conn.prepare_cached("SELECT url FROM runs WHERE id = ?")? + .query_one((id,), |row| row.get(0)) + }) + .await } /// Check whether or not there are untagged [Run]s in [super::Database] - pub fn has_untagged(db: &super::Database) -> rusqlite::Result { - db.prepare_cached("SELECT 1 FROM runs WHERE tag_schema IS NULL")? - .exists(()) + pub async fn has_untagged(db: &super::Database) -> rusqlite::Result { + db.call(|conn| { + conn.prepare_cached("SELECT 1 FROM runs WHERE tag_schema IS NULL")? + .exists(()) + }) + .await } /// Update the [crate::parse::TagSet] schema for all [Run]s in [super::Database] - pub fn update_all_tag_schema( + pub async fn update_all_tag_schema( db: &super::Database, new_schema: Option, ) -> rusqlite::Result { - db.execute( - "UPDATE runs SET tag_schema = ?", - (new_schema.map(u64::cast_signed),), - ) + db.call(move |conn| { + conn.execute( + "UPDATE runs SET tag_schema = ?", + (new_schema.map(u64::cast_signed),), + ) + }) + .await + } +} + +impl super::InDatabase { + /// Get a [Run]'s [Issue]s in [super::Database] + pub async fn select_all_issues( + &self, + db: &super::Database, + include_metadata: bool, + ) -> rusqlite::Result>> { + IssueInfo::select_all_by_run(db, self, include_metadata) + .and_then(|issues| async { + futures::stream::iter(issues) + .then(|i| async { + match TagInfo::select_one(db, i.tag_id).await?.field { + Field::Console => { + self.log + .as_ref() + .map_or(Err(rusqlite::Error::InvalidQuery), |l| { + Ok(i.into_issue(l)) + }) + } + Field::RunName => Ok(i.into_issue(&self.display_name)), + Field::Artifact => { + let artifact_id = + i.artifact_id.ok_or(rusqlite::Error::InvalidQuery)?; + Ok(i.into_issue( + // TODO: share Vec contents somehow? + &from_utf8( + &Artifact::select_one(db, artifact_id).await?.contents, + ) + .map(ArcStr::from)?, + )) + } + } + }) + .try_collect() + .await + }) + .await } } diff --git a/src/db/similarity.rs b/src/db/similarity.rs index 44f3d11..2861e8a 100644 --- a/src/db/similarity.rs +++ b/src/db/similarity.rs @@ -1,12 +1,16 @@ use std::{ - cmp::Reverse, collections::{HashMap, HashSet}, + str::from_utf8, }; -use arcstr::Substr; +use arcstr::{ArcStr, Substr}; +use futures::{TryFutureExt, TryStreamExt}; +use tokio::sync::mpsc; +use tokio_stream::wrappers::UnboundedReceiverStream; use crate::{ - db::{InDatabase, Issue, Queryable, Run, TagInfo}, + config::Field, + db::{Artifact, InDatabase, IssueInfo, Queryable, Run, TagInfo}, schema, }; @@ -31,88 +35,132 @@ schema! { } impl Queryable for SimilarityInfo { - fn map_row(_: ()) -> impl FnMut(&rusqlite::Row) -> rusqlite::Result> { - |row| { - Ok(InDatabase::new( - row.get(0)?, - Self { - similarity_hash: row.get(1).map(i64::cast_unsigned)?, - issue_id: row.get(2)?, - }, - )) - } + fn map_row(row: &rusqlite::Row) -> rusqlite::Result> { + Ok(InDatabase::new( + row.get(0)?, + Self { + similarity_hash: row.get(1).map(i64::cast_unsigned)?, + issue_id: row.get(2)?, + }, + )) } - fn as_params(&self, _: ()) -> rusqlite::Result { + fn as_params(&self) -> rusqlite::Result { Ok((self.similarity_hash.cast_signed(), self.issue_id)) } } impl Similarity { /// Get all similarities by [crate::parse::Tag] in [super::Database] - pub fn query_all(db: &super::Database, _: ()) -> rusqlite::Result> { - let mut hm: HashMap = HashMap::new(); - db.prepare_cached( - " - SELECT DISTINCT - s.similarity_hash, - i.tag_id, - i.run_id, - s.issue_id - FROM similarities s - JOIN issues i ON i.id = s.issue_id - WHERE EXISTS ( - SELECT 1 FROM similarities - JOIN issues ON issues.id = similarities.issue_id - JOIN runs ON runs.id = issues.run_id - WHERE similarity_hash = s.similarity_hash - AND build_id IN ( - SELECT id FROM builds - GROUP BY job_id - HAVING MAX(number) - ) - ) - ", - )? - .query_map((), |row| { - Ok(( - row.get(0).map(i64::cast_unsigned)?, - TagInfo::select_one(db, row.get(1)?, ())?, - row.get(2)?, - row.get(3)?, - )) - })? - .collect::>>()? - .into_iter() - .try_for_each(|(hash, tag, run_id, issue_id)| { - hm.entry(hash) - .or_insert({ - Self { - tag, - related: HashSet::new(), - example: Issue::select_one( - db, - issue_id, - (db, &Run::select_one(db, run_id, ())?), - )? - .item() - .snippet, - } - }) - .related - .insert(run_id); - - Ok::<_, rusqlite::Error>(()) - })?; + pub async fn query_all(db: &super::Database) -> rusqlite::Result> { + struct InfoSet { + similarity: SimilarityInfo, + tag_id: i64, + run_id: i64, + } - let mut similarities: Vec<_> = hm - .into_values() - // ignore similarities within the same run - .filter(|s| s.related.len() > 1) - .collect(); + // TODO: stream this! + db.call(|conn| -> rusqlite::Result> { + let (tx, rx) = mpsc::unbounded_channel(); + conn.prepare_cached( + " + SELECT DISTINCT + s.similarity_hash, + s.issue_id, + i.tag_id, + i.run_id + FROM similarities s + JOIN issues i ON i.id = s.issue_id + WHERE EXISTS ( + SELECT 1 FROM similarities + JOIN issues ON issues.id = similarities.issue_id + JOIN runs ON runs.id = issues.run_id + WHERE similarity_hash = s.similarity_hash + AND build_id IN ( + SELECT id FROM builds + GROUP BY job_id + HAVING MAX(number) + ) + ) + ", + )? + .query_and_then((), |row| { + Ok(InfoSet { + similarity: SimilarityInfo { + similarity_hash: row.get(0).map(i64::cast_unsigned)?, + issue_id: row.get(1)?, + }, + tag_id: row.get(2)?, + run_id: row.get(3)?, + }) + })? + .try_for_each(|info| { + tx.send(info) + .map_err(|e| rusqlite::Error::UserFunctionError(e.into())) + })?; - similarities.sort_by_cached_key(|s| Reverse(s.related.len())); + Ok(rx.into()) + }) + .await? + .try_fold( + HashMap::new(), + |mut hm, + InfoSet { + similarity: + SimilarityInfo { + similarity_hash, + issue_id, + }, + tag_id, + run_id, + }| async move { + let tag = TagInfo::select_one(db, tag_id).await?; + let field = tag.field; + hm.entry(similarity_hash) + .or_insert({ + Self { + tag, + related: HashSet::new(), + example: IssueInfo::select_one(db, issue_id) + .and_then(|i| async { + match field { + Field::Console => Run::select_one(db, run_id) + .await? + .item() + .log + .map_or(Err(rusqlite::Error::InvalidQuery), |l| { + Ok(i.into_issue(&l)) + }), + Field::RunName => Ok(i.into_issue( + &Run::select_one_display_name(db, run_id).await?.into(), + )), + Field::Artifact => { + let artifact_id = i + .artifact_id + .ok_or(rusqlite::Error::InvalidQuery)?; + Ok(i.into_issue( + // TODO: share Vec contents somehow? + &from_utf8( + &Artifact::select_one(db, artifact_id) + .await? + .contents, + ) + .map(ArcStr::from)?, + )) + } + } + }) + .map_ok(|i| i.item().snippet) + .await?, + } + }) + .related + .insert(run_id); - Ok(similarities) + Ok::<_, rusqlite::Error>(hm) + }, + ) + .map_ok(|hm| hm.into_values()) + .await } } diff --git a/src/db/stats.rs b/src/db/stats.rs index 05ec3cc..b96652a 100644 --- a/src/db/stats.rs +++ b/src/db/stats.rs @@ -35,103 +35,113 @@ pub struct Statistics { impl Statistics { /// Gets [super::Database]'s [Statistics] - pub fn query(db: &super::Database) -> rusqlite::Result { + pub async fn query(db: &super::Database) -> rusqlite::Result { // calculate success/failures for latest runs let mut stats = db - .conn - .prepare( - " - SELECT status, id FROM runs - WHERE build_id IN ( - SELECT id FROM builds - GROUP BY job_id - HAVING MAX(number) - ) - ", - )? - .query_map((), |row| Ok((read_value!(row, 0), row.get(1)?)))? - .try_fold(Statistics::default(), |mut stats, res| { - let (status, id) = res?; - match status { - Some(BuildStatus::Aborted) => stats.aborted.push(id), - Some(BuildStatus::Failure) => stats.failures.push(id), - Some(BuildStatus::NotBuilt) => stats.not_built.push(id), - Some(BuildStatus::Success) => stats.successful.push(id), - Some(BuildStatus::Unstable) => stats.unstable.push(id), - _ => {} - }; - - Ok::<_, rusqlite::Error>(stats) - })?; + .call(|conn| { + conn.prepare( + " + SELECT status, id FROM runs + WHERE build_id IN ( + SELECT id FROM builds + GROUP BY job_id + HAVING MAX(number) + ) + ", + )? + .query_map((), |row| Ok((read_value!(row, 0), row.get(1)?)))? + .try_fold(Statistics::default(), |mut stats, res| { + let (status, id) = res?; + match status { + Some(BuildStatus::Aborted) => stats.aborted.push(id), + Some(BuildStatus::Failure) => stats.failures.push(id), + Some(BuildStatus::NotBuilt) => stats.not_built.push(id), + Some(BuildStatus::Success) => stats.successful.push(id), + Some(BuildStatus::Unstable) => stats.unstable.push(id), + _ => {} + }; + + Ok::<_, rusqlite::Error>(stats) + }) + }) + .await?; stats.successful_jobs = db - .conn - .prepare( - " - SELECT COUNT(*) FROM jobs - WHERE id IN ( - SELECT job_id FROM builds - GROUP BY job_id - HAVING MAX(number) AND status = ? - ) - ", - )? - .query_one((write_value!(BuildStatus::Success),), |row| row.get(0))?; + .call(|conn| { + conn.prepare( + " + SELECT COUNT(*) FROM jobs + WHERE id IN ( + SELECT job_id FROM builds + GROUP BY job_id + HAVING MAX(number) AND status = ? + ) + ", + )? + .query_one((write_value!(BuildStatus::Success),), |row| row.get(0)) + }) + .await?; stats.total_jobs = db - .conn - .prepare("SELECT COUNT(*) FROM jobs")? - .query_one((), |row| row.get(0))?; + .call(|conn| { + conn.prepare("SELECT COUNT(*) FROM jobs")? + .query_one((), |row| row.get(0)) + }) + .await?; // don't count metadata issues in total stats.issues_found = db - .conn - .prepare( - " - SELECT COUNT(*) FROM issues - JOIN tags ON tags.id = issues.tag_id - JOIN runs ON runs.id = issues.run_id - WHERE tags.severity != ? AND runs.build_id IN ( - SELECT id FROM builds - GROUP BY job_id - HAVING MAX(number) - ) - ", - )? - .query_one((write_value!(Severity::Metadata),), |row| row.get(0))?; + .call(|conn| { + conn.prepare( + " + SELECT COUNT(*) FROM issues + JOIN tags ON tags.id = issues.tag_id + JOIN runs ON runs.id = issues.run_id + WHERE tags.severity != ? AND runs.build_id IN ( + SELECT id FROM builds + GROUP BY job_id + HAVING MAX(number) + ) + ", + )? + .query_one((write_value!(Severity::Metadata),), |row| row.get(0)) + }) + .await?; stats.unknown_runs = db - .conn - .prepare( - " - SELECT r.id FROM runs r - WHERE ( - r.status = ? - OR r.status = ? - OR r.status = ? - ) AND r.build_id IN ( - SELECT id FROM builds - GROUP BY job_id - HAVING MAX(number) - ) AND NOT EXISTS ( - SELECT 1 FROM issues - JOIN tags ON tags.id = issues.tag_id - WHERE - issues.run_id = r.id - AND tags.severity != ? - ) - ", - )? - .query_map( - ( - write_value!(Some(BuildStatus::Failure)), - write_value!(Some(BuildStatus::Unstable)), - write_value!(Some(BuildStatus::Aborted)), - write_value!(Severity::Metadata), - ), - |row| row.get(0), - )? - .collect::>>()?; + .call(|conn| { + conn.prepare( + " + SELECT r.id FROM runs r + WHERE ( + r.status = ? + OR r.status = ? + OR r.status = ? + ) AND r.build_id IN ( + SELECT id FROM builds + GROUP BY job_id + HAVING MAX(number) + ) AND NOT EXISTS ( + SELECT 1 FROM issues + JOIN tags ON tags.id = issues.tag_id + WHERE + issues.run_id = r.id + AND tags.severity != ? + ) + ", + )? + .query_map( + ( + write_value!(Some(BuildStatus::Failure)), + write_value!(Some(BuildStatus::Unstable)), + write_value!(Some(BuildStatus::Aborted)), + write_value!(Severity::Metadata), + ), + |row| row.get(0), + )? + .collect::>>() + }) + .await?; Ok(stats) } diff --git a/src/db/tag.rs b/src/db/tag.rs index 2f93ac9..477e1c5 100644 --- a/src/db/tag.rs +++ b/src/db/tag.rs @@ -1,17 +1,18 @@ +use arcstr::ArcStr; + use crate::{ - config::{Field, Severity}, + config::{ConfigTag, Field, Severity}, db::{Queryable, Run, Upsertable}, - parse::{Tag, TagSet}, read_value, schema, write_value, }; #[derive(PartialEq, Eq, Hash)] pub struct TagInfo { /// Name of [Tag] - pub name: String, + pub name: ArcStr, /// Description of [Tag] - pub desc: String, + pub desc: ArcStr, /// Field of [Tag] pub field: Field, @@ -20,17 +21,6 @@ pub struct TagInfo { pub severity: Severity, } -impl From<&Tag> for TagInfo { - fn from(value: &Tag) -> Self { - TagInfo { - name: value.name.clone(), - desc: value.desc.clone(), - field: value.from, - severity: value.severity, - } - } -} - schema! { tags for TagInfo { id INTEGER PRIMARY KEY, @@ -41,25 +31,34 @@ schema! { } } -impl Queryable for TagInfo { - fn map_row(_: ()) -> impl FnMut(&rusqlite::Row) -> rusqlite::Result> { - |row| { - Ok(super::InDatabase::new( - row.get(0)?, - Self { - name: row.get(1)?, - desc: row.get(2)?, - field: read_value!(row, 3), - severity: read_value!(row, 4), - }, - )) +impl From for TagInfo { + fn from(value: ConfigTag) -> Self { + Self { + name: value.name.into(), + desc: value.desc.into(), + field: value.from, + severity: value.severity, } } +} - fn as_params(&self, _: ()) -> rusqlite::Result { +impl Queryable for TagInfo { + fn map_row(row: &rusqlite::Row) -> rusqlite::Result> { + Ok(super::InDatabase::new( + row.get(0)?, + Self { + name: row.get::<_, String>(1)?.into(), + desc: row.get::<_, String>(2)?.into(), + field: read_value!(row, 3), + severity: read_value!(row, 4), + }, + )) + } + + fn as_params(&self) -> rusqlite::Result { Ok(( - &self.name, - &self.desc, + self.name.as_str(), + self.desc.as_str(), write_value!(self.field), write_value!(self.severity), )) @@ -67,74 +66,72 @@ impl Queryable for TagInfo { } impl Upsertable for TagInfo { - fn upsert(self, db: &super::Database, params: ()) -> rusqlite::Result> { - db.prepare_cached( - " - INSERT INTO tags (name, desc, field, severity) VALUES (?, ?, ?, ?) - ON CONFLICT(name) DO UPDATE SET - desc = excluded.desc, - field = excluded.field, - severity = excluded.severity - ", - )? - .execute(self.as_params(params)?)?; + async fn upsert(self, db: &super::Database) -> rusqlite::Result> { + let name = self.name.clone(); + db.call(move |conn| { + conn.prepare_cached( + " + INSERT INTO tags (name, desc, field, severity) VALUES (?, ?, ?, ?) + ON CONFLICT(name) DO UPDATE SET + desc = excluded.desc, + field = excluded.field, + severity = excluded.severity + ", + )? + .execute(self.as_params()?) + }) + .await?; - Self::select_one_by_name(db, &self.name, ()) + Self::select_one_by_name(db, name).await } } impl TagInfo { /// Get a [TagInfo] from [super::Database] by name - pub fn select_one_by_name( + pub async fn select_one_by_name( db: &super::Database, - name: &str, - params: (), + name: ArcStr, ) -> rusqlite::Result> { - db.prepare_cached("SELECT * FROM tags WHERE name = ?")? - .query_one((name,), Self::map_row(params)) + db.call(move |conn| { + conn.prepare_cached("SELECT * FROM tags WHERE name = ?")? + .query_one((name.as_str(),), Self::map_row) + }) + .await } /// Get all [TagInfo]s from [Run] - pub fn select_all_by_run( + pub async fn select_all_by_run( db: &super::Database, run: &super::InDatabase, - params: (), ) -> rusqlite::Result>> { - db.prepare_cached( - " + let id = run.id; + db.call(move |conn| { + conn.prepare_cached( + " SELECT DISTINCT tags.id, name, desc, field, severity FROM tags JOIN issues ON tags.id = issues.tag_id WHERE issues.run_id = ? ", - )? - .query_map((run.id,), Self::map_row(params))? - .collect() - } - - /// Upsert a [TagSet] into [super::Database] - pub fn upsert_tag_set( - db: &super::Database, - tags: TagSet, - params: (), - ) -> rusqlite::Result>> { - tags.try_swap_tags(|t| { - Ok(super::InDatabase::new( - TagInfo::from(&t).upsert(db, params)?.id, - t, - )) + )? + .query_map((id,), Self::map_row)? + .collect() }) + .await } /// Remove all [Tag]s which aren't referenced by [super::Issue]s from [super::Database] - pub fn delete_all_orphan(db: &super::Database) -> rusqlite::Result { - db.execute( - " - DELETE FROM tags WHERE NOT EXISTS ( - SELECT 1 FROM issues - WHERE tags.id = issues.tag_id + pub async fn delete_all_orphan(db: &super::Database) -> rusqlite::Result { + db.call(|conn| { + conn.execute( + " + DELETE FROM tags WHERE NOT EXISTS ( + SELECT 1 FROM issues + WHERE tags.id = issues.tag_id + ) + ", + (), ) - ", - (), - ) + }) + .await } } diff --git a/src/main.rs b/src/main.rs index 89fe094..d2f7514 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,6 +1,5 @@ //! A Jenkins CI/CD-based build analyzer and issue prioritizer. use std::{ - cell::Cell, hash::{DefaultHasher, Hash, Hasher}, path::Path, process::Stdio, @@ -9,8 +8,10 @@ use std::{ }; use anyhow::{Error, Result}; +use arcstr::ArcStr; use clap::{Parser, crate_name, crate_version}; use env_logger::Env; +use futures::{Stream, StreamExt, TryFutureExt, TryStreamExt, future, stream}; use jenkins_api::{ Jenkins, JenkinsBuilder, build::{Build, BuildStatus, ShortBuild}, @@ -18,22 +19,16 @@ use jenkins_api::{ use log::{Level, info, log, warn}; use regex::Regex; use time::UtcOffset; -use tokio::{ - fs, - io::AsyncWriteExt, - process::Command, - sync::Semaphore, - task::{self, JoinSet}, -}; +use tokio::{fs, io::AsyncWriteExt, process::Command, sync::Semaphore, task::JoinSet}; use crate::{ - api::{AsBuild, AsJob, AsRun, SparseMatrixProject}, + api::{AsBuild, AsJob, AsRun, SparseBuild, SparseJob, SparseMatrixProject}, config::{Config, ConfigArtifact, Field, Severity}, db::{ - Artifact, Database, InDatabase, Issue, Job, JobBuild, Queryable, Run, SimilarityInfo, - TagInfo, Upsertable, + Artifact, Database, InDatabase, Issue, IssueInfo, Job, JobBuild, Queryable, Run, + SimilarityInfo, TagInfo, Upsertable, }, - parse::{Tag, TagSet, normalized_levenshtein_distance}, + parse::{Pattern, PatternSet, normalized_levenshtein_distance}, }; mod api; @@ -61,6 +56,61 @@ struct Args { purge_cache: bool, } +/// State to move to each task when pulling a [SparseMatrixProject] +struct SparseMatrixProjectContext<'a> { + project: SparseMatrixProject, + blocklist: &'a [String], +} + +/// State to move to each task when pulling a [Job] +struct JobContext { + sj: SparseJob, +} + +/// State to move to each task when pulling a [JobBuild] +struct BuildContext { + job: InDatabase, + sb: SparseBuild, +} + +/// State to move to each task when pulling a [Run] +struct RunContext { + job: InDatabase, + build: InDatabase, + mb: ShortBuild, +} + +/// State to move to each task when pulling an [Artifact] +struct ArtifactContext { + full_mb: Arc, + run: InDatabase, + artifact: jenkins_api::build::Artifact, +} + +/// Puller for Jenkins CI +#[derive(Clone)] +struct JenkinsPuller { + db: Database, + jenkins: Arc, + artifacts: Arc>, + last_n_history: usize, +} + +trait TryPull: Sized { + type Error; + + async fn pull(self, ctx: C) -> Result; +} + +trait TryPullNested { + type Error; + + async fn pull_stream( + self, + ctx: C, + ) -> Result>, Self::Error>; +} + // [reqwest] will open new connections until the system `ulimit`, // we have to limit parallelism ourselves static RATE_LIMIT: Semaphore = Semaphore::const_new(20); @@ -73,6 +123,205 @@ macro_rules! rate_limit { }; } +impl TryPullNested for JenkinsPuller { + type Error = rusqlite::Error; + + /// Pull a [Job] and cache into [Database], returning the last-$n$ [BuildContext]s + async fn pull_stream( + self, + JobContext { sj }: JobContext, + ) -> Result>, Self::Error> { + sj.as_job(self.last_n_history) + .upsert(&self.db) + .map_ok(|job| { + if sj.builds.is_empty() { + info!("Job '{}' has no builds.", &sj.name) + } + + stream::iter(sj.builds) + .take(self.last_n_history) + .map(move |sb| { + Ok(BuildContext { + job: job.clone(), + sb, + }) + }) + }) + .await + } +} + +impl TryPullNested for JenkinsPuller { + type Error = rusqlite::Error; + + /// Pull a [JobBuild] and cache into [Database], returning its [RunContext]s + async fn pull_stream( + self, + BuildContext { job, sb }: BuildContext, + ) -> Result>, Self::Error> { + sb.as_build(job.id) + .upsert(&self.db) + .map_ok(|build| { + stream::iter(sb.runs.into_iter().flatten()) + .filter(move |mb| future::ready(mb.number == sb.number)) + .map(move |mb| { + Ok(RunContext { + job: job.clone(), + build: build.clone(), + mb, + }) + }) + }) + .await + } +} + +impl TryPull> for JenkinsPuller { + type Error = anyhow::Error; + + /// Pull a [Run] and cache into [Database] + async fn pull( + self, + RunContext { job, build, mb }: RunContext, + ) -> Result, Self::Error> { + match Run::select_one_by_url(&self.db, ArcStr::from(&mb.url)).await { + Ok(run) => Ok(run), + Err(rusqlite::Error::QueryReturnedNoRows) => { + mb.get_full_build(&self.jenkins.clone()) + .ok_into() + .and_then(|full_mb: Arc<_>| async move { + let run = full_mb + .as_run(build.id, &self.jenkins) + .await + .upsert(&self.db) + .await?; + + // pull artifacts + // FIXME: asyncify parsing + stream::iter(full_mb.artifacts.clone()) + .map(|artifact| ArtifactContext { + full_mb: full_mb.clone(), + run: run.clone(), + artifact, + }) + .then(|ctx| self.clone().pull(ctx)) + .inspect_err(|e| { + log::error!( + "Failed to retrieve artifact for run {}: {}", + run.display_name, + e + ); + }) + .try_collect::>() + .await?; + + log!( + match run.status { + Some( + BuildStatus::Failure + | BuildStatus::Unstable + | BuildStatus::Aborted, + ) => Level::Warn, + _ => Level::Info, + }, + "Job '{}#{}' run '{}' finished with status {:?}.", + job.name, + build.number, + run.display_name, + run.status + ); + + Ok(run) + }) + .map_err(Error::from_boxed) + .await + } + Err(e) => Err(Error::from(e)), + } + } +} + +impl TryPull, Option>> for JenkinsPuller { + type Error = anyhow::Error; + + async fn pull( + self, + ArtifactContext { + full_mb, + run, + artifact, + }: ArtifactContext, + ) -> Result>, Self::Error> { + match self.artifacts.grep_matches(&artifact.relative_path).next() { + Some(config) => { + let run_id = run.id; + full_mb + .get_artifact(&self.jenkins, &artifact) + .map_err(Error::from_boxed) + .and_then(|blob| async move { + match config + .post_process + .as_ref() + .map(|cmd| cmd.split_first()) + .flatten() + { + Some((program, argv)) => { + spawn_process(program, argv, &run.display_name, &run.url, &blob) + .map_err(Error::from) + .await + } + None => Ok(blob.to_vec()), + } + }) + .and_then(|contents| { + Artifact { + path: artifact.relative_path.clone(), // FIXME: make this an arc? + contents, + run_id, + } + .insert(&self.db) + .map_err(Error::from) + }) + .ok_into() + .await + } + None => Ok(None), + } + } +} + +impl TryPullNested, InDatabase> for JenkinsPuller { + type Error = Error; + + /// Pull jobs plus their underlying builds, runs, and artifacts from [SparseMatrixProject] and cache them into [Database] + async fn pull_stream( + self, + SparseMatrixProjectContext { project, blocklist }: SparseMatrixProjectContext<'_>, + ) -> Result, Self::Error>>, Self::Error> { + Ok(stream::iter( + project + .jobs + .into_iter() + .filter(|sj| !blocklist.contains(&sj.name)), + ) + .then({ + let puller = self.clone(); + move |sj| puller.clone().pull_stream(JobContext { sj }) + }) + .try_flatten_unordered(None) + .and_then({ + let puller = self.clone(); + move |ctx| puller.clone().pull_stream(ctx) + }) + .try_flatten_unordered(None) + .map_err(Error::from) + .and_then({ + let puller = self.clone(); + move |ctx: RunContext| puller.clone().pull(ctx) + })) + } +} + /// Spawns a process, pipes stdin, and waits for stdout #[inline] async fn spawn_process( @@ -105,207 +354,29 @@ where Ok(child.wait_with_output().await?.stdout) } -/// Pull builds from `project.jobs` and cache them into database `db` -async fn pull_build_logs( - project: SparseMatrixProject, - artifacts: Arc<[(Regex, ConfigArtifact)]>, - blocklist: &[String], - last_n_history: usize, - jenkins: Arc, - db: &Database, -) -> Result>> { - // Context struct to move around to each task - struct Context { - artifacts: Arc<[(Regex, ConfigArtifact)]>, - jenkins: Arc, - job: Arc>, - build: Arc>, - mb: ShortBuild, - } - - // https://morestina.net/1607/fallible-iteration - let err: Cell> = Cell::new(Ok(())); - fn until_err(err: &mut &Cell>, res: Result) -> Option { - res.map_err(|e| err.set(Err(e))).ok() - } - - // spawn tasks to pull builds - let mut runs = Vec::new(); - let mut handles: JoinSet<_> = project - .jobs - .into_iter() - .filter(|sj| { - !blocklist.contains(&sj.name) - && sj - .builds - .is_empty() - .then(|| info!("Job '{}' has no builds.", &sj.name)) - .is_none() - }) - .map(|sj| { - let job: Arc<_> = sj.as_job(last_n_history).upsert(db, ())?.into(); - Ok(sj - .builds - .into_iter() - .map(move |sb| (job.clone(), sb)) - .take(last_n_history)) - }) - .scan(&err, until_err) - .flatten() - .map(|(job, sb)| { - let artifacts = artifacts.clone(); - let jenkins = jenkins.clone(); - let build: Arc<_> = sb.as_build(job.id).upsert(db, ())?.into(); - Ok(sb - .runs - .into_iter() - .flatten() - .filter(move |mb| mb.number == sb.number) - .map(move |mb| Context { - artifacts: artifacts.clone(), - jenkins: jenkins.clone(), - job: job.clone(), - build: build.clone(), - mb, - })) - }) - .scan(&err, until_err) - .flatten() - .filter_map(|ctx| match Run::select_one_by_url(db, &ctx.mb.url, ()) { - Ok(run) => { - runs.push(run); - None - } // cached - Err(rusqlite::Error::QueryReturnedNoRows) => Some(Ok(ctx)), - Err(e) => Some(Err(Error::from(e))), - }) - .scan(&err, until_err) - .map( - |Context { - artifacts, - jenkins, - job, - build, - mb, - }| { - rate_limit!(async move { - let full_build: Arc<_> = mb.get_full_build(&jenkins).await.unwrap().into(); - let run = full_build.as_run(build.id, &jenkins).await; - - let artifacts = artifacts.clone(); - let display_name = run.display_name.clone(); - let url = run.url.clone(); - let artifacts: JoinSet<_> = full_build - .clone() - .artifacts - .iter() - .filter_map(move |artifact| { - let jenkins = jenkins.clone(); - let full_build = full_build.clone(); - let artifact = artifact.clone(); - let display_name = display_name.clone(); - let url = url.clone(); - artifacts - .iter() - .find(|(re, _)| re.is_match(&artifact.relative_path)) - .map(move |(_, c)| { - let post_process = c.post_process.clone(); - rate_limit!(async move { - let blob = full_build - .get_artifact(&jenkins, &artifact) - .await - .inspect_err(|e| { - log::error!( - "Failed to retrieve artifact for run {}: {}", - &display_name, - e - ) - }) - .ok()?; - - let contents = if let Some(mut iter) = - post_process.as_ref().map(|argv| argv.iter()) - && let Some(program) = iter.next() - { - spawn_process(program, iter, &display_name, &url, &blob) - .await - .unwrap() - } else { - blob.to_vec() - }; - - Some(|run_id| Artifact { - path: artifact.relative_path, - contents, - run_id, - }) - }) - }) - }) - .collect(); - - log!( - match run.status { - Some( - BuildStatus::Failure | BuildStatus::Unstable | BuildStatus::Aborted, - ) => Level::Warn, - _ => Level::Info, - }, - "Job '{}#{}' run '{}' finished with status {:?}.", - job.name, - build.number, - run.display_name, - run.status - ); - - (run, artifacts) - }) - }, - ) - .collect(); - - err.into_inner()?; // check for failures before continuing - runs.reserve(handles.len()); - - // collect them all here - while let Some(h) = handles.join_next().await { - let (run, mut artifacts) = h?; - let run = run.upsert(db, ())?; - - while let Some(artifact) = artifacts.join_next().await { - if let Ok(Some(artifact)) = artifact { - artifact(run.id).insert(db, ())?; - } - } - - runs.push(run); - } - - Ok(runs) -} - /// Parse all untagged runs for `tags` and cache them into database `db` async fn parse_unprocessed_runs( runs: Vec>, - tags: Arc>>, + tags: Arc>>, db: &Database, ) -> Result>> { let mut inserted_issues = Vec::new(); + let mut handles = JoinSet::new(); enum Dependent { - Run(Issue), - Artifact(Issue, Arc>), + Run(IssueInfo), + Artifact(IssueInfo, Arc>), } - let mut handles: JoinSet<_> = runs - .into_iter() - .filter_map(|run| match run.tag_schema { + // TODO: Stream this! + for run in runs { + match run.tag_schema { None => { let tags = tags.clone(); - let artifacts = Artifact::select_all_by_run(db, run.id, ()); - Some(async move { + let artifacts = Artifact::select_all_by_run(db, run.id).await?; + handles.spawn(async move { let issues: Vec<_> = { - let warn = |t: &InDatabase| match t.severity { + let warn = |t: &InDatabase| match t.severity { Severity::Metadata => {} _ => warn!( "Found issue(s) tagged '{}' in run '{}'", @@ -313,34 +384,41 @@ async fn parse_unprocessed_runs( ), }; let run_name = tags - .grep_tags(run.display_name.clone(), Field::RunName) - .flat_map(|t| t.grep_issue(run.display_name.clone())) - .map(Dependent::Run); + .grep_matches(run.display_name.clone()) + .flat_map(|t| t.grep_issue(&run.display_name)) + .map(|i| Dependent::Run(i.into_issue_info(&run, None))); let console = run .log .iter() .flat_map(|l| { - tags.grep_tags(l.clone(), Field::Console).flat_map(|t| { + tags.grep_matches(l.clone()).flat_map(|t| { warn(t); - t.grep_issue(l.clone()) + t.grep_issue(l) }) }) - .map(Dependent::Run); + .map(|i| Dependent::Run(i.into_issue_info(&run, None))); let artifact = artifacts .into_iter() - .flatten() - .map(|a| a.into()) - .filter_map(|a: Arc<_>| { + .map(Arc::from) + .filter_map(|a| { + let run = run.clone(); from_utf8(&a.contents) .ok() .map(|b| -> arcstr::ArcStr { b.into() }) .map(|blob| { - tags.grep_tags(blob.clone(), Field::Artifact) + tags.grep_matches(blob.clone()) .flat_map(move |t| { warn(t); - t.grep_issue(blob.clone()) + t.grep_issue(&blob.clone()) + .collect::>() // FIXME hack + .into_iter() + }) + .map(move |i: Issue| { + Dependent::Artifact( + i.into_issue_info(&run, Some(&a)), + a.clone(), + ) }) - .map(move |i| Dependent::Artifact(i, a.clone())) }) }) .flatten(); @@ -349,38 +427,43 @@ async fn parse_unprocessed_runs( }; (run, issues) - }) + }); } _ => { - inserted_issues.extend( - Issue::select_all_not_metadata(db, (db, &run)) - .into_iter() - .flatten(), - ); - None + inserted_issues.extend(run.select_all_issues(db, false).await?); } - }) - .collect(); + } + } while let Some(h) = handles.join_next().await { let (run, issues) = h?; - inserted_issues = issues.into_iter().try_fold(inserted_issues, |mut acc, i| { - let issue = match i { - Dependent::Run(issue) => issue.insert(db, (&run, None))?, - Dependent::Artifact(issue, artifact) => { - issue.insert(db, (&run, Some(&artifact)))? - } + + // TODO: deduplicate this... + for i in issues { + let (info, artifact) = match i { + Dependent::Run(issue) => (issue, None), + Dependent::Artifact(issue, artifact) => (issue, Some(artifact)), }; - match TagInfo::select_one(db, issue.tag_id, ())?.severity { - Severity::Metadata => {} - _ => acc.push(issue), + let tag = TagInfo::select_one(db, info.tag_id).await?; + + // FIXME: This is a complete hack + let issue = match tag.field { + Field::Console => info.insert(db).await?.into_issue(run.log.as_ref().unwrap()), + Field::RunName => info.insert(db).await?.into_issue(&run.display_name), + Field::Artifact => info + .insert(db) + .await? + .into_issue(&from_utf8(&artifact.unwrap().contents).unwrap().into()), + }; + + if matches!(tag.severity, Severity::Metadata) { + inserted_issues.push(issue); } - Ok::<_, Error>(acc) - })?; + } } // batch update tag schema for runs afterwards - Run::update_all_tag_schema(db, Some(tags.schema()))?; + Run::update_all_tag_schema(db, Some(tags.schema())).await?; Ok(inserted_issues) } @@ -456,20 +539,19 @@ async fn calculate_similarities( let (hash, g) = h?; // unique issues are discarded if g.len() > 1 { - g.iter().try_for_each(|i| { + for i in g { SimilarityInfo { similarity_hash: hash, issue_id: i.id, } - .insert(db, ())?; + .insert(db) + .await?; info!( "Issue '#{}' likely matches with similarity group '#{}'!", i.id, hash ); - - Ok::<_, Error>(()) - })?; + } } } @@ -479,23 +561,26 @@ async fn calculate_similarities( /// Copies the rendered versions of every [Artifact] into `folder` async fn copy_artifacts>( folder: P, - artifacts: Arc<[(Regex, ConfigArtifact)]>, + artifacts: Arc>, db: &Database, ) -> Result<()> { // create dir first fs::create_dir_all(&folder).await?; // render and copy - let mut handles: JoinSet<_> = Artifact::select_all(db, ())? + let mut handles: JoinSet<_> = Artifact::select_all(db) + .await? .into_iter() .map(|artifact| { let artifacts = artifacts.clone(); - let display_name = Run::select_one_display_name(db, artifact.run_id)?; - let url = Run::select_one_url(db, artifact.run_id)?; + let db = db.clone(); let path = folder.as_ref().join(artifact.id.to_string()); Ok(async move { - let blob = if let Some((_, c)) = - artifacts.iter().find(|(re, _)| re.is_match(&artifact.path)) + let display_name = Run::select_one_display_name(&db, artifact.run_id) + .await + .unwrap(); + let url = Run::select_one_url(&db, artifact.run_id).await.unwrap(); + let blob = if let Some(c) = artifacts.grep_matches(&artifact.path).next() && let Some(mut iter) = c.render.as_ref().map(|argv| argv.iter()) && let Some(program) = iter.next() { @@ -527,7 +612,7 @@ async fn main() -> Result<()> { info!("{} {}", crate_name!(), crate_version!()); // load config - info!("Compiling issue patterns..."); + info!("Loading config..."); let Config { artifact, blocklist, @@ -542,35 +627,55 @@ async fn main() -> Result<()> { username, view, } = toml::from_str(&fs::read_to_string(args.config).await?)?; - let tags = TagSet::from_config(tag)?; - let artifact: Arc<[_]> = artifact - .into_iter() - .map(|a| Regex::new(&a.path).map(|re| (re, a))) - .collect::, _>>()? - .into(); + let artifacts: Arc<_> = PatternSet::new( + artifact + .into_iter() + .map(|a| { + Regex::new(&a.path) + .map_err(Error::from) + .map(|regex| Pattern::new(a, regex)) + }) + .collect::>()?, + )? + .into(); // open db info!("Opening database..."); - let mut database = Database::open(&database)?; + let database = Database::open(&database).await?; // check for cache purge if args.purge_cache { warn!("Purging cache!"); - database.purge_cache()?; + database.purge_cache().await?; } - // update TagSet - info!("Updating tags..."); - let tags = TagInfo::upsert_tag_set(&database, tags, ())?; + // update PatternSet + info!("Compiling issue patterns..."); + let tags = PatternSet::new( + stream::iter(tag) + .then(|t| async { + let regex = Regex::new(&t.pattern)?; + Ok::<_, Error>(Pattern::new( + TagInfo::from(t) + .upsert(&database) + .map_err(Error::from) + .await?, + regex, + )) + }) + .try_collect() + .await?, + )?; // purge outdated issues - let outdated = Issue::delete_all_invalid_by_tag_schema(&mut database, tags.schema())?; + let outdated = IssueInfo::delete_all_invalid_by_tag_schema(&database, tags.schema()).await?; if outdated > 0 { warn!("Purged {outdated} runs' issues that parsed with an outdated tag schema!"); } // purge blocklisted jobs - let blocked = Job::delete_all_by_blocklist(&mut database, &blocklist)?; + // TODO don't clone this + let blocked = Job::delete_all_by_blocklist(&database, blocklist.clone()).await?; if blocked > 0 { warn!("Purged {blocked} jobs that are on the blocklist."); } @@ -592,20 +697,24 @@ async fn main() -> Result<()> { info!("----------------------------------------"); let project = SparseMatrixProject::pull_jobs(&jenkins, &project).await?; - let runs = pull_build_logs( - project, - artifact.clone(), - &blocklist, + let runs = JenkinsPuller { + db: database.clone(), + jenkins: jenkins.into(), last_n_history, - jenkins.into(), - &database, - ) + artifacts: artifacts.clone(), + } + .pull_stream(SparseMatrixProjectContext { + project, + blocklist: &blocklist, + }) + .await? + .try_collect() .await?; info!("Done!"); info!("----------------------------------------"); - if Run::has_untagged(&database)? { + if Run::has_untagged(&database).await? { info!("Parsing unprocessed run logs..."); let issues = parse_unprocessed_runs(runs, tags.into(), &database).await?; @@ -615,10 +724,10 @@ async fn main() -> Result<()> { // purge old data info!("Purging old runs..."); - JobBuild::delete_all_orphan(&database)?; + JobBuild::delete_all_orphan(&database).await?; info!("Purging extraneous tags..."); - TagInfo::delete_all_orphan(&database)?; + TagInfo::delete_all_orphan(&database).await?; info!("Calculating issue similarities..."); calculate_similarities(issues, threshold, &database).await?; @@ -632,25 +741,24 @@ async fn main() -> Result<()> { if let Some(output) = args.output { info!("Generating report..."); - copy_artifacts("artifacts", artifact, &database).await?; + copy_artifacts("artifacts", artifacts, &database).await?; - let markup = task::spawn(async move { - page::render( - &database, - &view, - UtcOffset::from_hms(timezone, 0, 0).unwrap(), - ) - .unwrap() - .into_string() - }); + let markup = page::render( + &database, + &view, + UtcOffset::from_hms(timezone, 0, 0).unwrap(), + ) + .await + .unwrap() + .into_string(); if let Some(filepath) = output { - fs::write(&filepath, markup.await?).await?; + fs::write(&filepath, markup).await?; info!("Written to {filepath}"); } else { info!("Dumping to stdout --"); - println!("{}", markup.await?); + println!("{}", markup); } } diff --git a/src/page.rs b/src/page.rs index 9b23071..7b20e95 100644 --- a/src/page.rs +++ b/src/page.rs @@ -1,5 +1,5 @@ //! HTML report generation using [maud] templating. -use std::{collections::HashMap, str::from_utf8_unchecked, time::SystemTime}; +use std::{cmp::Reverse, collections::HashMap, str::from_utf8_unchecked, time::SystemTime}; use anyhow::{Error, Result}; use jenkins_api::build::BuildStatus; @@ -9,8 +9,8 @@ use time::{OffsetDateTime, UtcOffset, macros::format_description}; use crate::{ config::{Severity, TagView}, db::{ - Artifact, BlobFormat, Database, InDatabase, Issue, Job, JobBuild, Queryable, Run, - Similarity, Statistics, TagInfo, + Artifact, BlobFormat, Database, InDatabase, Job, JobBuild, Queryable, Run, Similarity, + Statistics, TagInfo, }, tag_expr::TagExpr, }; @@ -64,17 +64,17 @@ fn severity_as_class(severity: Severity) -> Option<&'static str> { } /// Render a [crate::api::SparseJob] -fn render_job(job: &InDatabase, db: &Database, tz: UtcOffset) -> Result { +async fn render_job(job: &InDatabase, db: &Database, tz: UtcOffset) -> Result { Ok(html! { h2 { a href=(job.url) { (job.name) } } - @if let Some((last_build, rest)) = JobBuild::select_all_by_job(db, job.id, ())?.split_first() { - (render_build(&last_build, db, tz, true)?) + @if let Some((last_build, rest)) = JobBuild::select_all_by_job(db, job.id).await?.split_first() { + (render_build(&last_build, db, tz, true).await?) @for build in rest { - (render_build(&build, db, tz, false)?) + (render_build(&build, db, tz, false).await?) } } @else { p { @@ -85,13 +85,13 @@ fn render_job(job: &InDatabase, db: &Database, tz: UtcOffset) -> Result, db: &Database, tz: UtcOffset, latest: bool, ) -> Result { - let mut runs = Run::select_all_by_build(db, &build, ())?; + let mut runs = Run::select_all_by_build(db, &build).await?; runs.sort_by_cached_key(|r| match r.status { Some(BuildStatus::Failure) => 0, Some(BuildStatus::Unstable) => 1, @@ -127,7 +127,7 @@ fn render_build( } } @for run in runs { - (render_run(&run, db)?) + (render_run(&run, db).await?) br; } } @@ -135,8 +135,8 @@ fn render_build( } /// Render a [Run] -fn render_run(run: &InDatabase, db: &Database) -> Result { - let issues = Issue::select_all_not_metadata(db, (db, run))?; +async fn render_run(run: &InDatabase, db: &Database) -> Result { + let issues = run.select_all_issues(db, false).await?; Ok(html! { table { tr #(run.id) class=[status_as_class(run.status)] { @@ -158,7 +158,7 @@ fn render_run(run: &InDatabase, db: &Database) -> Result { } tr #(run.id) class=[status_as_class(run.status)] { td { // tags - @let tags = TagInfo::select_all_by_run(db, run, ())?; + @let tags = TagInfo::select_all_by_run(db, run).await?; @if !tags.is_empty() { @for t in tags { code title=(t.desc) { @@ -212,7 +212,7 @@ fn render_run(run: &InDatabase, db: &Database) -> Result { } } } - @let artifacts = Artifact::select_all_by_run(db, run.id, ())?; + @let artifacts = Artifact::select_all_by_run(db, run.id).await?; @for a in artifacts { tr class=[status_as_class(run.status)] { td colspan="3" { // artifacts @@ -240,8 +240,8 @@ fn render_run(run: &InDatabase, db: &Database) -> Result { } /// Render [crate::db::Statistics] -fn render_stats(db: &Database) -> Result { - let stats = Statistics::query(db)?; +async fn render_stats(db: &Database) -> Result { + let stats = Statistics::query(db).await?; Ok(html! { h3 { "Job Statistics" @@ -265,7 +265,7 @@ fn render_stats(db: &Database) -> Result { "Failures" } td { - (render_run_ids(stats.failures.iter(), db)?) + (render_run_ids(stats.failures.iter(), db).await?) } } tr { @@ -273,7 +273,7 @@ fn render_stats(db: &Database) -> Result { "Unstable" } td { - (render_run_ids(stats.unstable.iter(), db)?) + (render_run_ids(stats.unstable.iter(), db).await?) } } tr { @@ -281,7 +281,7 @@ fn render_stats(db: &Database) -> Result { "Healthy" } td { - (render_run_ids(stats.successful.iter(), db)?) + (render_run_ids(stats.successful.iter(), db).await?) } } tr { @@ -289,7 +289,7 @@ fn render_stats(db: &Database) -> Result { "Aborted" } td { - (render_run_ids(stats.aborted.iter(), db)?) + (render_run_ids(stats.aborted.iter(), db).await?) } } tr { @@ -297,7 +297,7 @@ fn render_stats(db: &Database) -> Result { "Not Built" } td { - (render_run_ids(stats.not_built.iter(), db)?) + (render_run_ids(stats.not_built.iter(), db).await?) } } tr { @@ -341,7 +341,7 @@ fn render_stats(db: &Database) -> Result { } td { b { - (render_run_ids(stats.unknown_runs.iter(), db)?) + (render_run_ids(stats.unknown_runs.iter(), db).await?) } } } @@ -350,15 +350,19 @@ fn render_stats(db: &Database) -> Result { } /// Render [crate::db::Similarity] -fn render_similarities(db: &Database) -> Result { - let similarities: HashMap<_, Vec<_>> = - Similarity::query_all(db, ())? - .into_iter() - .fold(HashMap::new(), |mut acc, s| { - acc.entry(s.tag.severity).or_default().push(s); +async fn render_similarities(db: &Database) -> Result { + let mut similarities: HashMap<_, Vec<_>> = Similarity::query_all(db) + .await? + .into_iter() + .filter(|s| s.related.len() > 1) // ignore similarities within the same run + .fold(HashMap::new(), |mut acc, s| { + acc.entry(s.tag.severity).or_default().push(s); - acc - }); + acc + }); + similarities + .values_mut() + .for_each(|s| s.sort_by_cached_key(|s| Reverse(s.related.len()))); Ok(html! { h4 { @@ -385,7 +389,7 @@ fn render_similarities(db: &Database) -> Result { } } td { - (render_run_ids(s.related.iter(), db)?) + (render_run_ids(s.related.iter(), db).await?) } } tr class=[severity_as_class(s.tag.severity)] { @@ -409,14 +413,14 @@ fn render_similarities(db: &Database) -> Result { } /// Render a [TagView] -fn render_view(view: &TagView, db: &Database) -> Result { +async fn render_view(view: &TagView, db: &Database) -> Result { let expr = match TagExpr::parse(&view.expr) { Ok(expr) => Ok(expr), Err(e) => Err(Error::msg( e.iter().fold(String::new(), |acc, e| format!("{acc}\n{e}")), )), }?; - let rows = expr.eval_rows(&TagInfo::select_all(db, ())?); + let rows = expr.eval_rows(&TagInfo::select_all(db).await?); Ok(html! { h4 { @@ -424,7 +428,7 @@ fn render_view(view: &TagView, db: &Database) -> Result { } table class="view" { @for expr in rows { - @let matches = Run::select_all_id_by_expr(db, &expr)?; + @let matches = Run::select_all_id_by_expr(db, &expr).await?; @if !matches.is_empty() { tr { td { @@ -433,7 +437,7 @@ fn render_view(view: &TagView, db: &Database) -> Result { } } td { - (render_run_ids(matches.iter(), db)?) + (render_run_ids(matches.iter(), db).await?) } } } @@ -443,7 +447,7 @@ fn render_view(view: &TagView, db: &Database) -> Result { } /// Render a list of [Run] ids as their display name -fn render_run_ids<'a, T>(ids: T, db: &Database) -> Result +async fn render_run_ids<'a, T>(ids: T, db: &Database) -> Result where T: ExactSizeIterator + Iterator, { @@ -459,7 +463,7 @@ where @for id in ids { li { a href={"#" (id)} { - (Run::select_one_display_name(db, *id)?) + (Run::select_one_display_name(db, *id).await?) } } } @@ -472,7 +476,7 @@ where } /// Render an HTML report for [Database] info -pub fn render(db: &Database, views: &[TagView], tz: UtcOffset) -> Result { +pub async fn render(db: &Database, views: &[TagView], tz: UtcOffset) -> Result { Ok(html! { (DOCTYPE) html lang="en" { @@ -487,13 +491,13 @@ pub fn render(db: &Database, views: &[TagView], tz: UtcOffset) -> Result h1 { "build-pulse" } - (render_stats(db)?) - (render_similarities(db)?) + (render_stats(db).await?) + (render_similarities(db).await?) @for view in views { - (render_view(view, db)?) + (render_view(view, db).await?) } - @for job in Job::select_all(db, ())? { - (render_job(&job, db, tz)?) + @for job in Job::select_all(db).await? { + (render_job(&job, db, tz).await?) } p { "Report generated on " diff --git a/src/parse.rs b/src/parse.rs index 9df6036..601bc64 100644 --- a/src/parse.rs +++ b/src/parse.rs @@ -1,151 +1,112 @@ -//! [Tag] and [TagSet] parsing behavior. +//! [Pattern] and [PatternSet] parsing behavior. use std::{ collections::HashMap, hash::{DefaultHasher, Hash, Hasher}, ops::Deref, }; -use arcstr::ArcStr; +use arcstr::{ArcStr, Substr}; use regex::{Regex, RegexSet}; -use crate::{ - config::{ConfigTag, Field, Severity}, - db::{InDatabase, Issue}, -}; +use crate::db::{InDatabase, Issue, TagInfo}; -/// Set of [Tag]s -pub struct TagSet { - /// [Vec] of underlying [Tag]s - tags: Vec, +/// Set of [Pattern]s +pub struct PatternSet { + /// [Vec] of underlying [Pattern]s + patterns: Vec>, - /// [RegexSet] matching [Tag]s - match_set: RegexSet, + /// [RegexSet] matching [Pattern]s + pattern_set: RegexSet, } -/// [Tag] that can be parsed for [Issue]s -pub struct Tag { - /// Unique name - pub name: String, - - /// Description of [Tag] - pub desc: String, - - /// [Regex] pattern to match for +/// [Pattern] that can be searched for +pub struct Pattern { + /// [Regex] pattern to match regex: Regex, - /// [Field] to pattern match - pub from: Field, - - /// [Severity] of [Tag] - pub severity: Severity, + /// Found item + item: T, } -impl Hash for TagSet -where - T: Deref, -{ +impl Hash for PatternSet { fn hash(&self, state: &mut H) { - self.tags.iter().for_each(|t| t.hash(state)); + self.patterns.hash(state); } } -impl Hash for Tag { +impl Hash for Pattern { fn hash(&self, state: &mut H) { - self.name.hash(state); + self.item.hash(state); self.regex.as_str().hash(state); - self.from.hash(state); } } -impl Deref for TagSet -where - T: Deref, -{ - type Target = Vec; +impl Deref for Pattern { + type Target = T; + fn deref(&self) -> &Self::Target { - &self.tags + &self.item } } -impl TagSet { - /// Load an array of [ConfigTag] into a [TagSet] - pub fn from_config(config_tags: Vec) -> Result { - let match_set = RegexSet::new(config_tags.iter().map(|i| &i.pattern))?; - let tags = config_tags - .into_iter() - .map(|i| { - Ok(Tag { - name: i.name, - desc: i.desc, - regex: Regex::new(&i.pattern)?, - from: i.from, - severity: i.severity, - }) - }) - .collect::, _>>()?; - - Ok(Self { tags, match_set }) +impl PatternSet { + /// Create a new [PatternSet] from a set of [Pattern]s + pub fn new(patterns: Vec>) -> Result { + let pattern_set = RegexSet::new(patterns.iter().map(|m| m.regex.as_str()))?; + Ok(Self { + patterns, + pattern_set, + }) } -} -impl TagSet -where - T: Deref, -{ - /// Grep `field` for [Tag]s - pub fn grep_tags(&self, field: ArcStr, from: Field) -> impl Iterator { + /// Grep `field` for [Pattern]s + pub fn grep_matches>(&self, field: S) -> impl Iterator> { // matches using the match set first, then the regex of all valid matches are ran again to find them - self.match_set - .matches(&field) + self.pattern_set + .matches(field.as_ref()) .into_iter() - .map(|i| &self.tags[i]) - .filter(move |t| t.from == from) + .map(|i| &self.patterns[i]) } - /// Get the [TagSet] schema/hash - pub fn schema(&self) -> u64 { + /// Get the [PatternSet] schema/hash + pub fn schema(&self) -> u64 + where + Self: Hash, + { let mut s = DefaultHasher::new(); self.hash(&mut s); s.finish() } } -impl TagSet { - /// Try to mutate tags in-place from `T` -> `U` - pub fn try_swap_tags(self, f: F) -> Result, E> - where - F: FnMut(T) -> Result, - { - Ok(TagSet { - tags: self - .tags - .into_iter() - .map(f) - .collect::, _>>()?, - match_set: self.match_set, - }) +impl Pattern { + /// Create a new pattern + pub fn new(item: T, regex: Regex) -> Self { + Self { item, regex } + } + + /// Grep `field` for substrings + pub fn grep_substr(&self, field: &ArcStr) -> impl Iterator { + self.regex + .find_iter(field) + .map(move |m| field.substr(m.range())) } } -impl InDatabase { +impl Pattern> { /// Grep `field` for [Issue]s - pub fn grep_issue(&self, field: ArcStr) -> impl Iterator { - let mut hm: HashMap = HashMap::new(); - self.regex - .find_iter(&field) - .map(|m| Issue { - snippet: field.substr_from(m.into()), + pub fn grep_issue(&self, field: &ArcStr) -> impl Iterator { + self.grep_substr(&field) + .fold(HashMap::new(), |mut acc, m| { + *acc.entry(m).or_default() += 1; + acc + }) + .into_iter() + .map(move |(snippet, duplicates)| Issue { + snippet, tag_id: self.id, - duplicates: 0, + duplicates, }) - .for_each(|i| { - hm.entry(i).and_modify(|e| *e += 1).or_insert(0); - }); - - hm.into_iter().map(|(mut i, d)| { - i.duplicates = d; - i - }) } } diff --git a/src/tag_expr.rs b/src/tag_expr.rs index cc2ba90..a22baab 100644 --- a/src/tag_expr.rs +++ b/src/tag_expr.rs @@ -2,7 +2,7 @@ use std::{fmt, ops::Deref}; use chumsky::{pratt::*, prelude::*}; use regex::Regex; -use rusqlite::{Error, Params, ToSql, params_from_iter}; +use rusqlite::{Error, ToSql}; use serde_json::to_value; use crate::{config::Severity, db::TagInfo}; @@ -131,8 +131,9 @@ impl TagExpr { } } - pub fn to_sql_select(&self) -> Result<(String, impl Params), Error> { - fn to_where_expr(expr: &TagExpr) -> Result<(String, Vec>), Error> { + pub fn to_sql_select(&self) -> Result<(String, Vec>), Error> { + // FIXME: avoid all these allocations when making the select! + fn to_where_expr(expr: &TagExpr) -> Result<(String, Vec>), Error> { match expr { TagExpr::Not(e) => { let (expr, params) = to_where_expr(e)?; @@ -192,7 +193,7 @@ impl TagExpr { ) " ), - params_from_iter(params), + params, )) } }