Start cloning the repository in the background as soon as possible

bot/code_review_bot/workflow.py

@@ -6,6 +6,7 @@
 import json
 import re
 import time
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timedelta
 from itertools import groupby
 
@@ -102,11 +103,18 @@ def __init__(
 
         # Is local clone already setup ?
         self.clone_available = False
+        # Background clone in progress, see start_clone & clone_repository
+        self.clone_executor = None
+        self.clone_future = None
 
     def run(self, revision):
         """
         Find all issues on remote tasks and publish them
         """
+        # Start cloning the local repo in the background ASAP
+        # It is only awaited when the issues hashes are needed
+        self.start_clone(revision)
+
         # Index ASAP Taskcluster task for this revision
         self.index(revision, state="started")
 
@@ -194,6 +202,10 @@ def ingest_revision(self, revision, group_id):
             self.backend_api.enabled
         ), "Backend storage is disabled, revision ingestion is not possible"
 
+        # Start cloning the local repo in the background ASAP
+        # It is only awaited when the issues hashes are needed
+        self.start_clone(revision)
+
         # Index ASAP Taskcluster task for this revision
         self.index(revision, state="ingestion")
 
@@ -325,22 +337,30 @@ def start_analysis(self, revision):
             cache_root=settings.mercurial_cache,
         )
 
-        # Try to update the state 5 consecutive time
-        for i in range(5):
-            # Update the internal build state using Phabricator infos
-            phabricator.update_state(build)
+        # Clone the required repository in the background
+        # while waiting for the build to become public
+        with ThreadPoolExecutor(max_workers=1) as executor:
+            clone = executor.submit(repository.clone)
 
-            # Continue with workflow once the build is public
-            if build.state is PhabricatorBuildState.Public:
-                break
+            # Try to update the state 5 consecutive time
+            for i in range(5):
+                # Update the internal build state using Phabricator infos
+                phabricator.update_state(build)
 
-            # Retry later if the build is not yet seen as public
-            logger.warning(
-                "Build is not public, retrying in 30s",
-                build=build,
-                retries_left=build.retries,
-            )
-            time.sleep(30)
+                # Continue with workflow once the build is public
+                if build.state is PhabricatorBuildState.Public:
+                    break
+
+                # Retry later if the build is not yet seen as public
+                logger.warning(
+                    "Build is not public, retrying in 30s",
+                    build=build,
+                    retries_left=build.retries,
+                )
+                time.sleep(30)
+
+            # Wait for the clone to be finished, raising on failure
+            clone.result()
 
         # Make sure the build is now public
         if build.state is not PhabricatorBuildState.Public:
@@ -350,9 +370,6 @@ def start_analysis(self, revision):
         if not build.stack:
             raise Exception("No stack of patches to apply.")
 
-        # We'll clone the required repository
-        repository.clone()
-
         # Apply the stack of patches and push to try
         worker = MercurialWorker()
         output = worker.run(repository, build)
@@ -378,19 +395,60 @@ def start_analysis(self, revision):
         else:
             logger.info("Skipping Lando publication")
 
-    def clone_repository(self, revision):
+    def start_clone(self, revision):
         """
-        Clone the repo locally when configured
-        On production this should use a Taskcluster cache
+        Start cloning the repo locally in a background thread when configured
+        Use clone_repository to wait for the clone to be available
         """
         if self.clone_available:
             logger.debug("Local clone already setup")
             return
 
+        if self.clone_future is not None:
+            logger.debug("Local clone already in progress")
+            return
+
         if not settings.mercurial_cache and not settings.git_cache:
             logger.info("Local clone not required")
             return
 
+        logger.info("Starting local clone in the background")
+        self.clone_executor = ThreadPoolExecutor(
+            max_workers=1, thread_name_prefix="clone"
+        )
+        self.clone_future = self.clone_executor.submit(self._clone, revision)
+
+    def clone_repository(self, revision):
+        """
+        Make sure the repo is cloned locally when configured
+        Wait for a background clone started with start_clone, or run it now
+        On production this should use a Taskcluster cache
+        """
+        if self.clone_available:
+            logger.debug("Local clone already setup")
+            return
+
+        if self.clone_future is None:
+            self.start_clone(revision)
+
+        # Clone not required
+        if self.clone_future is None:
+            return
+
+        logger.info("Waiting for local clone to be available")
+        try:
+            self.clone_future.result()
+        finally:
+            self.clone_future = None
+            self.clone_executor.shutdown(wait=False)
+            self.clone_executor = None
+
+        self.clone_available = True
+
+    def _clone(self, revision):
+        """
+        Effectively clone the repo locally
+        """
         if isinstance(revision, PhabricatorRevision):
             # Mercurial clone
             if not settings.mercurial_cache:
@@ -429,8 +487,6 @@ def clone_repository(self, revision):
         else:
             raise NotImplementedError
 
