2026 11 10 tjr commandeered task

apis/phabricator.py

@@ -24,6 +24,12 @@ class DiffTooLargeError(Exception):
     pass
 
 
+class ConduitError(Exception):
+    def __init__(self, method, result):
+        super().__init__("Got an error from phabricator calling %s: %s" % (method, result))
+        self.error_message = result.get('errorMessage') or ""
+
+
 class PhabricatorProvider(BaseProvider, INeedsCommandProvider, INeedsLoggingProvider):
     def __init__(self, config):
         if 'url' not in config:
@@ -36,6 +42,40 @@ def __init__(self, config):
         else:
             self.url = config['url']
 
+    def _call_conduit(self, method, params):
+        cmd = "echo " + quote_echo_string(json.dumps(params))
+        cmd += " | arc call-conduit --conduit-uri=%s %s --" % (self.url, method)
+        ret = self.run(cmd, shell=True)
+        try:
+            result = json.loads(ret.stdout.decode())
+        except Exception:
+            raise Exception("Could not decode response as JSON: %s" % ret.stdout.decode())
+        if result['error']:
+            raise ConduitError(method, result)
+        return result['response']
+
+    def _search_one(self, method, constraints, what):
+        data = self._call_conduit(method, {"constraints": constraints})['data']
+        if len(data) != 1:
+            raise Exception("When querying conduit for %s, we got back %i results" % (what, len(data)))
+        return data[0]
+
+    def _edit_revision(self, phab_revision, transactions):
+        return self._call_conduit("differential.revision.edit", {"transactions": transactions, "objectIdentifier": str(phab_revision)})
+
+    # Only called from is_commandeered, which retries
+    def _own_phid(self):
+        if not getattr(self, '_phid', None):
+            self._phid = self._call_conduit("user.whoami", {})['phid']
+        return self._phid
+
+    @logEntryExit
+    @retry
+    def is_commandeered(self, phab_revision):
+        """Whether someone else has made themselves the author of one of our revisions."""
+        revision = self._search_one("differential.revision.search", {"ids": [int(phab_revision)]}, "revision %s" % phab_revision)
+        return revision['fields']['authorPHID'] != self._own_phid()
+
     @logEntryExit
     def submit_patches(self, bug_id, num_commits):
         phab_revisions = []
@@ -95,37 +135,8 @@ def submit_to_phabricator(rev_id, retry_attempt=None):
         # Chain revisions together if needed
         @retry
         def chain_revisions(parent_rev, child_rev):
-            # First get the PHID of the parent revision
-            cmd = "echo " + quote_echo_string("""{"constraints": {"ids":[%s]}}""" % parent_rev)
-            cmd += " | arc call-conduit --conduit-uri=%s differential.revision.search --""" % self.url
-
-            ret = self.run(cmd, shell=True)
-            try:
-                result = json.loads(ret.stdout.decode())
-            except Exception:
-                raise Exception("Could not decode response as JSON: %s" % ret.stdout.decode())
-
-            if result['error']:
-                raise Exception("Got an error from phabricator when trying to search for %s" % (parent_rev))
-
-            assert 'response' in result
-            assert 'data' in result['response']
-            if len(result['response']['data']) != 1:
-                raise Exception("When querying conduit for diff %s, we got back %i results"
-                                % (parent_rev, len(result['response']['data'])))
-
-            parent_phid = result['response']['data'][0]['phid']
-
-            # Now connect them - add parent as parent of child
-            cmd = "echo " + quote_echo_string("""{"transactions": [{"type":"parents.add", "value":["%s"]}], "objectIdentifier": "%s"}""" % (parent_phid, child_rev))
-            cmd += " | arc call-conduit --conduit-uri=%s differential.revision.edit --""" % self.url
-            ret = self.run(cmd, shell=True)
-            try:
-                result = json.loads(ret.stdout.decode())
-            except Exception:
-                raise Exception("Could not decode response as JSON: %s" % ret.stdout.decode())
-            if result['error']:
-                raise Exception("Got an error from phabricator when trying chain revisions, parent: %s, child %s %s" % (parent_rev, child_rev, parent_phid))
+            parent_phid = self._search_one("differential.revision.search", {"ids": [int(parent_rev)]}, "revision %s" % parent_rev)['phid']
+            self._edit_revision(child_rev, [{"type": "parents.add", "value": [parent_phid]}])
 
         parent_rev = phab_revisions[0]
         for child_rev in phab_revisions[1:]:
