refactor: repo-actor: replace Mutex with RwLock

crates/git/src/repository/mod.rs

@@ -1,5 +1,5 @@
 //
-use std::sync::{atomic::AtomicBool, Arc, Mutex};
+use std::sync::{atomic::AtomicBool, Arc, RwLock};
 
 use derive_more::Deref as _;
 
@@ -80,7 +80,7 @@ struct RealRepositoryFactory;
 impl RepositoryFactory for RealRepositoryFactory {
     fn open(&self, gitdir: &GitDir) -> Result<Box<dyn OpenRepositoryLike>> {
         let gix_repo = gix::ThreadSafeRepository::open(gitdir.to_path_buf())?.to_thread_local();
-        let repo = RealOpenRepository::new(Arc::new(Mutex::new(gix_repo)));
+        let repo = RealOpenRepository::new(Arc::new(RwLock::new(gix_repo.into())));
         Ok(Box::new(repo))
     }
 
@@ -93,7 +93,7 @@ impl RepositoryFactory for RealRepositoryFactory {
         )?
         .fetch_only(gix::progress::Discard, &AtomicBool::new(false))?;
         tracing::info!("created");
-        let repo = RealOpenRepository::new(Arc::new(Mutex::new(gix_repo)));
+        let repo = RealOpenRepository::new(Arc::new(RwLock::new(gix_repo.into())));
         Ok(Box::new(repo))
     }
 

crates/git/src/repository/open/mod.rs

@@ -10,7 +10,7 @@ pub mod ofake;
 
 use std::{
     path::Path,
-    sync::{Arc, Mutex},
+    sync::{Arc, RwLock},
 };
 
 use crate as git;
