Detect try tree closures and retry after waiting an exponential time (#81)

* Detect closed try tree and exponentially wait before retrying

* Use requests instead of aiohttp

* Update existing tests

* Add a test case

Author: vrigal

.isort.cfg

@@ -1,5 +1,5 @@
 [settings]
 known_first_party = conftest,libmozevent
-known_third_party = aioamqp,aiohttp,aioredis,hglib,libmozdata,pytest,responses,rs_parsepatch,setuptools,structlog,taskcluster
+known_third_party = aioamqp,aiohttp,aioredis,hglib,libmozdata,pytest,requests,responses,rs_parsepatch,setuptools,structlog,taskcluster
 force_single_line = True
 line_length = 159

.pre-commit-config.yaml

@@ -52,6 +52,7 @@ repos:
     rev: v0.950
     hooks:
       - id: mypy
+        additional_dependencies: ['types-requests']
   - repo: meta
     hooks:
       - id: check-useless-excludes

libmozevent/mercurial.py

@@ -3,15 +3,18 @@
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
+import asyncio
 import atexit
 import enum
 import io
 import json
 import os
 import tempfile
 import time
+from datetime import datetime
 
 import hglib
+import requests
 import rs_parsepatch
 import structlog
 from libmozdata.phabricator import PhabricatorPatch
@@ -24,8 +27,14 @@
 
 TREEHERDER_URL = "https://treeherder.mozilla.org/#/jobs?repo={}&revision={}"
 DEFAULT_AUTHOR = "libmozevent <release-mgmt-analysis@mozilla.com>"
+# On build failure, check try status until available every 5 minutes and up to 24h
+TRY_STATUS_URL = "https://treestatus.mozilla-releng.net/trees/try"
+TRY_STATUS_DELAY = 5 * 60
+TRY_STATUS_MAX_WAIT = 24 * 60 * 60
 # Number of allowed retries on an unexpected push fail
 MAX_PUSH_RETRIES = 4
+# Wait successive exponential delays: 6sec, 36sec, 3.6min, 21.6min
+PUSH_RETRY_EXPONENTIAL_DELAY = 6
 
 
 class TryMode(enum.Enum):
@@ -344,6 +353,34 @@ def get_files_touched_in_diff(rawdiff):
             for patched_file in get_files_touched_in_diff(rev.patch)
         )
 
+    async def wait_try_available(self):
+        """
+        Wait until try status is "open"
+        On each failure, wait TRY_STATUS_DELAY before retrying up to TRY_STATUS_MAX_WAIT
+        """
+
+        def get_status():
+            try:
+                resp = requests.get(TRY_STATUS_URL)
+                resp.raise_for_status()
+                data = resp.json()
+            except Exception as e:
+                logger.warning(f"An error occurred retrieving try status: {e}")
+            else:
+                return data.get("result", {}).get("status")
+
+        start = datetime.utcnow()
+        while status := get_status() != "open":
+            if (datetime.utcnow() - start).seconds >= TRY_STATUS_MAX_WAIT:
+                logger.error(
+                    f"Try tree status still closed after {TRY_STATUS_MAX_WAIT} seconds, skipping"
+                )
+                break
+            logger.warning(
+                f"Try tree is not actually open (status: {status}), waiting {TRY_STATUS_DELAY} seconds before retrying"
+            )
+            await asyncio.sleep(TRY_STATUS_DELAY)
+
     async def handle_build(self, repository, build):
         """
         Try to load and apply a diff on local clone
@@ -405,8 +442,16 @@ async def handle_build(self, repository, build):
                 error_log = error_log.decode("utf-8")
 
             if "push failed on remote" in error_log.lower():
-                # In case of an unexpected push fail, put the build task back in the queue
                 build.retries += 1
+                # Ensure try is opened
+                await self.wait_try_available()
+                # Wait an exponential time before retrying the build
+                delay = PUSH_RETRY_EXPONENTIAL_DELAY**build.retries
+                logger.info(
+                    f"An error occurred pushing the build to try, retrying after {delay}s"
+                )
+                await asyncio.sleep(delay)
+                # Put the build task back in the queue
                 await self.bus.send(self.queue_name, build)
                 return
 

requirements.txt

@@ -4,6 +4,7 @@ aioredis>=2.0.0
 libmozdata
 python-hglib
 pytoml
+requests
 rs_parsepatch
 structlog
 taskcluster>=24.0.0

tests/test_mercurial.py

@@ -6,6 +6,7 @@
 
 import hglib
 import pytest
+import responses
 
 from conftest import MockBuild
 from libmozevent.bus import MessageBus
@@ -647,6 +648,7 @@ async def test_crash_utf8_author(PhabricatorMock, mock_mc):
     assert details["revision"] == mock_mc.repo.tip().node.decode("utf-8")
 
 
+@responses.activate
 @pytest.mark.asyncio
 async def test_unexpected_push_failure(PhabricatorMock, mock_mc):
     """
@@ -669,6 +671,14 @@ async def test_unexpected_push_failure(PhabricatorMock, mock_mc):
     from libmozevent import mercurial
 
     mercurial.MAX_PUSH_RETRIES = 1