@@ -135,15 +146,7 @@ def chain_revisions(parent_rev, child_rev):
         # Associate the patches with the bug
         @retry
         def associate_bug_id(phab_revision):
-            cmd = "echo " + quote_echo_string("""{"transactions": [{"type":"bugzilla.bug-id", "value":"%s"}], "objectIdentifier": "%s"}""" % (bug_id, phab_revision))
-            cmd += " | arc call-conduit --conduit-uri=%s differential.revision.edit --""" % self.url
-            ret = self.run(cmd, shell=True)
-            try:
-                result = json.loads(ret.stdout.decode())
-            except Exception:
-                raise Exception("Could not decode response as JSON: %s" % ret.stdout.decode())
-            if result['error']:
-                raise Exception("Got an error from phabricator when trying to set the bugzilla id for %s" % (phab_revision))
+            self._edit_revision(phab_revision, [{"type": "bugzilla.bug-id", "value": str(bug_id)}])
 
         for p in phab_revisions:
             associate_bug_id(p)
@@ -156,59 +159,25 @@ def associate_bug_id(phab_revision):
     @logEntryExit
     @retry
     def set_reviewer(self, phab_revision, phab_username):
-        # We have to call a different API endpoint if this is a review group
+        # A review group is a project, with its own search endpoint
         if phab_username[0] == "#":
-            # Get the group's phid (groups are implemented as 'projects'')
-            cmd = "echo " + quote_echo_string("""{"constraints": {"slugs":["%s"]}}""" % phab_username)
-            cmd += " | arc call-conduit --conduit-uri=%s project.search --""" % self.url
+            phid = self._search_one("project.search", {"slugs": [phab_username]}, "project %s" % phab_username)['phid']
         else:
-            # Get the user's phid
-            cmd = "echo " + quote_echo_string("""{"constraints": {"usernames":["%s"]}}""" % phab_username)
-            cmd += " | arc call-conduit --conduit-uri=%s user.search --""" % self.url
-
-        ret = self.run(cmd, shell=True)
-        try:
-            result = json.loads(ret.stdout.decode())
-        except Exception:
-            raise Exception("Could not decode response as JSON: %s" % ret.stdout.decode())
-
-        if result['error']:
-            raise Exception("Got an error from phabricator when trying to search for %s" % (phab_username))
-
-        assert 'response' in result
-        assert 'data' in result['response']
-        if len(result['response']['data']) != 1:
-            raise Exception("When querying conduit for username %s, we got back %i results"
-                            % (phab_username, len(result['response']['data'])))
-
-        phid = result['response']['data'][0]['phid']
+            phid = self._search_one("user.search", {"usernames": [phab_username]}, "username %s" % phab_username)['phid']
 
         # reviewers.add, not reviewers.set: anyone already reviewing should stay a reviewer
-        cmd = "echo " + quote_echo_string("""{"transactions": [{"type":"reviewers.add", "value":["%s"]}], "objectIdentifier": "%s"}""" % (phid, phab_revision))
-        cmd += " | arc call-conduit --conduit-uri=%s differential.revision.edit --""" % self.url
-        ret = self.run(cmd, shell=True)
-        try:
-            result = json.loads(ret.stdout.decode())
-        except Exception:
-            raise Exception("Could not decode response as JSON: %s" % ret.stdout.decode())
-        if result['error']:
-            raise Exception("Got an error from phabricator when trying to add %s (%s) as a reviewer of %s: %s" % (phab_username, phid, phab_revision, result))
+        self._edit_revision(phab_revision, [{"type": "reviewers.add", "value": [phid]}])
 
     @logEntryExit
     @retry
     def abandon(self, phab_revision):
-        cmd = "echo " + quote_echo_string("""{"transactions": [{"type":"abandon", "value":true}],"objectIdentifier": "%s"}""" % phab_revision)
-        cmd += " | arc call-conduit --conduit-uri=%s differential.revision.edit --""" % self.url
-        ret = self.run(cmd, shell=True)
         try:
-            result = json.loads(ret.stdout.decode())
-        except Exception:
-            raise Exception("Could not decode response as JSON: %s" % ret.stdout.decode())
-        if result['error']:
-            if "You can not abandon this revision because it has already been closed." in result['errorMessage']:
+            self._edit_revision(phab_revision, [{"type": "abandon", "value": True}])
+        except ConduitError as e:
+            if "You can not abandon this revision because it has already been closed." in e.error_message:
                 self.logger.log("Strangely, the phabricator revision %s was already closed when we tried to abandon it. Oh well." % phab_revision, level=LogLevel.Warning)
