limit number of concurrent gb connections

src-tauri/Cargo.lock

@@ -1545,6 +1545,7 @@ dependencies = [
  "rand 0.8.5",
  "refinery",
  "reqwest",
+ "rsevents-extra",
  "rusqlite",
  "rustix",
  "sentry",
@@ -3821,6 +3822,24 @@ dependencies = [
  "zeroize",
 ]
 
+[[package]]
+name = "rsevents"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e8e29769fda0194d58e06d876d82337aac382415df547f073f44ffea0bb0c651"
+dependencies = [
+ "parking_lot_core",
+]
+
+[[package]]
+name = "rsevents-extra"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8ff64e850751883803e4b4837a37fd5d349d9cea4fe665df3c16887a08e936e4"
+dependencies = [
+ "rsevents",
+]
+
 [[package]]
 name = "rusqlite"
 version = "0.29.0"

src-tauri/Cargo.toml

@@ -60,6 +60,7 @@ rustix = "0.38.8"
 backoff = "0.4.0"
 byteorder = "1.4.3"
 num_cpus = "1.16.0"
+rsevents-extra = "0.2.2"
 
 [features]
 # by default Tauri runs in production mode

src-tauri/src/watcher/handlers/fetch_gitbutler_data.rs

@@ -1,6 +1,9 @@
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
 use std::{path, time};
 
 use anyhow::{Context, Result};
+use rsevents_extra::Semaphore;
 use tauri::AppHandle;
 
 use crate::{gb_repository, projects, users};
@@ -9,14 +12,37 @@ use super::events;
 
 #[derive(Clone)]
 pub struct Handler {
-    local_data_dir: path::PathBuf,
-    project_storage: projects::Storage,
-    user_storage: users::Storage,
+    inner: Arc<HandlerInner>,
 }
 
 impl TryFrom<&AppHandle> for Handler {
     type Error = anyhow::Error;
 
+    fn try_from(value: &AppHandle) -> std::result::Result<Self, Self::Error> {
+        let inner = HandlerInner::try_from(value)?;
+        Ok(Self {
+            inner: Arc::new(inner),
+        })
+    }
+}
+
+impl Handler {
+    pub fn handle(&self, project_id: &str, now: &time::SystemTime) -> Result<Vec<events::Event>> {
+        self.inner.handle(project_id, now)
+    }
+}
+
+struct HandlerInner {
+    local_data_dir: path::PathBuf,
+    project_storage: projects::Storage,
+    user_storage: users::Storage,
+
+    semaphores: Arc<Mutex<HashMap<String, Semaphore>>>,
+}
+
+impl TryFrom<&AppHandle> for HandlerInner {
+    type Error = anyhow::Error;
+
     fn try_from(value: &AppHandle) -> std::result::Result<Self, Self::Error> {
         let local_data_dir = value
             .path_resolver()
@@ -26,12 +52,20 @@ impl TryFrom<&AppHandle> for Handler {
             local_data_dir: local_data_dir.to_path_buf(),
             project_storage: projects::Storage::try_from(value)?,
             user_storage: users::Storage::try_from(value)?,
+            semaphores: Arc::new(Mutex::new(HashMap::new())),
         })
     }
 }
 
-impl Handler {
+impl HandlerInner {
     pub fn handle(&self, project_id: &str, now: &time::SystemTime) -> Result<Vec<events::Event>> {
+        // one fetch at a time
+        let mut semaphores = self.semaphores.lock().unwrap();
+        let _guard = semaphores
+            .entry(project_id.to_string())
+            .or_insert_with(|| Semaphore::new(0, 1))
+            .wait();
+
         let gb_repo = gb_repository::Repository::open(
             self.local_data_dir.clone(),
             project_id,

src-tauri/src/watcher/handlers/push_gitbutler_data.rs

@@ -1,6 +1,9 @@
+use std::collections::HashMap;
 use std::path;
+use std::sync::{Arc, Mutex};
 
 use anyhow::{Context, Result};
+use rsevents_extra::Semaphore;
 use tauri::AppHandle;
 
 use crate::{gb_repository, projects, users};
@@ -9,14 +12,36 @@ use super::events;
 
 #[derive(Clone)]
 pub struct Handler {
-    local_data_dir: path::PathBuf,
-    project_storage: projects::Storage,
-    user_storage: users::Storage,
+    inner: Arc<HandlerInner>,
 }
 
 impl TryFrom<&AppHandle> for Handler {
     type Error = anyhow::Error;
 
+    fn try_from(value: &AppHandle) -> std::result::Result<Self, Self::Error> {
+        Ok(Self {
+            inner: Arc::new(HandlerInner::try_from(value)?),
+        })
+    }
+}
+
+impl Handler {
+    pub fn handle(&self, project_id: &str) -> Result<Vec<events::Event>> {
+        self.inner.handle(project_id)
+    }
+}
+
+struct HandlerInner {
+    local_data_dir: path::PathBuf,
+    project_storage: projects::Storage,
+    user_storage: users::Storage,
+
+    semaphores: Arc<Mutex<HashMap<String, Semaphore>>>,
+}
+
+impl TryFrom<&AppHandle> for HandlerInner {
+    type Error = anyhow::Error;
+
     fn try_from(value: &AppHandle) -> std::result::Result<Self, Self::Error> {
         let local_data_dir = value
             .path_resolver()
@@ -26,12 +51,20 @@ impl TryFrom<&AppHandle> for Handler {
             local_data_dir: local_data_dir.to_path_buf(),
             project_storage: projects::Storage::try_from(value)?,
             user_storage: users::Storage::try_from(value)?,
+            semaphores: Arc::new(Mutex::new(HashMap::new())),
         })
     }
 }
 
-impl Handler {
+impl HandlerInner {
     pub fn handle(&self, project_id: &str) -> Result<Vec<events::Event>> {
+        // one push at a time
+        let mut semaphores = self.semaphores.lock().unwrap();
+        let _guard = semaphores
+            .entry(project_id.to_string())
+            .or_insert_with(|| Semaphore::new(0, 1))
+            .wait();
+
         let gb_repo = gb_repository::Repository::open(
             &self.local_data_dir,
             project_id,