-        self.clone_available = True
-
     def publish(self, revision, issues, task_failures, notices, reviewers):
         """
         Publish issues on selected reporters

bot/tests/conftest.py

@@ -701,6 +701,8 @@ def __init__(self):
             self.update_build = False
             self.task_failures_ignored = []
             self.clone_available = True
+            self.clone_executor = None
+            self.clone_future = None
 
         def setup_mock_tasks(self, tasks):
             """

bot/tests/test_workflow.py

@@ -678,3 +678,121 @@ def test_cancel_previous_ignores_later_buildables(mock_config, mock_workflow):
     assert abort_calls(mock_workflow.phabricator) == [
         {"receiver": "PHID-HMBB-old", "type": "abort"}
     ]
+
+
+def test_clone_repository_background(
+    mock_workflow, mock_revision, monkeypatch, tmp_path
+):
+    """
+    The clone is started in the background and awaited by clone_repository
+    """
+    import threading
+
+    started = threading.Event()
+    release = threading.Event()
+    calls = []
+
+    def _robust_checkout(**kwargs):
+        calls.append(kwargs)
+        assert threading.current_thread() is not threading.main_thread()
+        started.set()
+        assert release.wait(timeout=5)
+
+    monkeypatch.setattr("code_review_bot.workflow.robust_checkout", _robust_checkout)
+    monkeypatch.setattr(
+        "code_review_bot.workflow.settings.mercurial_cache", tmp_path / "hg"
+    )
+    mock_workflow.clone_available = False
+
+    mock_workflow.start_clone(mock_revision)
+    assert mock_workflow.clone_future is not None
+    assert started.wait(timeout=5)
+    assert mock_workflow.clone_available is False
+
+    # Starting twice does not clone twice
+    mock_workflow.start_clone(mock_revision)
+    assert len(calls) == 1
+
+    # Waiting blocks until the clone is done
+    waiter = threading.Thread(
+        target=mock_workflow.clone_repository, args=(mock_revision,)
+    )
+    waiter.start()
+    waiter.join(timeout=0.5)
+    assert waiter.is_alive()
+    release.set()
+    waiter.join(timeout=5)
+    assert not waiter.is_alive()
+
+    assert mock_workflow.clone_available is True
+    assert mock_workflow.clone_future is None
+    assert calls == [
+        {
+            "repo_upstream_url": mock_revision.base_repository,
+            "repo_url": mock_revision.head_repository,
+            "revision": mock_revision.head_changeset,
+            "checkout_dir": tmp_path / "hg" / "checkout",
+            "sharebase_dir": tmp_path / "hg" / "shared",
+        }
+    ]
+
+    # Once cloned, nothing else is triggered
+    mock_workflow.clone_repository(mock_revision)
+    mock_workflow.start_clone(mock_revision)
+    assert len(calls) == 1
+
+
+def test_clone_repository_without_start(
+    mock_workflow, mock_revision, monkeypatch, tmp_path
+):
+    """
+    clone_repository runs the clone itself when it was not started before
+    """
+    checkout = mock.Mock()
+    monkeypatch.setattr("code_review_bot.workflow.robust_checkout", checkout)
+    monkeypatch.setattr(
+        "code_review_bot.workflow.settings.mercurial_cache", tmp_path / "hg"
+    )
+    mock_workflow.clone_available = False
+
+    mock_workflow.clone_repository(mock_revision)
+    assert checkout.call_count == 1
+    assert mock_workflow.clone_available is True
+
+
+def test_clone_repository_failure(mock_workflow, mock_revision, monkeypatch, tmp_path):
+    """
+    A failure in the background clone is raised when waiting for it
+    """
+
+    def _robust_checkout(**kwargs):
+        raise RuntimeError("hg is broken")
+
+    monkeypatch.setattr("code_review_bot.workflow.robust_checkout", _robust_checkout)
+    monkeypatch.setattr(
+        "code_review_bot.workflow.settings.mercurial_cache", tmp_path / "hg"
+    )
+    mock_workflow.clone_available = False
+
+    mock_workflow.start_clone(mock_revision)
+    with pytest.raises(RuntimeError, match="hg is broken"):
+        mock_workflow.clone_repository(mock_revision)
+    assert mock_workflow.clone_available is False
+    assert mock_workflow.clone_future is None
+
+
+def test_clone_repository_not_required(mock_workflow, mock_revision, monkeypatch):
+    """
+    No clone is started when no local cache is configured
+    """
+    checkout = mock.Mock()
+    monkeypatch.setattr("code_review_bot.workflow.robust_checkout", checkout)
+    monkeypatch.setattr("code_review_bot.workflow.settings.mercurial_cache", None)
+    monkeypatch.setattr("code_review_bot.workflow.settings.git_cache", None)
+    mock_workflow.clone_available = False
+
+    mock_workflow.start_clone(mock_revision)
+    assert mock_workflow.clone_future is None
+    mock_workflow.clone_repository(mock_revision)
+    assert checkout.call_count == 0
+    assert mock_workflow.clone_available is False