-            elif "You can not abandon this revision because you are not the author." in result['errorMessage']:
+            elif "You can not abandon this revision because you are not the author." in e.error_message:
                 # Someone commandeered the revision, so it's theirs to keep or abandon
                 self.logger.log("The phabricator revision %s has been commandeered by someone else, so we're leaving it alone instead of abandoning it." % phab_revision, level=LogLevel.Info)
             else:
-                raise Exception("Got an error from phabricator when trying to abandon %s: %s" % (phab_revision, result))
+                raise

components/bugzilla.py

@@ -264,6 +264,14 @@ def COMMENT_REJECTED(log_url):
     def TRY_RUN_SUBMITTED(revision):
         return "I've submitted a try run for this commit: https://treeherder.mozilla.org/jobs?repo=try&revision=" + revision
 
+    @staticmethod
+    def BUG_SUPERSEDED_BUT_COMMANDEERED(new_bug_id):
+        return """
+A newer version of this library is in Bug %s. Because someone has taken over the
+Phabricator revision here, I'm leaving this bug open instead of closing it as a
+duplicate of that one.
+""" % new_bug_id
+
     @staticmethod
     def BUG_SUPERSEDED():
         return """

components/db.py

@@ -10,13 +10,14 @@
 from components.logging import LogLevel
 from components.dbmodels import TryRun, PhabRevision, transform_job_and_try_results_into_objects, JOBSTATUS, JOBOUTCOME, JOBTYPE
 
+import os
 import pymysql
 
 
 # ==================================================================================
 
 
-CURRENT_DATABASE_CONFIG_VERSION = 16
+CURRENT_DATABASE_CONFIG_VERSION = 17
 
 CREATION_QUERIES = {
     "config": """
@@ -58,6 +59,7 @@
         `outcome` TINYINT NOT NULL,
         `relinquished` TINYINT NOT NULL,
         `bugzilla_id` INT NULL,