@@ -39,8 +39,8 @@ pub enum OpenRepository {
 
 #[cfg(not(tarpaulin_include))]
 pub fn real(gix_repo: gix::Repository) -> OpenRepository {
-    OpenRepository::Real(oreal::RealOpenRepository::new(Arc::new(Mutex::new(
+    OpenRepository::Real(oreal::RealOpenRepository::new(Arc::new(RwLock::new(
-        gix_repo,
+        gix_repo.into(),
     ))))
 }
 

crates/git/src/repository/open/ofake.rs

@@ -4,14 +4,14 @@ use git_next_config as config;
 
 use std::{
     path::Path,
-    sync::{Arc, Mutex},
+    sync::{Arc, RwLock},
 };
 
 #[derive(Clone, Debug, Default)]
 pub struct FakeOpenRepository {
-    default_push_remote: Arc<Mutex<Option<git::GitRemote>>>,
+    default_push_remote: Arc<RwLock<Option<git::GitRemote>>>,
-    default_fetch_remote: Arc<Mutex<Option<git::GitRemote>>>,
+    default_fetch_remote: Arc<RwLock<Option<git::GitRemote>>>,
-    operations: Arc<Mutex<Vec<String>>>,
+    operations: Arc<RwLock<Vec<String>>>,
 }
 impl FakeOpenRepository {
     pub fn new() -> Self {
@@ -26,7 +26,7 @@ impl FakeOpenRepository {
         match direction {
             git::repository::Direction::Push => self
                 .default_push_remote
-                .lock()
+                .write()
                 .map(|mut o| match remote {
                     Some(gr) => o.replace(gr),
                     None => o.take(),
@@ -34,7 +34,7 @@ impl FakeOpenRepository {
                 .unwrap(),
             git::repository::Direction::Fetch => self
                 .default_fetch_remote
-                .lock()
+                .write()
                 .map(|mut o| match remote {
                     Some(gr) => o.replace(gr),
                     None => o.take(),
@@ -45,7 +45,7 @@ impl FakeOpenRepository {
 
     pub fn operations(&self) -> Vec<String> {
         self.operations
-            .lock()
+            .read()
             .map(|operations| operations.clone())
             .unwrap_or_default()
     }
@@ -60,12 +60,12 @@ impl git::repository::OpenRepositoryLike for FakeOpenRepository {
         match direction {
             git::repository::Direction::Push => self
                 .default_push_remote
-                .lock()
+                .read()
                 .map(|r| r.clone())
                 .unwrap_or(None),
             git::repository::Direction::Fetch => self
                 .default_fetch_remote
-                .lock()
+                .read()
                 .map(|r| r.clone())
                 .unwrap_or(None),
         }
@@ -73,7 +73,7 @@ impl git::repository::OpenRepositoryLike for FakeOpenRepository {
 
     fn fetch(&self) -> core::result::Result<(), crate::fetch::Error> {
         self.operations
-            .lock()
+            .write()
             .map_err(|_| crate::fetch::Error::Lock)
             .map(|mut operations| operations.push("fetch".to_string()))?;
         Ok(())
@@ -89,7 +89,7 @@ impl git::repository::OpenRepositoryLike for FakeOpenRepository {
         let forge_alias = repo_details.forge.forge_alias();
         let repo_alias = &repo_details.repo_alias;
         self.operations
-            .lock()
+            .write()
             .map_err(|_| crate::fetch::Error::Lock)
             .map(|mut operations| {
                 operations.push(format!(

crates/git/src/repository/open/oreal.rs

@@ -7,20 +7,20 @@ use git_next_config as config;
 use gix::bstr::BStr;
 use std::{
     path::Path,
-    sync::{Arc, Mutex},
+    sync::{Arc, RwLock},
 };
 use tracing::{info, warn};
 
 #[derive(Clone, Debug, Constructor)]
-pub struct RealOpenRepository(Arc<Mutex<gix::Repository>>);
+pub struct RealOpenRepository(Arc<RwLock<gix::ThreadSafeRepository>>);
 impl super::OpenRepositoryLike for RealOpenRepository {
     fn remote_branches(&self) -> git::push::Result<Vec<config::BranchName>> {
         let refs = self
             .0
-            .lock()
+            .read()
             .map_err(|_| git::push::Error::Lock)
             .and_then(|repo| {
-                Ok(repo.references()?).and_then(|refs| {
+                Ok(repo.to_thread_local().references()?).and_then(|refs| {
                     Ok(refs.remote_branches().map(|rb| {
                         rb.filter_map(|rbi| rbi.ok())
                             .map(|r| r.name().to_owned())
@@ -37,11 +37,12 @@ impl super::OpenRepositoryLike for RealOpenRepository {
         Ok(refs)
     }
     fn find_default_remote(&self, direction: git::repository::Direction) -> Option<git::GitRemote> {
-        let Ok(repository) = self.0.lock() else {
+        let Ok(repository) = self.0.read() else {
             #[cfg(not(tarpaulin_include))] // don't test mutex lock failure
             return None;
         };
-        let Some(Ok(remote)) = repository.find_default_remote(direction.into()) else {
+        let thread_local = repository.to_thread_local();
+        let Some(Ok(remote)) = thread_local.find_default_remote(direction.into()) else {
             #[cfg(not(tarpaulin_include))] // test is on local repo - should always have remotes
             return None;
         };
@@ -51,12 +52,13 @@ impl super::OpenRepositoryLike for RealOpenRepository {
     #[tracing::instrument(skip_all)]
     #[cfg(not(tarpaulin_include))] // would require writing to external service
     fn fetch(&self) -> Result<(), git::fetch::Error> {
-        let Ok(repository) = self.0.lock() else {
+        let Ok(repository) = self.0.read() else {
             #[cfg(not(tarpaulin_include))] // don't test mutex lock failure
             return Err(git::fetch::Error::Lock);
         };
+        let thread_local = repository.to_thread_local();
         let Some(Ok(remote)) =
-            repository.find_default_remote(git::repository::Direction::Fetch.into())
+            thread_local.find_default_remote(git::repository::Direction::Fetch.into())
         else {
             #[cfg(not(tarpaulin_include))] // test is on local repo - should always have remotes
             return Err(git::fetch::Error::NoFetchRemoteFound);
@@ -98,7 +100,7 @@ impl super::OpenRepositoryLike for RealOpenRepository {
         .into();
         let git_dir = self
             .0
-            .lock()
+            .read()
             .map_err(|_| git::push::Error::Lock)
             .map(|r| r.git_dir().to_path_buf())?;
         let ctx = gix::diff::command::Context {
@@ -125,12 +127,13 @@ impl super::OpenRepositoryLike for RealOpenRepository {
             false => 50,
         };
         self.0
-            .lock()
+            .read()
             .map_err(|_| git::commit::log::Error::Lock)
             .map(|repo| {
                 let branch = format!("remotes/origin/{branch_name}");
                 let branch = BStr::new(&branch);
-                let branch_head = repo
+                let thread_local = repo.to_thread_local();
+                let branch_head = thread_local
                     .rev_parse_single(branch)
                     .map_err(|e| e.to_string())
                     .map_err(as_gix_error(branch_name.clone()))?;
@@ -142,7 +145,7 @@ impl super::OpenRepositoryLike for RealOpenRepository {
                     .try_into_commit()
                     .map_err(|e| e.to_string())
                     .map_err(as_gix_error(branch_name.clone()))?;
-                let walk = repo
+                let walk = thread_local
                     .rev_walk([commit.id])
                     .all()
                     .map_err(|e| e.to_string())
@@ -183,13 +186,15 @@ impl super::OpenRepositoryLike for RealOpenRepository {
         file_name: &Path,
     ) -> git::file::Result<String> {
         self.0
-            .lock()
+            .read()
             .map_err(|_| git::file::Error::Lock)
             .and_then(|repo| {
-                let fref = repo.find_reference(format!("origin/{}", branch_name).as_str())?;
+                let thread_local = repo.to_thread_local();
+                let fref =
+                    thread_local.find_reference(format!("origin/{}", branch_name).as_str())?;
                 let id = fref.try_id().ok_or(git::file::Error::TryId)?;
                 let oid = id.detach();
-                let obj = repo.find_object(oid)?;
+                let obj = thread_local.find_object(oid)?;
                 let commit = obj.into_commit();
                 let tree = commit.tree()?;
                 let ent = tree

crates/git/src/repository/open/otest.rs

@@ -5,7 +5,7 @@ use git_next_config as config;
 
 use std::{
     path::Path,
-    sync::{Arc, Mutex, RwLock},
+    sync::{Arc, RwLock},
 };
 
 pub type OnFetchFn =
@@ -154,7 +154,7 @@ impl TestOpenRepository {
             fetch_counter: Arc::new(RwLock::new(0)),
             on_push,
             push_counter: Arc::new(RwLock::new(0)),
-            real: git::repository::RealOpenRepository::new(Arc::new(Mutex::new(gix))),
+            real: git::repository::RealOpenRepository::new(Arc::new(RwLock::new(gix.into()))),
         }
     }
 

crates/repo-actor/src/lib.rs

@@ -16,9 +16,9 @@ use kxio::network::Network;
 use tracing::{info, warn, Instrument};
 
 #[derive(Clone, Debug, Default)]
-pub struct RepoActorLog(std::sync::Arc<std::sync::Mutex<Vec<String>>>);
+pub struct RepoActorLog(std::sync::Arc<std::sync::RwLock<Vec<String>>>);
 impl Deref for RepoActorLog {
-    type Target = std::sync::Arc<std::sync::Mutex<Vec<String>>>;
+    type Target = std::sync::Arc<std::sync::RwLock<Vec<String>>>;
 
     fn deref(&self) -> &Self::Target {
         &self.0
@@ -135,6 +135,6 @@ pub fn logger(log: &Option<crate::RepoActorLog>, message: impl Into<String>) {
     if let Some(log) = log {
         let message: String = message.into();
         tracing::debug!(message);
-        let _ = log.lock().map(|mut l| l.push(message));
+        let _ = log.write().map(|mut l| l.push(message));
     }
 }

crates/repo-actor/src/tests/handlers/advance_main.rs

@@ -38,7 +38,7 @@ async fn when_repo_config_should_fetch_then_push_then_revalidate() -> TestResult
 
     //then
     tracing::debug!(?log, "");
-    log.lock().map_err(|e| e.to_string()).map(|l| {
+    log.read().map_err(|e| e.to_string()).map(|l| {
         assert!(l
             .iter()
             .any(|message| message.contains("send: LoadConfigFromRepo")))
@@ -83,7 +83,7 @@ async fn when_server_config_should_fetch_then_push_then_revalidate() -> TestResu
 
     //then
     tracing::debug!(?log, "");
-    log.lock().map_err(|e| e.to_string()).map(|l| {
+    log.read().map_err(|e| e.to_string()).map(|l| {
         assert!(l
             .iter()
             .any(|message| message.contains("send: ValidateRepo")))

crates/repo-actor/src/tests/handlers/advance_next.rs

@@ -40,7 +40,7 @@ async fn should_fetch_then_push_then_revalidate() -> TestResult {
 
     //then
     tracing::debug!(?log, "");
-    log.lock().map_err(|e| e.to_string()).map(|l| {
+    log.read().map_err(|e| e.to_string()).map(|l| {
         assert!(l
             .iter()
             .any(|message| message.contains("send: ValidateRepo")))

crates/repo-actor/src/tests/handlers/check_ci_status.rs

@@ -26,7 +26,7 @@ async fn should_passthrough_to_receive_ci_status() -> TestResult {
 
     //then
     tracing::debug!(?log, "");
-    log.lock().map_err(|e| e.to_string()).map(|l| {
+    log.read().map_err(|e| e.to_string()).map(|l| {
         assert!(l
             .iter()
             .any(|message| message.contains("send: ReceiveCIStatus")))

crates/repo-actor/src/tests/handlers/clone_repo.rs

@@ -15,13 +15,13 @@ async fn should_clone() -> TestResult {
 
     // factory clones an open repository
     let mut repository_factory = MockRepositoryFactory::new();
-    let cloned = Arc::new(Mutex::new(vec![]));
+    let cloned = Arc::new(RwLock::new(vec![]));
     let cloned_ref = cloned.clone();
     repository_factory
         .expect_git_clone()
         .times(2)
         .return_once(move |_| {
-            let _ = cloned_ref.lock().map(|mut l| l.push(()));
+            let _ = cloned_ref.write().map(|mut l| l.push(()));
             Ok(Box::new(open_repository))
         });
 
@@ -32,7 +32,7 @@ async fn should_clone() -> TestResult {
 
     //then
     cloned
-        .lock()
+        .read()
         .map_err(|e| e.to_string())
         .map(|o| assert_eq!(o.len(), 1))?;
 
@@ -52,13 +52,13 @@ async fn should_open() -> TestResult {
 
     // factory opens a repository
     let mut repository_factory = MockRepositoryFactory::new();
-    let opened = Arc::new(Mutex::new(vec![]));
+    let opened = Arc::new(RwLock::new(vec![]));
     let opened_ref = opened.clone();
     repository_factory
         .expect_open()
         .times(1)
         .return_once(move |_| {
-            let _ = opened_ref.lock().map(|mut l| l.push(()));
+            let _ = opened_ref.write().map(|mut l| l.push(()));
             Ok(Box::new(open_repository))
         });
     fs.dir_create(&repo_details.gitdir)?;
@@ -70,7 +70,7 @@ async fn should_open() -> TestResult {
 
     //then
     opened
-        .lock()
+        .read()
         .map_err(|e| e.to_string())
         .map(|o| assert_eq!(o.len(), 1))?;
 
@@ -231,7 +231,7 @@ async fn valid_repo_with_default_remotes_should_assess_branch_positions() -> Tes
     System::current().stop();
 
     //then
-    log.lock()
+    log.read()
         .map_err(|e| e.to_string())
         .map(|l| assert!(l.iter().any(|l| l.contains("send: ValidateRepo"))))?;
 

crates/repo-actor/src/tests/handlers/load_config_from_repo.rs

@@ -45,7 +45,7 @@ async fn when_read_file_ok_should_send_config_loaded() -> TestResult {
 
     //then
     tracing::debug!(?log, "");
-    log.lock().map_err(|e| e.to_string()).map(|l| {
+    log.read().map_err(|e| e.to_string()).map(|l| {
         assert!(l
             .iter()
             .any(|message| message.contains("send: LoadedConfig")))
@@ -80,7 +80,7 @@ async fn when_read_file_err_should_notify_user() -> TestResult {
 
     //then
     tracing::debug!(?log, "");
-    log.lock().map_err(|e| e.to_string()).map(|l| {
+    log.read().map_err(|e| e.to_string()).map(|l| {
         assert!(l.iter().any(|message| message.contains("send: NotifyUser")));
         assert!(
             !l.iter()

crates/repo-actor/src/tests/handlers/loaded_config.rs

@@ -54,7 +54,7 @@ async fn should_validate_repo() -> TestResult {
 
     //then
     tracing::debug!(?log, "");
-    log.lock().map_err(|e| e.to_string()).map(|l| {
+    log.read().map_err(|e| e.to_string()).map(|l| {
         assert!(l
             .iter()
             .any(|message| message.contains("send: ValidateRepo")))
@@ -84,7 +84,7 @@ async fn should_register_webhook() -> TestResult {
 
     //then
     tracing::debug!(?log, "");
-    log.lock().map_err(|e| e.to_string()).map(|l| {
+    log.read().map_err(|e| e.to_string()).map(|l| {
         assert!(l
             .iter()
             .any(|message| message.contains("send: RegisterWebhook")))

crates/repo-actor/src/tests/handlers/receive_ci_status.rs

@@ -23,7 +23,7 @@ async fn when_pass_should_advance_main_to_next() -> TestResult {
 
     //then
     tracing::debug!(?log, "");
-    log.lock().map_err(|e| e.to_string()).map(|l| {
+    log.read().map_err(|e| e.to_string()).map(|l| {
         let expected = format!("send: AdvanceMain({:?})", next_commit);
         tracing::debug!(%expected,"");
         assert!(l.iter().any(|message| message.contains(&expected)))
@@ -53,7 +53,7 @@ async fn when_pending_should_recheck_ci_status() -> TestResult {
 
     //then
     tracing::debug!(?log, "");
-    log.lock().map_err(|e| e.to_string()).map(|l| {
+    log.read().map_err(|e| e.to_string()).map(|l| {
         assert!(l
             .iter()
             .any(|message| message.contains("send: ValidateRepo")))
@@ -84,7 +84,7 @@ async fn when_fail_should_notify_user() -> TestResult {
 
     //then
     tracing::debug!(?log, "");
-    log.lock()
+    log.read()
         .map_err(|e| e.to_string())
         .map(|l| assert!(l.iter().any(|message| message.contains("send: NotifyUser"))))?;
     Ok(())

crates/repo-actor/src/tests/handlers/register_webhook.rs

@@ -27,7 +27,7 @@ async fn when_registered_ok_should_send_webhook_registered() -> TestResult {
 
     //then
     tracing::debug!(?log, "");
-    log.lock().map_err(|e| e.to_string()).map(|l| {
+    log.read().map_err(|e| e.to_string()).map(|l| {
         assert!(l
             .iter()
             .any(|message| message.contains("send: WebhookRegistered")))
@@ -63,7 +63,7 @@ async fn when_registered_error_should_send_notify_user() -> TestResult {
 
     //then
     tracing::debug!(?log, "");
-    log.lock()
+    log.read()
         .map_err(|e| e.to_string())
         .map(|l| assert!(l.iter().any(|message| message.contains("send: NotifyUser"))))?;
     Ok(())

crates/repo-actor/src/tests/handlers/validate_repo.rs

@@ -43,12 +43,7 @@ async fn repo_with_next_not_an_ancestor_of_dev_should_be_reset() -> TestResult {
     System::current().stop();
 
     //then
-    tracing::debug!(?log, "");
+    log.require_message_containing("NextBranchResetRequired")?;
-    log.lock().map_err(|e| e.to_string()).map(|l| {
-        assert!(l
-            .iter()
-            .any(|message| message.contains("NextBranchResetRequired")))
-    })?;
     Ok(())
 }
 
@@ -94,12 +89,7 @@ async fn repo_with_next_not_on_or_near_main_should_be_reset() -> TestResult {
     System::current().stop();
 
     //then
-    tracing::debug!(?log, "");
+    log.require_message_containing("NextBranchResetRequired")?;
-    log.lock().map_err(|e| e.to_string()).map(|l| {
-        assert!(l
-            .iter()
-            .any(|message| message.contains("NextBranchResetRequired")))
-    })?;
     Ok(())
 }
 
@@ -145,12 +135,7 @@ async fn repo_with_next_not_based_on_main_should_be_reset() -> TestResult {
     System::current().stop();
 
     //then
-    tracing::debug!(?log, "");
+    log.require_message_containing("NextBranchResetRequired")?;
-    log.lock().map_err(|e| e.to_string()).map(|l| {
-        assert!(l
-            .iter()
-            .any(|message| message.contains("NextBranchResetRequired")))
-    })?;
     Ok(())
 }
 
@@ -195,11 +180,7 @@ async fn repo_with_next_ahead_of_main_should_check_ci_status() -> TestResult {
     System::current().stop();
 
     //then
-    tracing::debug!(?log, "");
+    log.require_message_containing(format!("send: CheckCIStatus({next_commit:?})"))?;
-    log.lock().map_err(|e| e.to_string()).map(|l| {
-        let expected = format!("send: CheckCIStatus({next_commit:?})");
-        assert!(l.iter().any(|message| message.contains(&expected)))
-    })?;
     Ok(())
 }
 
@@ -245,10 +226,7 @@ async fn repo_with_dev_and_next_on_main_should_do_nothing() -> TestResult {
     System::current().stop();
 
     //then
-    tracing::debug!(?log, "");
+    log.no_message_contains("send:")?;
-    log.lock()
-        .map_err(|e| e.to_string())
-        .map(|l| assert!(!l.iter().any(|message| message.contains("send:"))))?;
     Ok(())
 }
 
@@ -295,11 +273,9 @@ async fn repo_with_dev_ahead_of_next_should_advance_next() -> TestResult {
     System::current().stop();
 
     //then
-    tracing::debug!(?log, "");
+    log.require_message_containing(format!(
-    log.lock().map_err(|e| e.to_string()).map(|l| {
+        "send: AdvanceNext(({next_commit:?}, {dev_branch_log:?}))"
-        let expected = format!("send: AdvanceNext(({next_commit:?}, {dev_branch_log:?}))");
+    ))?;
-        assert!(l.iter().any(|message| message.contains(&expected)))
-    })?;
     Ok(())
 }
 
@@ -325,12 +301,7 @@ async fn should_accept_message_with_current_token() -> TestResult {
     System::current().stop();
 
     //then
-    tracing::debug!(?log, "");
+    log.require_message_containing("accepted token: 2")?;
-    log.lock().map_err(|e| e.to_string()).map(|l| {
-        assert!(l
-            .iter()
-            .any(|message| message.contains("accepted token: 2")))
-    })?;
     Ok(())
 }
 
@@ -354,12 +325,7 @@ async fn should_accept_message_with_new_token() -> TestResult {
     System::current().stop();
 
     //then
-    tracing::debug!(?log, "");
+    log.require_message_containing("accepted token: 3")?;
-    log.lock().map_err(|e| e.to_string()).map(|l| {
-        assert!(l
-            .iter()
-            .any(|message| message.contains("accepted token: 3")))
-    })?;
     Ok(())
 }
 
@@ -383,9 +349,6 @@ async fn should_reject_message_with_expired_token() -> TestResult {
     System::current().stop();
 
     //then
-    tracing::debug!(?log, "");
+    log.no_message_contains("accepted token")?;
-    log.lock()
-        .map_err(|e| e.to_string())
-        .map(|l| assert!(!l.iter().any(|message| message.contains("accepted token"))))?;
     Ok(())
 }

crates/repo-actor/src/tests/mod.rs

@@ -22,7 +22,7 @@ use git_next_git as git;
 use mockall::predicate::eq;
 use std::{
     collections::{BTreeMap, HashMap},
-    sync::{Arc, Mutex},
+    sync::{Arc, RwLock},
 };
 
 type TestResult = Result<(), Box<dyn std::error::Error>>;
@@ -59,7 +59,7 @@ impl RepoActorLog {
         needle: impl AsRef<str>,
     ) -> Result<bool, Box<dyn std::error::Error>> {
         let found = self
-            .lock()
+            .read()
             .map_err(|e| e.to_string())?
             .iter()
             .any(|message| message.contains(needle.as_ref()));