queue: move docbuilder::queue::BuildQueue to queue::BuildQueue

Author: pietroalbini

src/docbuilder/mod.rs

@@ -7,13 +7,13 @@ mod rustwide_builder;
 
 pub(crate) use self::limits::Limits;
 pub(self) use self::metadata::Metadata;
-pub use self::queue::BuildQueue;
 pub(crate) use self::rustwide_builder::BuildResult;
 pub use self::rustwide_builder::RustwideBuilder;
 
 use crate::db::Pool;
 use crate::error::Result;
 use crate::index::Index;
+use crate::queue::BuildQueue;
 use crate::DocBuilderOptions;
 use log::debug;
 use std::collections::BTreeSet;

src/docbuilder/queue.rs

@@ -1,8 +1,6 @@
 //! Updates registry index and builds new packages
 
 use super::{DocBuilder, RustwideBuilder};
-use crate::config::Config;
-use crate::db::Pool;
 use crate::error::Result;
 use crate::utils::get_crate_priority;
 use crates_index_diff::ChangeKind;
@@ -87,319 +85,3 @@ impl DocBuilder {
         Ok(processed)
     }
 }
-
-#[derive(Debug, Eq, PartialEq, serde::Serialize)]
-pub(crate) struct QueuedCrate {
-    #[serde(skip)]
-    id: i32,
-    pub(crate) name: String,
-    pub(crate) version: String,
-    pub(crate) priority: i32,
-}
-
-#[derive(Debug)]
-pub struct BuildQueue {
-    db: Pool,
-    max_attempts: i32,
-}
-
-impl BuildQueue {
-    pub fn new(db: Pool, config: &Config) -> Self {
-        BuildQueue {
-            db,
-            max_attempts: config.build_attempts.into(),
-        }
-    }
-
-    pub fn add_crate(&self, name: &str, version: &str, priority: i32) -> Result<()> {
-        self.db.get()?.execute(
-            "INSERT INTO queue (name, version, priority) VALUES ($1, $2, $3);",
-            &[&name, &version, &priority],
-        )?;
-        Ok(())
-    }
-
-    pub(crate) fn pending_count(&self) -> Result<usize> {
-        let res = self.db.get()?.query(
-            "SELECT COUNT(*) FROM queue WHERE attempt < $1;",
-            &[&self.max_attempts],
-        )?;
-        Ok(res.get(0).get::<_, i64>(0) as usize)
-    }
-
-    pub(crate) fn prioritized_count(&self) -> Result<usize> {
-        let res = self.db.get()?.query(
-            "SELECT COUNT(*) FROM queue WHERE attempt < $1 AND priority <= 0;",
-            &[&self.max_attempts],
-        )?;
-        Ok(res.get(0).get::<_, i64>(0) as usize)
-    }
-
-    pub(crate) fn failed_count(&self) -> Result<usize> {
-        let res = self.db.get()?.query(
-            "SELECT COUNT(*) FROM queue WHERE attempt >= $1;",
-            &[&self.max_attempts],
-        )?;
-        Ok(res.get(0).get::<_, i64>(0) as usize)
-    }
-
-    pub(crate) fn queued_crates(&self) -> Result<Vec<QueuedCrate>> {
-        let query = self.db.get()?.query(
-            "SELECT id, name, version, priority
-             FROM queue
-             WHERE attempt < $1
-             ORDER BY priority ASC, attempt ASC, id ASC",
-            &[&self.max_attempts],
-        )?;
-
-        Ok(query
-            .into_iter()
-            .map(|row| QueuedCrate {
-                id: row.get("id"),
-                name: row.get("name"),
-                version: row.get("version"),
-                priority: row.get("priority"),
-            })
-            .collect())
-    }
-
-    pub(crate) fn process_next_crate(
-        &self,
-        f: impl FnOnce(&QueuedCrate) -> Result<()>,
-    ) -> Result<()> {
-        let conn = self.db.get()?;
-
-        let queued = self.queued_crates()?;
-        let to_process = match queued.get(0) {
-            Some(krate) => krate,
-            None => return Ok(()),
-        };
-
-        match f(&to_process) {
-            Ok(()) => {
-                conn.execute("DELETE FROM queue WHERE id = $1;", &[&to_process.id])?;
-                crate::web::metrics::TOTAL_BUILDS.inc();
-            }
-            Err(e) => {
-                // Increase attempt count
-                let rows = conn.query(
-                    "UPDATE queue SET attempt = attempt + 1 WHERE id = $1 RETURNING attempt;",
-                    &[&to_process.id],
-                )?;
-                let attempt: i32 = rows.get(0).get(0);
-
-                if attempt >= self.max_attempts {
-                    crate::web::metrics::FAILED_BUILDS.inc();
-                }
-
-                error!(
-                    "Failed to build package {}-{} from queue: {}\nBacktrace: {}",
-                    to_process.name,
-                    to_process.version,
-                    e,
-                    e.backtrace()
-                );
-            }
-        }
-
-        Ok(())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_add_and_process_crates() {
-        const MAX_ATTEMPTS: u16 = 3;
-
-        crate::test::wrapper(|env| {
-            env.override_config(|config| {
-                config.build_attempts = MAX_ATTEMPTS;
-            });
-
-            let queue = env.build_queue();
-
-            let test_crates = [
-                ("low-priority", "1.0.0", 1000),
-                ("high-priority-foo", "1.0.0", -1000),
-                ("medium-priority", "1.0.0", -10),
-                ("high-priority-bar", "1.0.0", -1000),
-                ("standard-priority", "1.0.0", 0),
-                ("high-priority-baz", "1.0.0", -1000),
-            ];
-            for krate in &test_crates {
-                queue.add_crate(krate.0, krate.1, krate.2)?;
-            }
-
-            let assert_next = |name| -> Result<()> {
-                queue.process_next_crate(|krate| {
-                    assert_eq!(name, krate.name);
-                    Ok(())
-                })?;
-                Ok(())
-            };
-            let assert_next_and_fail = |name| -> Result<()> {
-                queue.process_next_crate(|krate| {
-                    assert_eq!(name, krate.name);
-                    failure::bail!("simulate a failure");
-                })?;
-                Ok(())
-            };
-
-            // The first processed item is the one with the highest priority added first.
-            assert_next("high-priority-foo")?;
-
-            // Simulate a failure in high-priority-bar.
-            assert_next_and_fail("high-priority-bar")?;
-
-            // Continue with the next high priority crate.
-            assert_next("high-priority-baz")?;
-
-            // After all the crates with the max priority are processed, before starting to process
-            // crates with a lower priority the failed crates with the max priority will be tried
-            // again.
-            assert_next("high-priority-bar")?;
-
-            // Continue processing according to the priority.
-            assert_next("medium-priority")?;
-            assert_next("standard-priority")?;
-
-            // Simulate the crate failing many times.
-            for _ in 0..MAX_ATTEMPTS {
-                assert_next_and_fail("low-priority")?;
-            }
-
-            // Since low-priority failed many times it will be removed from the queue. Because of
-            // that the queue should now be empty.
-            let mut called = false;
-            queue.process_next_crate(|_| {
-                called = true;
-                Ok(())
-            })?;
-            assert!(!called, "there were still items in the queue");
-
-            Ok(())
-        })
-    }
-
-    #[test]
-    fn test_pending_count() {
-        crate::test::wrapper(|env| {
-            let queue = env.build_queue();
-
-            assert_eq!(queue.pending_count()?, 0);
-            queue.add_crate("foo", "1.0.0", 0)?;
-            assert_eq!(queue.pending_count()?, 1);
-            queue.add_crate("bar", "1.0.0", 0)?;
-            assert_eq!(queue.pending_count()?, 2);
-
-            queue.process_next_crate(|krate| {
-                assert_eq!("foo", krate.name);
-                Ok(())
-            })?;
-            assert_eq!(queue.pending_count()?, 1);
-
-            Ok(())
-        });
-    }
-
-    #[test]
-    fn test_prioritized_count() {
-        crate::test::wrapper(|env| {
-            let queue = env.build_queue();
-
-            assert_eq!(queue.prioritized_count()?, 0);
-            queue.add_crate("foo", "1.0.0", 0)?;
-            assert_eq!(queue.prioritized_count()?, 1);
-            queue.add_crate("bar", "1.0.0", -100)?;
-            assert_eq!(queue.prioritized_count()?, 2);
-            queue.add_crate("baz", "1.0.0", 100)?;
-            assert_eq!(queue.prioritized_count()?, 2);
-
-            queue.process_next_crate(|krate| {
-                assert_eq!("bar", krate.name);
-                Ok(())
-            })?;
-            assert_eq!(queue.prioritized_count()?, 1);
-
-            Ok(())
-        });
-    }
-
-    #[test]
-    fn test_failed_count() {
-        const MAX_ATTEMPTS: u16 = 3;
-        crate::test::wrapper(|env| {
-            env.override_config(|config| {
-                config.build_attempts = 3;
-            });
-            let queue = env.build_queue();
-
-            assert_eq!(queue.failed_count()?, 0);
-            queue.add_crate("foo", "1.0.0", -100)?;
-            assert_eq!(queue.failed_count()?, 0);
-            queue.add_crate("bar", "1.0.0", 0)?;
-
-            for _ in 0..MAX_ATTEMPTS {
-                assert_eq!(queue.failed_count()?, 0);
-                queue.process_next_crate(|krate| {
-                    assert_eq!("foo", krate.name);
-                    failure::bail!("this failed");
-                })?;
-            }
-            assert_eq!(queue.failed_count()?, 1);
-
-            queue.process_next_crate(|krate| {
-                assert_eq!("bar", krate.name);
-                Ok(())
-            })?;
-            assert_eq!(queue.failed_count()?, 1);
-
-            Ok(())
-        });
-    }
-
-    #[test]
-    fn test_queued_crates() {
-        crate::test::wrapper(|env| {
-            let queue = env.build_queue();
-
-            let test_crates = [
-                ("foo", "1.0.0", -10),
-                ("bar", "1.0.0", 0),
-                ("baz", "1.0.0", 10),
-            ];
-            for krate in &test_crates {
-                queue.add_crate(krate.0, krate.1, krate.2)?;
-            }
-
-            assert_eq!(
-                vec![
-                    QueuedCrate {
-                        id: 1,
-                        name: "foo".into(),
-                        version: "1.0.0".into(),
-                        priority: -10,
-                    },
-                    QueuedCrate {
-                        id: 2,
-                        name: "bar".into(),
-                        version: "1.0.0".into(),
-                        priority: 0,
-                    },
-                    QueuedCrate {
-                        id: 3,
-                        name: "baz".into(),
-                        version: "1.0.0".into(),
-                        priority: 10,
-                    },
-                ],
-                queue.queued_crates()?
-            );
-
-            Ok(())
-        });
-    }
-}

src/lib.rs

@@ -4,16 +4,17 @@
 
 pub use self::config::Config;
 pub use self::docbuilder::options::DocBuilderOptions;
-pub use self::docbuilder::BuildQueue;
 pub use self::docbuilder::DocBuilder;
 pub use self::docbuilder::RustwideBuilder;
+pub use self::queue::BuildQueue;
 pub use self::web::Server;
 
 mod config;
 pub mod db;
 mod docbuilder;
 mod error;
 mod index;
+mod queue;
 pub mod storage;
 #[cfg(test)]
 mod test;