From 581e8b865cd6fc03aca357c7428a69f5c59ec6da Mon Sep 17 00:00:00 2001 From: Ben Hearsum Date: Fri, 21 Aug 2026 10:11:10 -0400 Subject: [PATCH] feat: cancel in-progress reviewbot pushes before new ones are started This was originally motivated by https://github.com/mozilla/code-review/pull/3578, but it's valid and worthwhile on its own. Aside from the very roundabout way we have to find task group ids, this is pretty straightforward: simply find and cancel all previous task groups for the revision. This avoids doing work for something that's already stale. If someone knows of a way to find the task group id of the reviewbot push other than the roundabout way it's happening here, I'd be very happy to switch to it. I could not find any way to pull it (or even the treeherder link) through the API directly; the only thing that seems to available there is the task id of the initial code review task. In addition to the unit tests, I managed to run code review bot locally as some sort of integration test. I didn't have it actually cancel any tasks, but I _think_ I've done enough to make this landable. Ideally we can test in a non-prod environment before production. --- bot/code_review_bot/workflow.py | 174 ++++++++++++++++++++++++++++ bot/tests/conftest.py | 20 +++- bot/tests/test_workflow.py | 195 +++++++++++++++++++++++++++++++- 3 files changed, 387 insertions(+), 2 deletions(-) diff --git a/bot/code_review_bot/workflow.py b/bot/code_review_bot/workflow.py index bcdc25537..b35ac7a20 100644 --- a/bot/code_review_bot/workflow.py +++ b/bot/code_review_bot/workflow.py @@ -2,6 +2,9 @@ # 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 base64 +import json +import re import time from datetime import datetime, timedelta from itertools import groupby @@ -45,6 +48,13 @@ TASKCLUSTER_NAMESPACE = "project.relman.{channel}.code-review.{name}" TASKCLUSTER_INDEX_TTL = 7 # in days +DECISION_TASK_ROUTE = "gecko.v2.{repo}.revision.{revision}.taskgraph.decision" +PUBLICATION_LOG_ARTIFACT = "public/logs/live.log" +TREEHERDER_LINK_REGEX = re.compile( + rb"treeherder\.mozilla\.org/(?:#/)?jobs\?repo=(?P[\w-]+)" + rb"&revision=(?P[0-9a-f]{12,40})" +) + class Workflow: """ @@ -347,6 +357,11 @@ def start_analysis(self, revision): worker = MercurialWorker() output = worker.run(repository, build) + # Cancel any in-progress tasks from an earlier update + # This is done after pushing to try to avoid delaying runs of the + # new tasks. + self.cancel_previous(revision) + # Update index when the patch has been pushed to try self.index(revision, state="pushed_to_try") @@ -465,6 +480,165 @@ def publish(self, revision, issues, task_failures, notices, reviewers): else BuildState.Pass, ) + def cancel_previous(self, revision): + """Cancel the try pushes triggered by earlier updates of a revision""" + + # In order to cancel tasks from a previoush push we need the task group id + # of the previous push (which is the same as the decision task for that push). + # The only way to retrieve this through the Phabricator API is by a long series + # of requests: + # * Find all of the prior Build Targets (via builds, via buildables) + # * Find the log for each Build Target, which will contain a taskId of Code Review task + # * Pull the Code Review task log to fetch the treeherder link with the revision pushed to Try in it + # * Look up the decision task id in the task index via the revision + # + # (Despite the fact that the Phabricator UI shows the treeherder link in it, this is + # not available through the API, so we have to take the long way to get here.) + try: + publication_tasks = self.list_previous_publication_tasks(revision) + logger.info("Found previous publication tasks", tasks=publication_tasks) + except Exception as e: + logger.warn( + "Failed to list previous publication tasks", + rev=str(revision), + error=str(e), + ) + return + + for task_id in publication_tasks: + task_group_id = self.find_try_decision_task(task_id) + if task_group_id is None: + continue + + try: + self.queue_service.cancelTaskGroup(task_group_id) + except Exception as e: + logger.warn( + "Failed to cancel a previous try push", + task_group_id=task_group_id, + error=str(e), + ) + continue + + logger.info("Cancelled a previous try push", task_group_id=task_group_id) + + def list_previous_publication_tasks(self, revision): + logger.debug( + "Finding previous publication tasks", phid=revision.phabricator_phid + ) + buildables = self.phabricator.request( + "harbormaster.buildable.search", + constraints={"containerPHIDs": [revision.phabricator_phid]}, + )["data"] + if not buildables: + logger.debug("No buildables found", phid=revision.phabricator_phid) + return [] + + buildable_phids = [b["phid"] for b in buildables] + + logger.debug( + "Found buildables", + phid=revision.phabricator_phid, + buildables=buildable_phids, + ) + builds = self.phabricator.request( + "harbormaster.build.search", + constraints={"buildables": buildable_phids}, + )["data"] + if not builds: + logger.debug("No builds found", buildables=buildable_phids) + return [] + + build_phids = [b["phid"] for b in builds] + + logger.debug("Found builds", buildables=buildable_phids, builds=build_phids) + targets = [ + target + for target in self.phabricator.request( + "harbormaster.target.search", + constraints={"buildPHIDs": build_phids}, + )["data"] + # Ignore the current Build Target; there will be no code review + # task for it anyways, and we wouldn't want to cancel it even if + # there was one. + if target["phid"] != revision.build_target_phid + ] + if not targets: + return [] + + build_target_phids = [target["phid"] for target in targets] + + logger.debug( + "Found build targets", builds=build_phids, build_targets=build_target_phids + ) + + logs = self.phabricator.request( + "harbormaster.log.search", + constraints={"buildTargetPHIDs": build_target_phids}, + )["data"] + + task_ids = [] + for log in logs: + phid = log["fields"]["filePHID"] + logger.debug("Downloading log", file=phid) + blob = self.phabricator.request("file.download", phid=phid) + try: + payload = json.loads(base64.b64decode(blob).decode("utf-8", "replace")) + except (TypeError, ValueError): + logger.debug("Couldn't parse log", file=phid) + continue + + if not isinstance(payload, dict): + logger.debug("Log is not an object", file=phid) + continue + + task_id = payload.get("taskId") + if not task_id: + logger.debug("Couldn't find task id", file=phid) + continue + + if task_id not in task_ids: + task_ids.append(task_id) + + return task_ids + + def find_try_decision_task(self, publication_task_id): + """ + Find the decision task of the try push made by a publication task + + The Treeherder link it published is only available in its own live log. + """ + url = self.queue_service.buildUrl( + "getLatestArtifact", publication_task_id, PUBLICATION_LOG_ARTIFACT + ) + # Allows HTTP_30x redirections retrieving the artifact + response = self.queue_service.session.get( + url, stream=True, allow_redirects=True + ) + if not response.ok: + logger.warn( + "Failed to read the log of a publication task", + task=publication_task_id, + error=response.status_code, + ) + return + + match = TREEHERDER_LINK_REGEX.search(response.content) + if match is None: + logger.info( + "No try push found for a publication task", task=publication_task_id + ) + return + + route = DECISION_TASK_ROUTE.format( + repo=match.group("repo").decode("utf-8"), + revision=match.group("revision").decode("utf-8"), + ) + try: + return self.index_service.findTask(route)["taskId"] + except Exception as e: + logger.warn("Failed to find a decision task", route=route, error=str(e)) + def index(self, revision, **kwargs): """ Index current task on Taskcluster index diff --git a/bot/tests/conftest.py b/bot/tests/conftest.py index da6d97580..f20ea5879 100644 --- a/bot/tests/conftest.py +++ b/bot/tests/conftest.py @@ -468,6 +468,14 @@ def raise_for_status(self): else: return json.dumps(self.body) + @property + def ok(self): + return self.code < 300 + + @property + def status_code(self): + return self.code + @property def content(self): return self.body.encode() @@ -508,6 +516,7 @@ class MockQueue: def __init__(self): self._artifacts = {} self.session = SessionMock() + self.cancelled_groups = [] def configure(self, relations): # Reset the session mock @@ -584,10 +593,19 @@ def listArtifacts(self, task_id, run_id): def listLatestArtifacts(self, task_id): return self._artifacts.get(task_id, {}) - def buildUrl(self, route_name, task, run, name): + def buildUrl(self, route_name, *args): + if route_name == "getLatestArtifact": + task, name = args + return f"http://tc.test/{task}/artifacts/{name}" + assert route_name == "getArtifact" + task, run, name = args return f"http://tc.test/{task}/{run}/artifacts/{name}" + def cancelTaskGroup(self, group_id): + self.cancelled_groups.append(group_id) + return {"taskGroupId": group_id, "taskIds": []} + def createArtifact(self, task_id, run_id, name, payload): if task_id not in self._artifacts: self._artifacts[task_id] = {"artifacts": []} diff --git a/bot/tests/test_workflow.py b/bot/tests/test_workflow.py index 51d8a699f..8725fb906 100644 --- a/bot/tests/test_workflow.py +++ b/bot/tests/test_workflow.py @@ -2,6 +2,7 @@ # 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 base64 import os from datetime import datetime from unittest import mock @@ -11,7 +12,7 @@ import responses from libmozdata.phabricator import ConduitError -from code_review_bot.config import Settings +from code_review_bot.config import Settings, TaskCluster from code_review_bot.revisions import PhabricatorRevision from code_review_bot.tasks.clang_format import ClangFormatIssue, ClangFormatTask from code_review_bot.tasks.clang_tidy import ClangTidyTask @@ -322,3 +323,195 @@ def test_publish_link(mock_phabricator, mock_workflow): unquote_plus(call.request.body) == 'params={"buildTargetPHID": "PHID-HMBT-test", "artifactType": "uri", "artifactKey": "some-unique-code", "artifactData": {"uri": "http://taskcluster/x.y.z", "name": "A nice display name", "ui.external": true}, "__conduit__": {"token": "deadbeef"}}&output=json' ) + + +TREEHERDER_LOG = """ +2026-08-21 05:51:27 [INFO] Mercurial stdout=b'remote: Follow the progress of your build on Treeherder:\\n' +2026-08-21 05:51:27 [INFO] Mercurial stdout=b'remote: https://treeherder.mozilla.org/jobs?repo=try&revision=02af25daadb64b42939b5a3f382c6d5f2a6f311e\\n' +2026-08-21 05:51:41 [INFO] Created HarborMaster link on PHID-HMBT-old : https://treeherder.mozilla.org/#/jobs?repo=try&revision=02af25daadb64b42939b5a3f382c6d5f2a6f311e +""" + +DECISION_ROUTE = ( + "gecko.v2.try.revision.02af25daadb64b42939b5a3f382c6d5f2a6f311e.taskgraph.decision" +) + + +def mock_harbormaster(logs): + """ + Mock the Conduit calls used to list the publication tasks of previous updates + + logs maps a file PHID to a (build target PHID, raw log content) tuple. + """ + api = mock.MagicMock() + + def request(path, **payload): + if path == "harbormaster.buildable.search": + return {"data": [{"phid": "PHID-HMBB-1"}]} + + if path == "harbormaster.build.search": + return {"data": [{"phid": "PHID-HMBD-1"}]} + + if path == "harbormaster.target.search": + targets = {target for target, _ in logs.values()} + return {"data": [{"phid": phid} for phid in sorted(targets)]} + + if path == "harbormaster.log.search": + wanted = payload["constraints"]["buildTargetPHIDs"] + return { + "data": [ + {"fields": {"filePHID": phid}} + for phid, (target, _) in sorted(logs.items()) + if target in wanted + ] + } + + if path == "file.download": + content = logs[payload["phid"]][1] + return base64.b64encode(content.encode("utf-8")).decode("utf-8") + + raise AssertionError(f"Unexpected conduit call {path}") + + api.request.side_effect = request + return api + + +def test_list_previous_publication_tasks(mock_config, mock_workflow): + """ + Publication task ids are read out of the Harbormaster logs of every build + target but the one currently running + """ + mock_config.taskcluster = TaskCluster("/tmp/dummy", "currentTask", 0, False) + mock_workflow.phabricator = mock_harbormaster( + { + "PHID-FILE-old-headers": ( + "PHID-HMBT-old", + "HTTP 200\nContent-Length: 98\n", + ), + "PHID-FILE-old-body": ("PHID-HMBT-old", '{"taskId": "oldPublicationTask"}'), + "PHID-FILE-current-body": ( + "PHID-HMBT-current", + '{"taskId": "currentTask"}', + ), + } + ) + + revision = mock.MagicMock(spec=PhabricatorRevision) + revision.phabricator_phid = "PHID-DREV-1" + revision.build_target_phid = "PHID-HMBT-current" + + assert mock_workflow.list_previous_publication_tasks(revision) == [ + "oldPublicationTask" + ] + + +def test_find_try_decision_task(mock_config, mock_workflow): + """ + The try changeset is read from the publication task log, then resolved to a + decision task through the Taskcluster index + """ + mock_workflow.queue_service.session.add( + "get", + "http://tc.test/oldPublicationTask/artifacts/public/logs/live.log", + TREEHERDER_LOG, + ) + mock_workflow.index_service.configure( + {"decisionTask": {"route": DECISION_ROUTE}}, + ) + + assert mock_workflow.find_try_decision_task("oldPublicationTask") == "decisionTask" + + +def test_find_try_decision_task_without_try_push(mock_config, mock_workflow): + """ + A build that never reached the try push stage is simply skipped + """ + mock_workflow.queue_service.session.add( + "get", + "http://tc.test/oldPublicationTask/artifacts/public/logs/live.log", + "Nothing was pushed to try", + ) + mock_workflow.index_service.configure({}) + + assert mock_workflow.find_try_decision_task("oldPublicationTask") is None + + +def test_find_try_decision_task_missing_log(mock_config, mock_workflow): + """ + An expired or missing publication log does not raise + """ + mock_workflow.index_service.configure({}) + + assert mock_workflow.find_try_decision_task("oldPublicationTask") is None + + +def test_cancel_previous(mock_config, mock_workflow): + """ + Task groups of previous try pushes are cancelled + """ + mock_config.taskcluster = TaskCluster("/tmp/dummy", "currentTask", 0, False) + mock_workflow.phabricator = mock_harbormaster( + { + "PHID-FILE-old-headers": ("PHID-HMBT-old", "HTTP 200\n"), + "PHID-FILE-old-body": ("PHID-HMBT-old", '{"taskId": "oldPublicationTask"}'), + "PHID-FILE-current-body": ( + "PHID-HMBT-current", + '{"taskId": "currentTask"}', + ), + } + ) + mock_workflow.queue_service.session.add( + "get", + "http://tc.test/oldPublicationTask/artifacts/public/logs/live.log", + TREEHERDER_LOG, + ) + mock_workflow.index_service.configure({"decisionTask": {"route": DECISION_ROUTE}}) + + revision = mock.MagicMock(spec=PhabricatorRevision) + revision.phabricator_phid = "PHID-DREV-1" + revision.build_target_phid = "PHID-HMBT-current" + + mock_workflow.cancel_previous(revision) + + assert mock_workflow.queue_service.cancelled_groups == ["decisionTask"] + + +def test_cancel_previous_without_previous_build(mock_config, mock_workflow): + """ + A revision whose only build target is the current one has nothing to cancel + """ + mock_config.taskcluster = TaskCluster("/tmp/dummy", "currentTask", 0, False) + mock_workflow.phabricator = mock_harbormaster( + { + "PHID-FILE-current-body": ( + "PHID-HMBT-current", + '{"taskId": "currentTask"}', + ), + } + ) + mock_workflow.index_service.configure({}) + + revision = mock.MagicMock(spec=PhabricatorRevision) + revision.phabricator_phid = "PHID-DREV-1" + revision.build_target_phid = "PHID-HMBT-current" + + mock_workflow.cancel_previous(revision) + + assert mock_workflow.queue_service.cancelled_groups == [] + + +def test_cancel_previous_conduit_failure(mock_config, mock_workflow): + """ + A Conduit failure while listing previous builds is swallowed + """ + mock_config.taskcluster = TaskCluster("/tmp/dummy", "currentTask", 0, False) + mock_workflow.phabricator = mock.MagicMock() + mock_workflow.phabricator.request.side_effect = ConduitError("Boom") + + revision = mock.MagicMock(spec=PhabricatorRevision) + revision.phabricator_phid = "PHID-DREV-1" + revision.build_target_phid = "PHID-HMBT-current" + + mock_workflow.cancel_previous(revision) + + assert mock_workflow.queue_service.cancelled_groups == [] + mock_workflow.phabricator.request.assert_called_once()