refactor: Add support for multithreading

Cargo.toml

@@ -35,6 +35,7 @@ octocrab = "0.35.0"
 # Async
 futures = "0.3"
 tokio = { workspace = true }
+arc-swap = "1.7.1"
 
 # Web
 hyper = "1.2.0"

src/bin/bors.rs

@@ -80,9 +80,8 @@ fn try_main(opts: Opts) -> anyhow::Result<()> {
     let state = runtime.block_on(GithubAppState::load(
         opts.app_id.into(),
         opts.private_key.into_bytes().into(),
-        db,
     ))?;
-    let ctx = BorsContext::new(CommandParser::new(opts.cmd_prefix));
+    let ctx = BorsContext::new(CommandParser::new(opts.cmd_prefix), Arc::new(db));
     let (tx, gh_process) = create_bors_process(state, ctx);
 
     let refresh_tx = tx.clone();

src/bors/context.rs

@@ -1,11 +1,13 @@
-use crate::bors::command::CommandParser;
+use crate::{bors::command::CommandParser, database::DbClient};
+use std::sync::Arc;
 
 pub struct BorsContext {
     pub parser: CommandParser,
+    pub db: Arc<dyn DbClient>,
 }
 
 impl BorsContext {
-    pub fn new(parser: CommandParser) -> Self {
-        Self { parser }
+    pub fn new(parser: CommandParser, db: Arc<dyn DbClient>) -> Self {
+        Self { parser, db }
     }
 }

src/bors/handlers/help.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use crate::bors::RepositoryClient;
 use crate::bors::RepositoryState;
 use crate::github::PullRequest;
@@ -10,7 +12,7 @@ const HELP_MESSAGE: &str = r#"
 "#;
 
 pub(super) async fn command_help<Client: RepositoryClient>(
-    repo: &mut RepositoryState<Client>,
+    repo: Arc<RepositoryState<Client>>,
     pr: &PullRequest,
 ) -> anyhow::Result<()> {
     repo.client.post_comment(pr.number, HELP_MESSAGE).await?;
@@ -25,7 +27,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_help() {
-        let mut state = ClientBuilder::default().create_state().await;
+        let state = ClientBuilder::default().create_state().await;
         state.comment("@bors help").await;
         state
             .client()

src/bors/handlers/labels.rs

@@ -7,27 +7,28 @@ use crate::github::{LabelModification, LabelTrigger, PullRequestNumber};
 /// If there are any label modifications that should be performed on the given PR when `trigger`
 /// happens, this function will perform them.
 pub async fn handle_label_trigger<Client: RepositoryClient>(
-    repo: &mut RepositoryState<Client>,
+    repo: &RepositoryState<Client>,
     pr: PullRequestNumber,
     trigger: LabelTrigger,
 ) -> anyhow::Result<()> {
-    if let Some(modifications) = repo.config.labels.get(&trigger) {
+    let mut add: Vec<String> = Vec::new();
+    let mut remove: Vec<String> = Vec::new();
+    if let Some(modifications) = repo.config.load().labels.get(&trigger) {
         log::debug!("Performing label modifications {modifications:?}");
-        let (add, remove): (Vec<_>, Vec<_>) =
-            modifications
-                .iter()
-                .partition_map(|modification| match modification {
-                    LabelModification::Add(label) => itertools::Either::Left(label.clone()),
-                    LabelModification::Remove(label) => itertools::Either::Right(label.clone()),
-                });
-        if !add.is_empty() {
-            log::info!("Adding label(s) {add:?}");
-            repo.client.add_labels(pr, &add).await?;
-        }
-        if !remove.is_empty() {
-            log::info!("Removing label(s) {remove:?}");
-            repo.client.remove_labels(pr, &remove).await?;
-        }
+        (add, remove) = modifications
+            .iter()
+            .partition_map(|modification| match modification {
+                LabelModification::Add(label) => itertools::Either::Left(label.clone()),
+                LabelModification::Remove(label) => itertools::Either::Right(label.clone()),
+            });
+    }
+    if !add.is_empty() {
+        log::info!("Adding label(s) {add:?}");
+        repo.client.add_labels(pr, &add).await?;
+    }
+    if !remove.is_empty() {
+        log::info!("Removing label(s) {remove:?}");
+        repo.client.remove_labels(pr, &remove).await?;
     }
     Ok(())
 }

src/bors/handlers/mod.rs

@@ -1,4 +1,5 @@
 use anyhow::Context;
+use std::sync::Arc;
 use tracing::Instrument;
 
 use crate::bors::command::BorsCommand;
@@ -26,9 +27,10 @@
 /// This function performs a single BORS event, it is the main execution function of the bot.
 pub async fn handle_bors_event<Client: RepositoryClient>(
     event: BorsEvent,
-    state: &mut dyn BorsState<Client>,
-    ctx: &BorsContext,
+    state: Arc<dyn BorsState<Client>>,
+    ctx: Arc<BorsContext>,
 ) -> anyhow::Result<()> {
+    let db = Arc::clone(&ctx.db);
     match event {
         BorsEvent::Comment(comment) => {
             // We want to ignore comments made by this bot
@@ -37,14 +39,14 @@
                 return Ok(());
             }
 
-            if let Some((repo, db)) = get_repo_state(state, &comment.repository) {
+            if let Some(repo) = get_repo_state(state, &comment.repository) {
                 let span = tracing::info_span!(
                     "Comment",
                     pr = format!("{}#{}", comment.repository, comment.pr_number),
                     author = comment.author.username
                 );
                 let pr_number = comment.pr_number;
-                if let Err(error) = handle_comment(repo, db, ctx, comment)
+                if let Err(error) = handle_comment(Arc::clone(&repo), db, ctx, comment)
                     .instrument(span.clone())
                     .await
                 {
@@ -66,7 +68,7 @@
             }
         }
         BorsEvent::WorkflowStarted(payload) => {
-            if let Some((_, db)) = get_repo_state(state, &payload.repository) {
+            if let Some(_) = get_repo_state(state, &payload.repository) {
                 let span = tracing::info_span!(
                     "Workflow started",
                     repo = payload.repository.to_string(),
@@ -81,7 +83,7 @@
             }
         }
         BorsEvent::WorkflowCompleted(payload) => {
-            if let Some((repo, db)) = get_repo_state(state, &payload.repository) {
+            if let Some(repo) = get_repo_state(state, &payload.repository) {
                 let span = tracing::info_span!(
                     "Workflow completed",
                     repo = payload.repository.to_string(),
@@ -96,7 +98,7 @@
             }
         }
         BorsEvent::CheckSuiteCompleted(payload) => {
-            if let Some((repo, db)) = get_repo_state(state, &payload.repository) {
+            if let Some(repo) = get_repo_state(state, &payload.repository) {
                 let span = tracing::info_span!(
                     "Check suite completed",
                     repo = payload.repository.to_string(),
@@ -111,10 +113,15 @@
         }
         BorsEvent::Refresh => {
             let span = tracing::info_span!("Refresh");
-            let (repos, db) = state.get_all_repos_mut();
-            futures::future::join_all(repos.into_iter().map(|repo| async {
-                let subspan = tracing::info_span!("Repo", repo = repo.repository.to_string());
-                refresh_repository(repo, db).instrument(subspan).await
+            let repos = state.get_all_repos();
+            futures::future::join_all(repos.into_iter().map(|repo| {
+                let repo = Arc::clone(&repo);
+                async {
+                    let subspan = tracing::info_span!("Repo", repo = repo.repository.to_string());
+                    refresh_repository(repo, Arc::clone(&db))
+                        .instrument(subspan)
+                        .await
+                }
             }))
             .instrument(span)
             .await;
@@ -123,11 +130,11 @@
     Ok(())
 }
 
-fn get_repo_state<'a, Client: RepositoryClient>(
-    state: &'a mut dyn BorsState<Client>,
+fn get_repo_state<Client: RepositoryClient>(
+    state: Arc<dyn BorsState<Client>>,
     repo: &GithubRepoName,
-) -> Option<(&'a mut RepositoryState<Client>, &'a mut dyn DbClient)> {
-    match state.get_repo_state_mut(repo) {
+) -> Option<Arc<RepositoryState<Client>>> {
+    match state.get_repo_state(repo) {
         Some(result) => Some(result),
         None => {
             tracing::warn!("Repository {} not found", repo);
@@ -137,9 +144,9 @@
 }
 
 async fn handle_comment<Client: RepositoryClient>(
-    repo: &mut RepositoryState<Client>,
-    database: &mut dyn DbClient,
-    ctx: &BorsContext,
+    repo: Arc<RepositoryState<Client>>,
+    database: Arc<dyn DbClient>,
+    ctx: Arc<BorsContext>,
     comment: PullRequestComment,
 ) -> anyhow::Result<()> {
     let pr_number = comment.pr_number;
@@ -163,6 +170,8 @@
     for command in commands {
         match command {
             Ok(command) => {
+                let repo = Arc::clone(&repo);
+                let database = Arc::clone(&database);
                 let result = match command {
                     BorsCommand::Help => {
                         let span = tracing::info_span!("Help");
@@ -233,7 +242,7 @@
 
     #[tokio::test]
     async fn test_ignore_bot_comment() {
-        let mut state = ClientBuilder::default().create_state().await;
+        let state = ClientBuilder::default().create_state().await;
         state
             .comment(comment("@bors ping").author(test_bot_user()).create())
             .await;
@@ -242,8 +251,10 @@
 
     #[tokio::test]
     async fn test_do_not_comment_when_pr_fetch_fails() {
-        let mut state = ClientBuilder::default().create_state().await;
-        state.client().get_pr_fn = Box::new(|_| Err(anyhow::anyhow!("Foo")));
+        let state = ClientBuilder::default().create_state().await;
+        state
+            .client()
+            .set_get_pr_fn(|_| Err(anyhow::anyhow!("Foo")));
         state.comment(comment("foo").create()).await;
         state.client().check_comments(default_pr_number(), &[]);
     }

src/bors/handlers/ping.rs

@@ -1,9 +1,11 @@
+use std::sync::Arc;
+
 use crate::bors::RepositoryClient;
 use crate::bors::RepositoryState;
 use crate::github::PullRequest;
 
 pub(super) async fn command_ping<Client: RepositoryClient>(
-    repo: &mut RepositoryState<Client>,
+    repo: Arc<RepositoryState<Client>>,
     pr: &PullRequest,
 ) -> anyhow::Result<()> {
     repo.client.post_comment(pr.number, "Pong 🏓!").await?;
@@ -17,7 +19,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_ping() {
-        let mut state = ClientBuilder::default().create_state().await;
+        let state = ClientBuilder::default().create_state().await;
         state.comment("@bors ping").await;
         state
             .client()

src/bors/handlers/refresh.rs

@@ -1,31 +1,39 @@
+use std::sync::Arc;
 use std::time::Duration;
 
 use chrono::{DateTime, Utc};
 
 use crate::bors::handlers::trybuild::cancel_build_workflows;
 use crate::bors::{RepositoryClient, RepositoryState};
 use crate::database::{BuildStatus, DbClient};
+use crate::permissions::load_permissions;
 
 pub async fn refresh_repository<Client: RepositoryClient>(
-    repo: &mut RepositoryState<Client>,
-    db: &dyn DbClient,
+    repo: Arc<RepositoryState<Client>>,
+    db: Arc<dyn DbClient>,
 ) -> anyhow::Result<()> {
-    let res = cancel_timed_out_builds(repo, db).await;
-    reload_permission(repo).await;
-    reload_config(repo).await?;
-
-    res
+    let repo = repo.as_ref();
+    if let (Ok(_), _, Ok(_)) = tokio::join!(
+        cancel_timed_out_builds(repo, db.as_ref()),
+        reload_permission(repo),
+        reload_config(repo)
+    ) {
+        Ok(())
+    } else {
+        tracing::error!("Failed to refresh repository");
+        anyhow::bail!("Failed to refresh repository")
+    }
 }
 
 async fn cancel_timed_out_builds<Client: RepositoryClient>(
-    repo: &mut RepositoryState<Client>,
+    repo: &RepositoryState<Client>,
     db: &dyn DbClient,
 ) -> anyhow::Result<()> {
     let running_builds = db.get_running_builds(&repo.repository).await?;
     tracing::info!("Found {} running build(s)", running_builds.len());
 
+    let timeout = repo.config.load().timeout.clone();
     for build in running_builds {
-        let timeout = repo.config.timeout;
         if elapsed_time(build.created_at) >= timeout {
             tracing::info!("Cancelling build {}", build.commit_sha);
 
@@ -54,15 +62,19 @@
     Ok(())
 }
 
-async fn reload_permission<Client: RepositoryClient>(repo: &mut RepositoryState<Client>) {
-    repo.permissions_resolver.reload().await
+async fn reload_permission<Client: RepositoryClient>(
+    repo: &RepositoryState<Client>,
+) -> anyhow::Result<()> {
+    let permissions = load_permissions(&repo.repository).await?;
+    repo.permissions.store(Arc::new(permissions));
+    Ok(())
 }
 
 async fn reload_config<Client: RepositoryClient>(
-    repo: &mut RepositoryState<Client>,
+    repo: &RepositoryState<Client>,
 ) -> anyhow::Result<()> {
     let config = repo.client.load_config().await?;
-    repo.config = config;
+    repo.config.store(Arc::new(config));
     Ok(())
 }
 
@@ -89,7 +101,6 @@
 #[cfg(test)]
 mod tests {
     use std::future::Future;
-    use std::sync::{Arc, Mutex};
     use std::time::Duration;
 
     use chrono::Utc;
@@ -99,7 +110,6 @@
     use crate::bors::handlers::trybuild::TRY_BRANCH_NAME;
     use crate::database::DbClient;
     use crate::tests::event::{default_pr_number, WorkflowStartedBuilder};
-    use crate::tests::permissions::MockPermissions;
     use crate::tests::state::{default_repo_name, ClientBuilder, RepoConfigBuilder};
 
     #[tokio::test(flavor = "current_thread")]
@@ -108,17 +118,6 @@
         state.refresh().await;
     }
 
-    #[tokio::test(flavor = "current_thread")]
-    async fn refresh_permission() {
-        let permission_resolver = Arc::new(Mutex::new(MockPermissions::default()));
-        let mut state = ClientBuilder::default()
-            .permission_resolver(Box::new(Arc::clone(&permission_resolver)))
-            .create_state()
-            .await;
-        state.refresh().await;
-        assert_eq!(permission_resolver.lock().unwrap().num_reload_called, 1);
-    }
-
     #[tokio::test(flavor = "current_thread")]
     async fn refresh_do_nothing_before_timeout() {
         let mut state = ClientBuilder::default()