+    mercurial.TRY_STATUS_URL = "http://test.status/try"
+    mercurial.PUSH_RETRY_EXPONENTIAL_DELAY = 0
+    mercurial.TRY_STATUS_DELAY = 0
+    mercurial.TRY_STATUS_MAX_WAIT = 0
+
+    responses.get(
+        "http://test.status/try", status=200, json={"result": {"status": "open"}}
+    )
 
     repository_mock = MagicMock(spec=Repository)
     repository_mock.push_to_try.side_effect = [
@@ -712,8 +722,12 @@ async def test_unexpected_push_failure(PhabricatorMock, mock_mc):
         tip.node.decode("utf-8")
     )
     assert details["revision"] == tip.node.decode("utf-8")
+    assert [(call.request.method, call.request.url) for call in responses.calls] == [
+        ("GET", "http://test.status/try")
+    ]
 
 
+@responses.activate
 @pytest.mark.asyncio
 async def test_push_failure_max_retries(PhabricatorMock, mock_mc):
     """
@@ -736,6 +750,23 @@ async def test_push_failure_max_retries(PhabricatorMock, mock_mc):
     from libmozevent import mercurial
 
     mercurial.MAX_PUSH_RETRIES = 2
+    mercurial.TRY_STATUS_URL = "http://test.status/try"
+    mercurial.PUSH_RETRY_EXPONENTIAL_DELAY = 2
+    mercurial.TRY_STATUS_DELAY = 0
+    mercurial.TRY_STATUS_MAX_WAIT = 0
+
+    sleep_history = []
+
+    class AsyncioMock(object):
+        async def sleep(self, value):
+            nonlocal sleep_history
+            sleep_history.append(value)
+
+    mercurial.asyncio = AsyncioMock()
+
+    responses.get(
+        "http://test.status/try", status=200, json={"result": {"status": "open"}}
+    )
 
     repository_mock = MagicMock(spec=Repository)
     repository_mock.push_to_try.side_effect = hglib.error.CommandError(
@@ -767,3 +798,104 @@ async def test_push_failure_max_retries(PhabricatorMock, mock_mc):
         details["message"]
         == "Max number of retries has been reached pushing the build to try repository"
     )
+
+    assert [(call.request.method, call.request.url) for call in responses.calls] == [
+        ("GET", "http://test.status/try"),
+        ("GET", "http://test.status/try"),
+        ("GET", "http://test.status/try"),
+    ]
+    assert sleep_history == [2, 4, 8]
+
+
+@responses.activate
+@pytest.mark.asyncio
+async def test_push_closed_try(PhabricatorMock, mock_mc):
+    """
+    Detect when try tree is in a closed state and wait before it is opened to retry
+    """
+
+    diff = {
+        "revisionPHID": "PHID-DREV-badutf8",
+        "baseRevision": "missing",
+        "phid": "PHID-DIFF-badutf8",
+        "id": 555,
+    }
+    build = MockBuild(4444, "PHID-REPO-mc", 5555, "PHID-build-badutf8", diff)
+    with PhabricatorMock as phab:
+        phab.load_patches_stack(build)
+
+    bus = MessageBus()
+    bus.add_queue("phabricator")
+
+    from libmozevent import mercurial
+
+    mercurial.MAX_PUSH_RETRIES = 2
+    mercurial.TRY_STATUS_URL = "http://test.status/try"
+    mercurial.PUSH_RETRY_EXPONENTIAL_DELAY = 2
+    mercurial.TRY_STATUS_DELAY = 42
+    mercurial.TRY_STATUS_MAX_WAIT = 1
+
+    sleep_history = []
+
+    class AsyncioMock(object):
+        async def sleep(self, value):
+            nonlocal sleep_history
+            sleep_history.append(value)
+
+    mercurial.asyncio = AsyncioMock()
+
+    responses.get(
+        "http://test.status/try", status=200, json={"result": {"status": "closed"}}
+    )
+    responses.get("http://test.status/try", status=500)
+    responses.get(
+        "http://test.status/try", status=200, json={"result": {"status": "open"}}
+    )
+
+    repository_mock = MagicMock(spec=Repository)
+    repository_mock.push_to_try.side_effect = [
+        hglib.error.CommandError(
+            args=("push", "try_url"),
+            ret=1,
+            err="abort: push failed on remote",
+            out="",
+        ),
+        mock_mc.repo.tip(),
+    ]
+    repository_mock.try_name = "try"
+    repository_mock.retries = 0
+
+    worker = MercurialWorker(
+        "mercurial", "phabricator", repositories={"PHID-REPO-mc": repository_mock}
+    )
+    worker.register(bus)
+
+    assert bus.queues["mercurial"].qsize() == 0
+
+    resp = await worker.handle_build(repository_mock, build)
+    assert resp is None
+    assert repository_mock.push_to_try.call_count == 1
+
+    assert bus.queues["mercurial"].qsize() == 1
+    assert bus.queues["phabricator"].qsize() == 0
+
+    build = await bus.receive("mercurial")
+    resp = await worker.handle_build(repository_mock, build)
+    assert resp is not None
+    mode, out_build, details = resp
+
+    assert mode == "success"
+    assert out_build == build
+    tip = mock_mc.repo.tip()
+    assert details[
+        "treeherder_url"
+    ] == "https://treeherder.mozilla.org/#/jobs?repo=try&revision={}".format(
+        tip.node.decode("utf-8")
+    )
+    assert details["revision"] == tip.node.decode("utf-8")
+    assert [(call.request.method, call.request.url) for call in responses.calls] == [
+        ("GET", "http://test.status/try"),
+        ("GET", "http://test.status/try"),
+        ("GET", "http://test.status/try"),
+    ]
+    assert sleep_history == [42, 42, 2]