fix(core): automatically cleanup dangling git processes

pyproject.toml

@@ -20,7 +20,7 @@
 
 [tool.poetry]
 name = "renku"
-version = "0.0.0" # placeholder, see poetry-dynamic-versioning
+version = "1.2.4.dev17+geb9386e8" # placeholder, see poetry-dynamic-versioning
 description = "Python SDK and CLI for the Renku platform."
 license = "Apache License 2.0"
 keywords = ["Renku", "CLI"]

renku/command/command_builder/command.py

@@ -171,6 +171,7 @@ def __init__(self) -> None:
         self._track_std_streams: bool = False
         self._working_directory: Optional[str] = None
         self._client: Optional["LocalClient"] = None
+        self._client_was_created: bool = False
 
     def __getattr__(self, name: str) -> Any:
         """Bubble up attributes of wrapped builders."""
@@ -205,6 +206,7 @@ def _injection_pre_hook(self, builder: "Command", context: dict, *args, **kwargs
                 dispatcher.push_created_client_to_stack(self._client)
             else:
                 self._client = dispatcher.push_client_to_stack(path=default_path(self._working_directory or "."))
+                self._client_was_created = True
             ctx = click.Context(click.Command(builder._operation))  # type: ignore
         else:
             if not self._client:
@@ -237,6 +239,9 @@ def _post_hook(self, builder: "Command", context: dict, result: "CommandResult",
         """
         remove_injector()
 
+        if self._client_was_created and self._client and self._client.repository is not None:
+            self._client.repository.close()
+
         if result.error:
             raise result.error
 

renku/core/management/git.py

@@ -295,6 +295,10 @@ def __attrs_post_init__(self):
         except errors.GitError:
             self.repository = None
 
+    def __del__(self):
+        if self.repository:
+            self.repository.close()
+
     @property
     def modified_paths(self):
         """Return paths of modified files."""

renku/infrastructure/repository.py

@@ -86,6 +86,15 @@ def __init__(self, path: Union[Path, str] = ".", repository: Optional[git.Repo]
     def __repr__(self) -> str:
         return f"<{self.__class__.__name__} {self.path}>"
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self.close()
+
+    def __del__(self):
+        self.close()
+
     @property
     def path(self) -> Path:
         """Absolute path to the repository's root."""
@@ -675,6 +684,22 @@ def get_user(self) -> "Actor":
         configuration = self.get_configuration()
         return Repository._get_user_from_configuration(configuration)
 
+    def close(self) -> None:
+        """Close the underlying repository.
+
+        Cleans up dangling processes.
+        """
+        if self._repository:
+            self._repository.close()
+            del self._repository
+            self._repository = None
+
+    def refresh_repository(self):
+        """Refreshes the underlying repository."""
+        self.close()
+
+        self._repository = git.Repo(self.path, search_parent_directories=False)
+
     @staticmethod
     def get_global_user() -> "Actor":
         """Return the global git user."""
@@ -795,6 +820,7 @@ def __init__(
         self, path: Union[Path, str] = ".", search_parent_directories: bool = False, repository: git.Repo = None
     ):
         repo = repository or _create_repository(path, search_parent_directories)
+
         super().__init__(path=Path(repo.working_dir).resolve(), repository=repo)  # type: ignore
 
     @classmethod
@@ -881,6 +907,9 @@ def __str__(self) -> str:
     def __repr__(self) -> str:
         return f"<Submodule {self.relative_path}>"
 
+    def __del__(self) -> None:
+        self._repository.close()
+
     @property
     def name(self) -> str:
         """Return submodule's name."""
@@ -902,22 +931,32 @@ class SubmoduleManager:
 
     def __init__(self, repository: git.Repo):
         self._repository = repository
+        self._submodule_cache: Dict[git.Submodule, Submodule] = {}  # type: ignore
         try:
             self.update()
         except errors.GitError:
             # NOTE: Update fails if submodule repo cannot be cloned. Repository still works but submodules are broken.
             pass
 
+    def _get_submodule(self, submodule: git.Submodule) -> Submodule:  # type: ignore
+        """Get a submodule from local cache."""
+        if submodule not in self._submodule_cache:
+            submodule_result = Submodule.from_submodule(self._repository, submodule)
+            self._submodule_cache[submodule] = submodule_result
+        return self._submodule_cache[submodule]
+
     def __getitem__(self, name: str) -> Submodule:
         try:
             submodule = self._repository.submodules[name]
         except IndexError:
             raise errors.GitError(f"Submodule '{name}' not found")
         else:
-            return Submodule.from_submodule(self._repository, submodule)
+            return self._get_submodule(submodule)
 
     def __iter__(self):
-        return (Submodule.from_submodule(self._repository, s) for s in self._repository.submodules)
+        for s in self._repository.submodules:
+
+            yield self._get_submodule(s)
 
     def __len__(self) -> int:
         return len(self._repository.submodules)

renku/ui/service/controllers/api/mixins.py

@@ -190,37 +190,37 @@ def execute_op(self):
                 ref = self.request_data.get("ref", None)
 
                 if ref:
-                    repository = Repository(project.abs_path)
-                    if ref != repository.active_branch.name:
-                        # NOTE: Command called for different branch than the one used in cache, change branch
-                        if len(repository.remotes) != 1:
-                            raise RenkuException("Couldn't find remote for project in cache.")
-                        origin = repository.remotes[0]
-                        remote_branch = f"{origin}/{ref}"
-
-                        with project.write_lock():
-                            # NOTE: Add new ref to remote branches
-                            repository.run_git_command("remote", "set-branches", "--add", origin, ref)
-                            if self.migrate_project or self.clone_depth == PROJECT_CLONE_NO_DEPTH:
-                                repository.fetch(origin, ref)
-                            else:
-                                repository.fetch(origin, ref, depth=self.clone_depth)
-
-                            # NOTE: Switch to new ref
-                            repository.run_git_command("checkout", "--track", "-f", "-b", ref, remote_branch)
-
-                            # NOTE: cleanup remote branches in case a remote was deleted (fetch fails otherwise)
-                            repository.run_git_command("remote", "prune", origin)
-
-                            for branch in repository.branches:
-                                if branch.remote_branch and not branch.remote_branch.is_valid():
-                                    repository.branches.remove(branch, force=True)
-                                    # NOTE: Remove left-over refspec
-                                    try:
-                                        with repository.get_configuration(writable=True) as config:
-                                            config.remove_value(f"remote.{origin}.fetch", f"origin.{branch}$")
-                                    except GitConfigurationError:
-                                        pass
+                    with Repository(project.abs_path) as repository:
+                        if ref != repository.active_branch.name:
+                            # NOTE: Command called for different branch than the one used in cache, change branch
+                            if len(repository.remotes) != 1:
+                                raise RenkuException("Couldn't find remote for project in cache.")
+                            origin = repository.remotes[0]
+                            remote_branch = f"{origin}/{ref}"
+
+                            with project.write_lock():
+                                # NOTE: Add new ref to remote branches
+                                repository.run_git_command("remote", "set-branches", "--add", origin, ref)
+                                if self.migrate_project or self.clone_depth == PROJECT_CLONE_NO_DEPTH:
+                                    repository.fetch(origin, ref)
+                                else:
+                                    repository.fetch(origin, ref, depth=self.clone_depth)
+
+                                # NOTE: Switch to new ref
+                                repository.run_git_command("checkout", "--track", "-f", "-b", ref, remote_branch)
+
+                                # NOTE: cleanup remote branches in case a remote was deleted (fetch fails otherwise)
+                                repository.run_git_command("remote", "prune", origin)
+
+                                for branch in repository.branches:
+                                    if branch.remote_branch and not branch.remote_branch.is_valid():
+                                        repository.branches.remove(branch, force=True)
+                                        # NOTE: Remove left-over refspec
+                                        try:
+                                            with repository.get_configuration(writable=True) as config:
+                                                config.remove_value(f"remote.{origin}.fetch", f"origin.{branch}$")
+                                        except GitConfigurationError:
+                                            pass
                 else:
                     self.reset_local_repo(project)
 
@@ -250,33 +250,33 @@ def reset_local_repo(self, project):
                     # NOTE: return immediately in case of multiple writers waiting
                     return
 
-                repository = Repository(project.abs_path)
-                origin = None
-                tracking_branch = repository.active_branch.remote_branch
-                if tracking_branch:
-                    origin = tracking_branch.remote
-                elif len(repository.remotes) == 1:
-                    origin = repository.remotes[0]
-
-                if origin:
-                    unshallow = self.migrate_project or self.clone_depth == PROJECT_CLONE_NO_DEPTH
-                    if unshallow:
-                        try:
-                            # NOTE: It could happen that repository is already un-shallowed,
-                            # in this case we don't want to leak git exception, but still want to fetch.
-                            repository.fetch("origin", repository.active_branch, unshallow=True)
-                        except GitCommandError:
-                            repository.fetch("origin", repository.active_branch)
-
-                        repository.reset(f"{origin}/{repository.active_branch}", hard=True)
-                    else:
-                        try:
-                            # NOTE: it rarely happens that origin is not reachable. Try again if it fails.
-                            repository.fetch("origin", repository.active_branch)
+                with Repository(project.abs_path) as repository:
+                    origin = None
+                    tracking_branch = repository.active_branch.remote_branch
+                    if tracking_branch:
+                        origin = tracking_branch.remote
+                    elif len(repository.remotes) == 1:
+                        origin = repository.remotes[0]
+
+                    if origin:
+                        unshallow = self.migrate_project or self.clone_depth == PROJECT_CLONE_NO_DEPTH
+                        if unshallow:
+                            try:
+                                # NOTE: It could happen that repository is already un-shallowed,
+                                # in this case we don't want to leak git exception, but still want to fetch.
+                                repository.fetch("origin", repository.active_branch, unshallow=True)
+                            except GitCommandError:
+                                repository.fetch("origin", repository.active_branch)
+
                             repository.reset(f"{origin}/{repository.active_branch}", hard=True)
-                        except GitCommandError as e:
-                            project.purge()
-                            raise IntermittentCacheError(e)
+                        else:
+                            try:
+                                # NOTE: it rarely happens that origin is not reachable. Try again if it fails.
+                                repository.fetch("origin", repository.active_branch)
+                                repository.reset(f"{origin}/{repository.active_branch}", hard=True)
+                            except GitCommandError as e:
+                                project.purge()
+                                raise IntermittentCacheError(e)
                 project.last_fetched_at = datetime.utcnow()
                 project.save()
         except (portalocker.LockException, portalocker.AlreadyLocked) as e:
@@ -346,7 +346,8 @@ def sync(self, remote="origin"):
         if self.project_path is None:
             raise RenkuException("unable to sync with remote since no operation has been executed")
 
-        return push_changes(Repository(self.project_path), remote=remote)
+        with Repository(self.project_path) as repository:
+            return push_changes(repository, remote=remote)
 
     def execute_and_sync(self, remote="origin"):
         """Execute operation which controller implements and sync with the remote."""

renku/version.py

@@ -24,7 +24,7 @@
 except ImportError:
     from importlib_metadata import distribution
 
-__version__ = "0.0.0"
+__version__ = "1.2.4.dev17+geb9386e8"
 __template_version__ = "0.3.1"
 __minimum_project_version__ = "1.2.0"