diff --git a/bot/code_review_bot/workflow.py b/bot/code_review_bot/workflow.py index a4f873c0d..fb612e3fe 100644 --- a/bot/code_review_bot/workflow.py +++ b/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 diff --git a/bot/tests/conftest.py b/bot/tests/conftest.py index a103e47c7..8ead0302e 100644 --- a/bot/tests/conftest.py +++ b/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): """ diff --git a/bot/tests/test_workflow.py b/bot/tests/test_workflow.py index 29eb99f21..6d59371a8 100644 --- a/bot/tests/test_workflow.py +++ b/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