+        `task_id` VARCHAR(32) NULL,
         PRIMARY KEY (`id`)
       ) ENGINE = InnoDB;
       """,
@@ -439,6 +441,11 @@ def any_in(needles, haystack):
 
                         self._query_execute("ALTER TABLE jobs DROP phab_revision")
 
+                    if config_version <= 16 and CURRENT_DATABASE_CONFIG_VERSION >= 17:
+                        self.logger.log("Upgrading to database version 17", level=LogLevel.Warning)
+                        # The Taskcluster task of the Updatebot run that created the job
+                        self._query_execute("ALTER TABLE `jobs` ADD COLUMN `task_id` VARCHAR(32) NULL AFTER `bugzilla_id`")
+
                     query = "UPDATE config SET v=%s WHERE k = 'database_version'"
                     args = (CURRENT_DATABASE_CONFIG_VERSION)
                     self._query_execute(query, args)
@@ -573,8 +580,8 @@ def get_job(self, library, new_version):
     @logEntryExit
     def create_job(self, jobtype, library, new_version, ff_version, status, outcome, bug_id):
         # Omitting the created column initializes it to current timestamp
-        query = "INSERT INTO jobs(job_type, library, version, status, outcome, relinquished, bugzilla_id) VALUES(%s, %s, %s, %s, %s, 0, %s)"
-        args = (jobtype, library.name, new_version, status, outcome, bug_id)
+        query = "INSERT INTO jobs(job_type, library, version, status, outcome, relinquished, bugzilla_id, task_id) VALUES(%s, %s, %s, %s, %s, 0, %s, %s)"
+        args = (jobtype, library.name, new_version, status, outcome, bug_id, os.environ.get('TASK_ID'))
         job_id = self._query_execute(query, args)
 
         query = "INSERT INTO job_to_ff_version(job_id, ff_version) VALUES(%s, %s)"

components/dbmodels.py

@@ -120,6 +120,7 @@ def __init__(self, row=None):
             self.outcome = JOBOUTCOME(row['outcome'])
             self.relinquished = not not row['relinquished']
             self.bugzilla_id = row['bugzilla_id']
+            self.task_id = row.get('task_id')
             self.ff_versions = set()
             self.phab_revisions = []
             self.try_runs = []

tasktypes/vendoring.py

@@ -140,6 +140,10 @@ def _process_prior_job(self, prior_job, superseceding_bugzilla_id):
 
         if not prior_job.bugzilla_is_open or self.bugzillaProvider.bug_has_landing_link(prior_job.bugzilla_id):
             self.logger.log("The prior job's bugzilla bug is closed or about to be closed, so we only need to relinquish it.", level=LogLevel.Info)
+        elif superseceding_bugzilla_id and self._has_commandeered_revision(prior_job):
+            # Someone is working on the revision themselves, so it isn't ours to close
+            self.logger.log("The prior job's phabricator revision has been commandeered, so we're leaving its bug open.", level=LogLevel.Info)
+            self.bugzillaProvider.comment_on_bug(prior_job.bugzilla_id, CommentTemplates.BUG_SUPERSEDED_BUT_COMMANDEERED(superseceding_bugzilla_id))
         elif superseceding_bugzilla_id:
             self.logger.log("The prior job's bugzilla bug is open, marking it as superseded by Bug ID %s ." % superseceding_bugzilla_id, level=LogLevel.Info)
 
@@ -154,6 +158,16 @@ def _process_prior_job(self, prior_job, superseceding_bugzilla_id):
 
         self.dbProvider.update_job_relinquish(prior_job)
 
+    def _has_commandeered_revision(self, job):
+        for p in job.phab_revisions:
+            try:
+                if self.phabricatorProvider.is_commandeered(p.revision):
+                    return True
+            except Exception as e:
+                # If we can't tell, close the bug as we always have
+                self.logger.log_exception(e)
+        return False
+
     @logEntryExit
     def _process_new_job(self, library, task, new_version, timestamp, most_recent_job):
         if not self._should_process_new_job(library, task, new_version):

tests/database.py

@@ -4,6 +4,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 os
 import sys
 import time
 import copy
@@ -68,8 +69,9 @@ def testJobs(self):
         try:
             self.assertEqual(None, self.db.get_job(library, version))
 
-            self.db.create_job(JOBTYPE.VENDORING, library, version,
-                               JOBSTATUS.AWAITING_SECOND_PLATFORMS_TRY_RESULTS, JOBOUTCOME.PENDING, bugid)
+            with mock.patch.dict(os.environ, {'TASK_ID': 'abc123'}):
+                self.db.create_job(JOBTYPE.VENDORING, library, version,
+                                   JOBSTATUS.AWAITING_SECOND_PLATFORMS_TRY_RESULTS, JOBOUTCOME.PENDING, bugid)
 
             newJob = self.db.get_job(library, version)
             self.assertNotEqual(None, newJob)
@@ -79,9 +81,39 @@ def testJobs(self):
             self.assertEqual(newJob.status, JOBSTATUS.AWAITING_SECOND_PLATFORMS_TRY_RESULTS)
             self.assertEqual(newJob.outcome, JOBOUTCOME.PENDING)
             self.assertEqual(newJob.bugzilla_id, bugid)
+            self.assertEqual(newJob.task_id, 'abc123')
         finally:
             self.db.delete_job(job_id=newJob.id)
 
+    def testJobWithoutTaskId(self):
+        # Outside of Taskcluster there's no task to record
+        library = Struct(**{
+            'name': 'test_library_local',
+            'yaml_path': 'path/to/moz.yaml',
+        })
+        version = "test_local_version"
+        with mock.patch.dict(os.environ):
+            os.environ.pop('TASK_ID', None)
+            self.db.create_job(JOBTYPE.VENDORING, library, version,
+                               JOBSTATUS.AWAITING_SECOND_PLATFORMS_TRY_RESULTS, JOBOUTCOME.PENDING, 50)
+        job = self.db.get_job(library, version)
+        try:
+            self.assertIsNone(job.task_id)
+        finally:
+            self.db.delete_job(job_id=job.id)
+
+    def testUpgradeAddsTaskId(self):
+        # Put the database back to version 16, before jobs had a task_id
+        mysql = self.db.db
+        mysql._query_execute("ALTER TABLE jobs DROP COLUMN task_id")
+        mysql._query_execute("UPDATE config SET v=16 WHERE k = 'database_version'")
+
+        self.assertEqual(self.db.check_database(), 17)
+        columns = [r['COLUMN_NAME'] for r in mysql._query_get_rows(
+            "SELECT COLUMN_NAME FROM information_schema.columns WHERE table_schema = %s AND table_name = 'jobs'",
+            (mysql.database_config['db'],))]
+        self.assertIn('task_id', columns)
+
 
 class TestDatabaseConnection(unittest.TestCase):
     def testRetriesWhileTheServerIsNotUpYet(self):