sync/downstream.py

# This Source Code Form is subject to the terms of the Mozilla Public
# 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/.

"""Functionality to support VCS syncing for WPT."""


from __future__ import annotations
import os
import re
import subprocess
import traceback
from collections import defaultdict
from datetime import datetime


import enum
import git
import newrelic

from . import bugcomponents
from . import gitutils
from . import log
from . import notify
from . import trypush
from . import commit as sync_commit
from .base import FrozenDict, entry_point
from .commit import GeckoCommit, WptCommit
from .env import Environment
from .errors import AbortError
from .gitutils import update_repositories
from .lock import SyncLock, mut, constructor
from .projectutil import Mach, WPT
from .sync import LandableStatus, SyncProcess
from .trypush import TryPush

from git.objects.tree import Tree
from git.repo.base import Repo
from typing import Any, List, Mapping, MutableMapping, cast, TYPE_CHECKING

logger = log.get_logger(__name__)
env = Environment()


@enum.unique
class DownstreamAction(enum.Enum):
    ready = 0
    manual_fix = 1
    try_push = 2
    try_push_stability = 3
    wait_try = 4
    wait_upstream = 5
    try_rebase = 6

    def reason_str(self):
        # type () -> Text
        return {DownstreamAction.ready: "",
                DownstreamAction.manual_fix: "",
                DownstreamAction.try_push: "valid try push required",
                DownstreamAction.try_push_stability: "stability try push required",
                DownstreamAction.wait_try: "waiting for try to complete",
                DownstreamAction.wait_upstream: "waiting for PR to be merged",
                DownstreamAction.try_rebase: "Need to rebase try push"}.get(self, "")


class DownstreamSync(SyncProcess):
    sync_type = "downstream"
    obj_id = "pr"
    statuses = ("open", "complete")
    status_transitions = [("open", "complete"),
                          ("complete", "open")]  # Unfortunately, if a backout occurs

    @classmethod
    @constructor(lambda args: ("downstream", str(args["pr_id"])))
    def new(cls,
            lock: SyncLock,
            git_gecko: Repo,
            git_wpt: Repo,
            wpt_base: str,
            pr_id: int,
            pr_title: str,
            pr_body: str,
            ) -> DownstreamSync:
        # TODO: add PR link to the comment
        sync = super().new(lock,
                           git_gecko,
                           git_wpt,
                           pr=pr_id,
                           gecko_base=DownstreamSync.gecko_landing_branch(),
                           gecko_head=DownstreamSync.gecko_landing_branch(),
                           wpt_base=wpt_base,
                           wpt_head="origin/pr/%s" % pr_id)

        with sync.as_mut(lock):
            sync.create_bug(git_wpt, pr_id, pr_title, pr_body)
        return sync

    def make_bug_comment(self, git_wpt: Repo, pr_id: int, pr_title: str,
                         pr_body: str | None) -> str:
        pr_msg = env.gh_wpt.cleanup_pr_body(pr_body)
        # TODO: Ensure we have the right set of commits before geting here
        author = self.wpt_commits[0].author if self.wpt_commits else b""

        msg = ["Sync web-platform-tests PR %s into mozilla-central"
               " (this bug is closed when the sync is complete)." % pr_id,
               "",
               "PR: %s" % env.gh_wpt.pr_url(pr_id),
               ""
               "Details from upstream follow.",
               "",
               "%s wrote:" % author.decode("utf8", "ignore"),
               ">  %s" % pr_title]
        if pr_msg:
            msg.append(">  ")
            msg.extend(">  %s" % line for line in pr_msg.split("\n"))
        return "\n".join(msg)

    @classmethod
    def has_metadata(cls, message: bytes) -> bool:
        required_keys = ["wpt-commits",
                         "wpt-pr"]
        metadata = sync_commit.get_metadata(message)
        return all(item in metadata for item in required_keys)

    @property
    def landable_status(self) -> LandableStatus:
        if self.skip:
            return LandableStatus.skip
        if self.metadata_ready:
            return LandableStatus.ready
        if self.error:
            return LandableStatus.error

        return LandableStatus.missing_try_results

    @property
    def pr_head(self) -> WptCommit:
        """Head commit of the PR. Typically this is equal to
        self.wpt_commits.head but if the PR is rebased onto master
        then the head of the PR won't match the commit that actually
        merged unless it happens to be a fast-forward"""
        return sync_commit.WptCommit(self.git_wpt, "origin/pr/%s" % self.pr)

    @SyncProcess.error.setter  # type: ignore
    @mut()
    def error(self, value: str | None) -> Any | None:
        if self.pr:
            if value is not None:
                env.gh_wpt.add_labels(self.pr, "mozilla:gecko-blocked")
            elif self.error is not None:
                env.gh_wpt.remove_labels(self.pr, "mozilla:gecko-blocked")
        return SyncProcess.error.fset(self, value)  # type: ignore

    @property
    def pr_status(self):
        return self.data.get("pr-status", "open")

    @pr_status.setter  # type: ignore
    @mut()
    def pr_status(self, value):
        self.data["pr-status"] = value

    @property
    def notify_bugs(self) -> FrozenDict:
        return FrozenDict(**self.data.get("notify-bugs", {}))

    @notify_bugs.setter  # type: ignore
    @mut()
    def notify_bugs(self, value: FrozenDict) -> None:
        self.data["notify-bugs"] = value.as_dict()

    @property
    def next_action(self) -> DownstreamAction:
        """Work out the next action for the sync based on the current status.

        Returns a DownstreamAction indicating the next step to take."""

        if self.data.get("force-metadata-ready"):
            # This is mostly for testing
            return DownstreamAction.ready
        if self.skip:
            return DownstreamAction.ready
        if self.error:
            if self.tried_to_rebase is False:
                return DownstreamAction.try_rebase
            return DownstreamAction.manual_fix

        latest_try_push = self.latest_valid_try_push
        if (latest_try_push and not latest_try_push.taskgroup_id):
            if latest_try_push.status == "open":
                return DownstreamAction.wait_try
            elif latest_try_push.infra_fail:
                if self.tried_to_rebase is False:
                    return DownstreamAction.try_rebase
                return DownstreamAction.manual_fix

        assert self.pr is not None
        pr = env.gh_wpt.get_pull(self.pr)
        if pr.merged:  # Wait till PR is merged to do anything
            if not latest_try_push:
                if self.requires_stability_try:
                    logger.debug("Sync for PR %s requires a stability try push" % self.pr)
                    return DownstreamAction.try_push_stability
                elif self.requires_try:
                    return DownstreamAction.try_push
                else:
                    return DownstreamAction.ready

            if latest_try_push.status != "complete":
                return DownstreamAction.wait_try

            if self.requires_stability_try and not latest_try_push.stability:
                return DownstreamAction.try_push_stability

            # If we have infra failure, flag for human intervention. Retrying stability
            # runs would be very costly
            if latest_try_push.infra_fail:
                tasks = latest_try_push.tasks()
                if tasks is None:
                    return DownstreamAction.manual_fix

                # Check if we had any successful tests
                if tasks.has_completed_tests():
                    return DownstreamAction.ready
                else:
                    return DownstreamAction.manual_fix

            return DownstreamAction.ready
        else:
            return DownstreamAction.wait_upstream

    @property
    def metadata_ready(self) -> bool:
        return self.next_action == DownstreamAction.ready

    @property
    def results_notified(self) -> bool:
        return self.data.get("results-notified", False)

    @results_notified.setter  # type: ignore
    @mut()
    def results_notified(self, value):
        self.data["results-notified"] = value

    @property
    def skip(self) -> bool:
        return self.data.get("skip", False)

    @skip.setter  # type: ignore
    @mut()
    def skip(self, value: bool) -> None:
        self.data["skip"] = value

    @property
    def tried_to_rebase(self) -> bool:
        return self.data.get("tried_to_rebase", False)

    @tried_to_rebase.setter  # type: ignore
    @mut()
    def tried_to_rebase(self, value: bool) -> None:
        self.data["tried_to_rebase"] = value

    @mut()
    def try_rebase(self) -> None:
        logger.info("Rebasing onto %s" % self.gecko_landing_branch())
        initial_tried_to_rebase = self.tried_to_rebase
        self.tried_to_rebase = True

        commit_hash_before_rebase = self.gecko_commits.base.sha1
        self.gecko_rebase(self.gecko_landing_branch(), abort_on_fail=True)
        commit_hash_after_rebase = self.gecko_commits.base.sha1

        if commit_hash_before_rebase == commit_hash_after_rebase:
            self.tried_to_rebase = initial_tried_to_rebase

    @property
    def wpt(self) -> WPT:
        git_work = self.wpt_worktree.get()
        git_work.git.reset(hard=True)
        return WPT(os.path.join(git_work.working_dir))

    @property
    def requires_try(self) -> bool:
        return not self.skip and len(self.gecko_commits) > 0

    @property
    def requires_stability_try(self) -> bool:
        return self.requires_try and self.has_affected_tests_readonly

    @property
    def latest_valid_try_push(self) -> TryPush | None:
        """Try push for the current head of the PR, if any.

        In legacy cases we don't store the wpt-head for the try push
        so we always assume that any try push is valid"""

        latest_try_push = self.latest_try_push
        if latest_try_push is None:
            return None

        if len(self.gecko_commits) == 0:
            # Something most likely is not correct, but we can't fix it here.
            return latest_try_push
        if self.metadata_commit is not None:
            if len(self.gecko_commits) == 1:
                # Apparently we only have a metadata commit and the actual change got rebased away
                # In this case the metadata commit is probably wrong, but we can't fix that here
                return latest_try_push
            gecko_head = self.gecko_commits[-2]
        else:
            gecko_head = self.gecko_commits[-1]

        # Check if the try push is for the current PR head
        if (latest_try_push.wpt_head and
            latest_try_push.wpt_head not in (self.pr_head.sha1, self.wpt_commits.head.sha1)):
            logger.info("Got more commits since latest try push")
            return None
        if latest_try_push.gecko_head not in {gecko_head.sha1,
                                              self.metadata_commit.sha1
                                              if self.metadata_commit else None}:
            logger.info("Gecko commits changed since latest try push")
            return None

        return latest_try_push

    @mut()
    def try_paths(self) -> Mapping[str, list[str]]:
        """Return a mapping of {test_type: path} for tests that should be run on try.

        Paths are relative to the gecko root"""
        affected_tests = self.affected_tests()
        base_path = env.config["gecko"]["path"]["wpt"]
        # Filter out paths that aren't in the head.
        # This can happen if the files were moved in a previous PR that we haven't yet
        # merged
        affected_paths = {}

        head_tree = self.gecko_commits.head.commit.tree

        def contains(tree: Tree, path: str) -> bool:
            path_parts = path.split(os.path.sep)
            for part in path_parts:
                try:
                    tree = tree[part]
                except KeyError:
                    return False
            return True

        for test_type, wpt_paths in affected_tests.items():
            paths = []
            for path in wpt_paths:
                gecko_path = os.path.join(base_path, path)
                if contains(head_tree, gecko_path):
                    paths.append(gecko_path)
            if paths:
                affected_paths[test_type] = paths

        if not affected_paths:
            # Default to just running infra tests
            infra_path = os.path.join(base_path, "infrastructure/")
            affected_paths = {
                "testharness": [infra_path],
                "reftest": [infra_path],
                "wdspec": [infra_path],
            }
        return affected_paths

    @mut()
    def next_try_push(self, try_cls: type = trypush.TryFuzzyCommit) -> TryPush | None:
        """Schedule a new try push for the sync, if required.

        A stability try push will only be scheduled if the upstream PR is
        approved, which we check directly from GitHub. Therefore returning
        None is not an indication that the sync is ready to land, just that
        there's no further action at this time.
        """
        if self.skip or self.status != "open":
            return None

        self.update_commits()
        # Ensure affected tests is up to date
        self.affected_tests()

        action = self.next_action
        if action == DownstreamAction.try_rebase:
            self.try_rebase()
            action = self.next_action
        if action == DownstreamAction.try_push:
            return TryPush.create(self._lock,
                                  self,
                                  affected_tests=self.try_paths(),
                                  stability=False,
                                  hacks=False,
                                  try_cls=try_cls)
        elif action == DownstreamAction.try_push_stability:
            return TryPush.create(self._lock,
                                  self,
                                  affected_tests=self.try_paths(),
                                  stability=True,
                                  hacks=False,
                                  try_cls=try_cls)
        return None

    @mut()
    def create_bug(self, git_wpt: Repo, pr_id: int, pr_title: str, pr_body: str | None) -> None:
        if self.bug is not None:
            return
        comment = self.make_bug_comment(git_wpt, pr_id, pr_title, pr_body)
        summary = f"[wpt-sync] Sync PR {pr_id} - {pr_title}"
        if len(summary) > 255:
            summary = summary[:254] + "\u2026"
        bug = env.bz.new(summary=summary,
                         comment=comment,
                         product="Testing",
                         component="web-platform-tests",
                         whiteboard="[wptsync downstream]",
                         priority="P4",
                         url=env.gh_wpt.pr_url(pr_id))
        self.bug = bug  # type: ignore

    @mut()
    def update_wpt_commits(self) -> None:
        """Update the set of commits in the PR from the latest upstream."""
        if not self.wpt_commits.head or self.wpt_commits.head.sha1 != self.pr_head.sha1:
            self.wpt_commits.head = self.pr_head  # type: ignore

        if (len(self.wpt_commits) == 0 and
            self.git_wpt.is_ancestor(self.wpt_commits.head.commit,
                                     self.git_wpt.rev_parse("origin/master"))):
            # The commits landed on master so we need to change the commit
            # range to not use origin/master as a base
            base_commit = None
            assert isinstance(self.wpt_commits.head, sync_commit.WptCommit)
            assert self.wpt_commits.head.pr() == self.pr
            for commit in self.git_wpt.iter_commits(self.wpt_commits.head.sha1):
                wpt_commit = sync_commit.WptCommit(self.git_wpt, commit)
                if wpt_commit.pr() != self.pr:
                    base_commit = wpt_commit
                    break
            assert base_commit is not None
            self.data["wpt-base"] = base_commit.sha1
            self.wpt_commits.base = base_commit  # type: ignore

    @mut()
    def update_github_check(self) -> None:
        if not env.config["web-platform-tests"]["github"]["checks"]["enabled"]:
            return
        title = "gecko/sync"
        head_sha = self.wpt_commits.head.sha1

        # TODO: maybe just get this from GitHub rather than store it
        existing = self.data.get("check")
        check_id = None
        if existing is not None:
            if existing.get("sha1") == head_sha:
                check_id = existing.get("id")

        assert self.bug is not None
        url = env.bz.bugzilla_url(self.bug)
        external_id = str(self.bug)

        # For now hardcode the status at completed
        status = "completed"
        conclusion = "neutral"

        completed_at = datetime.now()

        output = {"title": "Gecko sync for PR %s" % self.pr,
                  "summary": "Gecko sync status: %s" % self.landable_status.reason_str(),
                  "test": self.build_check_text(head_sha)}

        try:
            logger.info("Generating GH check status")
            resp = env.gh_wpt.set_check(title, check_id=check_id, commit_sha=head_sha, url=url,
                                        external_id=external_id, status=status, started_at=None,
                                        conclusion=conclusion, completed_at=completed_at,
                                        output=output)
            self.data["check"] = {"id": resp["id"], "sha1": head_sha}
        except AssertionError:
            raise
        except Exception:
            # Just log errors trying to update the check status, but otherwise don't fail
            newrelic.agent.record_exception()
            import traceback
            logger.error("Creating PR status check failed")
            logger.error(traceback.format_exc())

    def build_check_text(self, commit_sha: str) -> str:
        text = """

        # Summary

        [Bugzilla](%(bug_link)s)

        # Try pushes
        %(try_push_section)s

        %(error_section)s
        """
        try_pushes = [try_push for try_push in
                      sorted(self.try_pushes(), key=lambda x: -x.process_name.seq_id)
                      if try_push.wpt_head == commit_sha]
        if not try_pushes:
            try_push_section = "No current try pushes"
        else:
            items = []
            for try_push in try_pushes:
                link_str = "Try push" + (" (stability)" if try_push.stability else "")
                items.append(" * [{}]({}): {}{}".format(link_str,
                                                        try_push.treeherder_url,
                                                        try_push.status,
                                                        " infra-fail" if try_push.infra_fail
                                                        else ""))
                try_push_section = "\n".join(items)

        error_section = "# Errors:\n ```%s```" % self.error if self.error else ""
        assert self.bug is not None
        return text % {"bug_link": env.bz.bugzilla_url(self.bug),
                       "try_push_section": try_push_section,
                       "error_section": error_section}

    def files_changed(self) -> set[str]:
        # TODO: Would be nice to do this from mach with a gecko worktree
        return set(self.wpt.files_changed().decode("utf8", "replace").split("\n"))

    @property
    def metadata_commit(self) -> GeckoCommit | None:
        if len(self.gecko_commits) == 0:
            return None
        if self.gecko_commits[-1].metadata.get("wpt-type") == "metadata":
            commit = self.gecko_commits[-1]
            assert isinstance(commit, GeckoCommit)
            return commit
        return None

    @mut()
    def ensure_metadata_commit(self) -> GeckoCommit:
        if self.metadata_commit:
            return self.metadata_commit

        git_work = self.gecko_worktree.get()

        if "metadata-commit" in self.data:
            gitutils.cherry_pick(git_work, self.data["metadata-commit"])
            # We only care about the cached value inside this function, so
            # remove it now so we don't have to maintain it
            del self.data["metadata-commit"]
        else:
            assert all(item.metadata.get("wpt-type") != "metadata" for item in self.gecko_commits)
            metadata = {
                "wpt-pr": str(self.pr),
                "wpt-type": "metadata"
            }
            msg = sync_commit.Commit.make_commit_msg(
                b"Bug %s [wpt PR %s] - Update wpt metadata, a=testonly" %
                (str(self.bug).encode("utf8") if self.bug is not None else b"None",
                 str(self.pr).encode("utf8") if self.pr is not None else b"None"),
                metadata)
            sync_commit.create_commit(git_work, msg, allow_empty=True)
        commit = git_work.commit("HEAD")
        return sync_commit.GeckoCommit(self.git_gecko, commit.hexsha)

    @mut()
    def set_bug_component(self, files_changed: set[str]) -> None:
        new_component = bugcomponents.get(self.gecko_worktree.get(),
                                          files_changed,
                                          default=("Testing", "web-platform-tests"))
        env.bz.set_component(self.bug, *new_component)

    @mut()
    def move_metadata(self, renames: dict[str, str]) -> None:
        if not renames:
            return

        self.ensure_metadata_commit()

        gecko_work = self.gecko_worktree.get()
        metadata_base = env.config["gecko"]["path"]["meta"]
        for old_path, new_path in renames.items():
            old_meta_path = os.path.join(metadata_base,
                                         old_path + ".ini")
            if os.path.exists(os.path.join(gecko_work.working_dir,
                                           old_meta_path)):
                new_meta_path = os.path.join(metadata_base,
                                             new_path + ".ini")
                dir_name = os.path.join(gecko_work.working_dir,
                                        os.path.dirname(new_meta_path))
                if not os.path.exists(dir_name):
                    os.makedirs(dir_name)
                gecko_work.index.move((old_meta_path, new_meta_path))

        self._commit_metadata()

    @mut()
    def update_bug_components(self, renames: dict[str, str]) -> None:
        if not renames:
            return

        self.ensure_metadata_commit()

        gecko_work = self.gecko_worktree.get()
        bugcomponents.update(gecko_work, renames)

        self._commit_metadata()

    @mut()
    def _commit_metadata(self, amend: bool = True) -> None:
        assert self.metadata_commit
        gecko_work = self.gecko_worktree.get()
        if gecko_work.is_dirty():
            logger.info("Updating metadata commit")
            try:
                gecko_work.git.commit(amend=True, no_edit=True)
            except git.GitCommandError as e:
                if amend and e.status == 1 and "--allow-empty" in e.stdout:
                    logger.warning("Amending commit made it empty, resetting")
                    gecko_work.git.reset("HEAD^")

    @mut()
    def update_commits(self) -> bool:
        exception = None
        try:
            self.update_wpt_commits()

            # Check if this sync reverts some unlanded earlier PR and if so mark both
            # as skip and don't try to apply the commits here
            reverts = self.reverts_syncs()
            if reverts:
                all_open = all(item.status == "open" for item in reverts)
                for revert_sync in reverts:
                    if revert_sync.status == "open":
                        logger.info("Skipping sync for PR %s because it is later reverted" %
                                    revert_sync.pr)
                        with SyncLock.for_process(revert_sync.process_name) as revert_lock:
                            assert isinstance(revert_lock, SyncLock)
                            with revert_sync.as_mut(revert_lock):
                                revert_sync.skip = True  # type: ignore
                # TODO: If this commit reverts some closed syncs, then set the metadata
                # commit of this commit to the revert of the metadata commit from that
                # sync
                if all_open:
                    logger.info("Sync was a revert of other open syncs, skipping")
                    self.skip = True  # type: ignore
                    return False

            old_gecko_head = self.gecko_commits.head.sha1
            logger.debug(f"PR {self.pr} gecko HEAD was {old_gecko_head}")

            def plain_apply() -> bool:
                logger.info("Applying on top of the current commits")
                self.wpt_to_gecko_commits()
                return True

            def rebase_apply() -> bool:
                logger.info("Applying with a rebase onto latest integration branch")
                new_base = self.gecko_integration_branch()
                gecko_work = self.gecko_worktree.get()
                reset_head = "HEAD"
                if (len(self.gecko_commits) > 0 and
                    self.gecko_commits[0].metadata.get("wpt-type") == "dependent"):
                    # If we have any dependent commits first reset to the new
                    # head. This prevents conflicts if the dependents already
                    # landed
                    # TODO: Actually check if they landed?
                    reset_head = new_base
                gecko_work.git.reset(reset_head, hard=True)
                self.gecko_rebase(new_base, abort_on_fail=True)
                self.wpt_to_gecko_commits()
                return True

            def dependents_apply() -> bool:
                logger.info("Applying with upstream dependents")
                dependencies = self.unlanded_commits_same_files()
                if dependencies:
                    logger.info("Found dependencies:\n%s" %
                                "\n".join(item.msg.splitlines()[0].decode("utf8", "replace")
                                          for item in dependencies))
                    self.wpt_to_gecko_commits(dependencies)
                    assert self.bug is not None
                    env.bz.comment(self.bug,
                                   "PR %s applied with additional changes from upstream: %s"
                                   % (self.pr, ", ".join(item.sha1 for item in dependencies)))
                    return True
                return False

            error = None
            for fn in [plain_apply, rebase_apply, dependents_apply]:
                try:
                    if fn():
                        error = None
                        break
                    else:
                        logger.error("Applying with %s was a no-op" % fn.__name__)
                except Exception as e:
                    import traceback
                    error = e
                    logger.error("Applying with %s errored" % fn.__name__)
                    logger.error(traceback.format_exc())

            if error is not None:
                raise error

            logger.debug(f"PR {self.pr} gecko HEAD now {self.gecko_commits.head.sha1}")
            if old_gecko_head == self.gecko_commits.head.sha1:
                logger.info("Gecko commits did not change for PR %s" % self.pr)
                return False

            # If we have a metadata commit already, ensure it's applied now
            if "metadata-commit" in self.data:
                self.ensure_metadata_commit()

            renames = self.wpt_renames()
            self.move_metadata(renames)
            self.update_bug_components(renames)

            files_changed = self.files_changed()
            self.set_bug_component(files_changed)
        except Exception as e:
            exception = e
            raise
        finally:
            # If we managed to apply all the commits without error, reset the error flag
            # otherwise update it with the current exception
            self.error = exception
        return True

    @mut()
    def wpt_to_gecko_commits(self, dependencies: list[WptCommit] | None = None) -> None:
        """Create a patch based on wpt branch, apply it to corresponding gecko branch.

        If there is a commit with wpt-type metadata, this function will remove it. The
        sha1 will be stashed in self.data["metadata-commit"] so it can be restored next time
        we call ensure_metadata_commit()
        """
        # The logic here is that we can retain any dependent commits as long as we have
        # at least the set in the dependencies array, followed by the gecko commits created
        # from the wpt_commits, interspersed with any number of manifest commits,
        # followed by zero or one metadata commits

        if dependencies:
            expected_commits: list[tuple[str, WptCommit | None, bool]] = [
                (item.sha1, item, True)
                for item in dependencies]
        else:
            # If no dependencies are supplied, retain the ones that we alredy have, if any
            expected_commits = []
            for commit in self.gecko_commits:
                assert isinstance(commit, sync_commit.GeckoCommit)
                if commit.metadata.get("wpt-type") == "dependency":
                    expected_commits.append((commit.metadata["wpt-commit"], None, True))
                else:
                    break

        # Expect all the new commits
        for commit in self.wpt_commits:
            assert isinstance(commit, WptCommit)
            if not commit.is_merge:
                expected_commits.append((commit.sha1, commit, False))

        existing = [
            commit for commit in self.gecko_commits
            if commit.metadata.get("wpt-commit") and
            commit.metadata.get("wpt-type") in ("dependency", None)]
        if TYPE_CHECKING:
            existing_commits = cast(List[GeckoCommit], existing)
        else:
            existing_commits = existing

        retain_commits = 0
        for gecko_commit, (wpt_sha1, _, _) in zip(existing_commits, expected_commits):
            if gecko_commit.metadata.get("wpt-commit") != wpt_sha1:
                break
            retain_commits += 1

        keep_commits = existing_commits[:retain_commits]
        maybe_add_commits = expected_commits[retain_commits:]

        # Strip out any leading commits that come from currently applied dependencies that are
        # not being retained
        strip_count = 0
        for _, wpt_commit, _ in maybe_add_commits:
            if wpt_commit is not None:
                break
            strip_count += 1
        add_commits = maybe_add_commits[strip_count:]

        if len(keep_commits) == len(existing_commits) and not add_commits:
            logger.info("Commits did not change")
            return

        logger.info("Keeping %i existing commits; adding %i new commits" % (len(keep_commits),
                                                                            len(add_commits)))

        if self.metadata_commit:
            # If we have a metadata commit, store it in self.data["metadata-commit"]
            # remove it when updating commits, and reapply it when we next call
            # ensure_metadata_commit
            self.data["metadata-commit"] = self.metadata_commit.sha1

        reset_head = None
        if not keep_commits:
            reset_head = self.data["gecko-base"]
        elif len(keep_commits) < len(existing_commits):
            reset_head = keep_commits[-1]
        elif ("metadata-commit" in self.data and
              self.gecko_commits[-1].metadata.get("wpt-type") == "metadata"):
            reset_head = self.gecko_commits[-2]

        # Clear the set of affected tests since there are updates
        del self.data["affected-tests"]

        gecko_work = self.gecko_worktree.get()

        if reset_head:
            self.gecko_commits.head = reset_head  # type: ignore
        gecko_work.git.reset(hard=True)

        for _, wpt_commit, is_dependency in add_commits:
            assert wpt_commit is not None
            logger.info("Moving commit %s" % wpt_commit.sha1)
            if is_dependency:
                metadata = {
                    "wpt-type": "dependency",
                    "wpt-commit": wpt_commit.sha1
                }
                msg_filter = None
            else:
                metadata = {
                    "wpt-pr": str(self.pr),
                    "wpt-commit": wpt_commit.sha1
                }
                msg_filter = self.message_filter

            wpt_commit.move(gecko_work,
                            dest_prefix=env.config["gecko"]["path"]["wpt"],
                            msg_filter=msg_filter,
                            metadata=metadata,
                            patch_fallback=True)

    def unlanded_commits_same_files(self) -> list[WptCommit]:
        from . import landing

        sync_point = landing.load_sync_point(self.git_gecko, self.git_wpt)
        base = sync_point["upstream"]
        head = "origin/master"
        changed = self.wpt_commits.files_changed
        commits = []
        for commit in self.git_wpt.iter_commits(f"{base}..{head}",
                                                reverse=True,
                                                paths=list(changed)):
            wpt_commit = sync_commit.WptCommit(self.git_wpt, commit)
            # Check for same-pr rather than same-commit because we always
            # use the commits on the PR branch, not the merged commits.
            # The other option is to use the GH API to decide if the PR
            # merged and if so what the merge commit was, although in that
            # case we would still not know the commit prior to merge, which
            # is what we need
            if wpt_commit.pr() == self.pr:
                break
            commits.append(wpt_commit)
        return commits

    def message_filter(self, msg: bytes) -> tuple[bytes, dict[str, str]]:
        msg = sync_commit.try_filter(msg)
        parts = msg.split(b"\n", 1)
        if len(parts) > 1:
            summary, body = parts
        else:
            summary = parts[0]
            body = b""

        new_msg = b"Bug %s [wpt PR %s] - %s, a=testonly\n\nSKIP_BMO_CHECK\n%s" % (
            str(self.bug).encode("utf8"),
            str(self.pr).encode("utf8"),
            summary,
            body)
        return new_msg, {}

    @mut()
    def affected_tests(self, revish: Any | None = None) -> Mapping[str, list[str]]:
        # TODO? support files, harness changes -- don't want to update metadata
        if "affected-tests" not in self.data:
            tests_by_type: MutableMapping[str, list[str]] = defaultdict(list)
            logger.info("Updating MANIFEST.json")
            self.wpt.manifest()
            args = ["--show-type", "--new"]
            if revish:
                args.append(revish)
            logger.info("Getting a list of tests affected by changes.")
            try:
                output = self.wpt.tests_affected(*args)
            except subprocess.CalledProcessError:
                logger.error("Calling wpt tests-affected failed")
                return tests_by_type
            if output:
                for item_bytes in output.strip().split(b"\n"):
                    item = item_bytes.decode("utf8", "replace")
                    path, test_type = item.strip().split("\t")
                    tests_by_type[test_type].append(path)
            self.data["affected-tests"] = tests_by_type
        return self.data["affected-tests"]

    @property
    def affected_tests_readonly(self) -> Mapping[str, list[str]]:
        if "affected-tests" not in self.data:
            logger.warning("Trying to get affected tests before it's set")
            return {}
        return self.data["affected-tests"]

    @property
    def has_affected_tests_readonly(self) -> bool:
        if "affected-tests" not in self.data:
            logger.warning("Trying to get affected tests before it's set")
            return True
        return bool(self.data["affected-tests"])

    @mut()
    def update_metadata(self, log_files, stability=False):
        meta_path = env.config["gecko"]["path"]["meta"]
        gecko_work = self.gecko_worktree.get()

        mach = Mach(gecko_work.working_dir)
        args = []

        if stability:
            help_text = mach.wpt_update("--help").decode("utf8")
            if "--stability " in help_text:
                args.extend(["--stability", "wpt-sync Bug %s" % self.bug])
            else:
                args.append("--update-intermittent")
        args.extend(log_files)

        logger.debug("Updating metadata")
        output = mach.wpt_update(*args)
        prefix = b"disabled:"
        disabled = []
        for line in output.split(b"\n"):
            if line.startswith(prefix):
                disabled.append(line[len(prefix):].decode("utf8", "replace").strip())

        if gecko_work.is_dirty(untracked_files=True, path=meta_path):
            self.ensure_metadata_commit()
            gecko_work.git.add(meta_path, all=True)
            self._commit_metadata()

        return disabled

    @mut()
    def try_notify(self, force: bool = False) -> None:
        newrelic.agent.record_custom_event("try_notify", params={
            "sync_bug": self.bug,
            "sync_pr": self.pr
        })

        if self.results_notified and not force:
            return

        if not self.bug:
            logger.error("Sync for PR %s has no associated bug" % self.pr)
            return

        if not self.affected_tests():
            logger.debug("PR %s doesn't have affected tests so skipping results notification" %
                         self.pr)
            newrelic.agent.record_custom_event("try_notify_no_affected", params={
                "sync_bug": self.bug,
                "sync_pr": self.pr
            })
            return

        logger.info("Trying to generate results notification for PR %s" % self.pr)

        results = notify.results.for_sync(self)

        if not results:
            # TODO handle errors here better, perhaps
            logger.error("Failed to get results notification for PR %s" % self.pr)
            newrelic.agent.record_custom_event("try_notify_failed", params={
                "sync_bug": self.bug,
                "sync_pr": self.pr
            })
            return

        message, truncated = notify.msg.for_results(results)

        with env.bz.bug_ctx(self.bug) as bug:
            if truncated:
                bug.add_attachment(data=message.encode("utf8"),
                                   file_name="wpt-results.md",
                                   summary="Notable wpt changes",
                                   is_markdown=True,
                                   comment=truncated)
            else:
                env.bz.comment(self.bug, message, is_markdown=True)

        bugs = notify.bugs.for_sync(self, results)
        notify.bugs.update_metadata(self, bugs)

        self.results_notified = True  # type: ignore

        with SyncLock.for_process(self.process_name) as lock:
            assert isinstance(lock, SyncLock)
            for try_push in self.try_pushes():
                with try_push.as_mut(lock):
                    try_push.cleanup_logs()

    def reverts_syncs(self) -> set[DownstreamSync]:
        """Return a set containing the previous syncs reverted by this one, if any"""
        revert_re = re.compile(b"This reverts commit ([0-9A-Fa-f]+)")
        unreverted_commits = defaultdict(set)
        for commit in self.wpt_commits:
            if not commit.msg.startswith(b"Revert "):
                # If not everything is a revert then return
                return set()
            revert_shas = revert_re.findall(commit.msg)
            if len(revert_shas) == 0:
                return set()
            # Just use the first match for now
            sha = revert_shas[0].decode("ascii")
            try:
                # Reassign the hash here, in case a short hash was used for reverting.
                sha = str(self.git_wpt.rev_parse(sha))
            except (ValueError, git.BadName):
                # Commit isn't in this repo (could be upstream)
                return set()
            pr = env.gh_wpt.pr_for_commit(sha)
            if pr is None:
                return set()
            sync = DownstreamSync.for_pr(self.git_gecko, self.git_wpt, pr)
            if sync is None:
                return set()
            assert isinstance(sync, DownstreamSync)
            if sync not in unreverted_commits:
                # Ensure we have the latest commits for the reverted sync
                with SyncLock.for_process(sync.process_name) as revert_lock:
                    assert isinstance(revert_lock, SyncLock)
                    with sync.as_mut(revert_lock):
                        sync.update_wpt_commits()
                unreverted_commits[sync] = {item.sha1 for item in sync.wpt_commits}
            if sha in unreverted_commits[sync]:
                unreverted_commits[sync].remove(sha)
            # If the commit is not part of the sync, check if the PR was squashed and then reverted,
            # in that case all commits of the sync should be reverted.
            elif sha == env.gh_wpt.merge_sha(pr):
                unreverted_commits[sync] = set()

        rv = {sync for sync, unreverted in unreverted_commits.items()
              if not unreverted}
        return rv


@entry_point("downstream")
def new_wpt_pr(git_gecko: Repo, git_wpt: Repo, pr_data: Mapping[str, Any],
               raise_on_error: bool = True, repo_update: bool = True) -> None:
    """ Start a new downstream sync """
    if pr_data["user"]["login"] == env.config["web-platform-tests"]["github"]["user"]:
        raise ValueError("Tried to create a downstream sync for a PR created "
                         "by the wpt bot")
    if repo_update:
        update_repositories(git_gecko, git_wpt)
    pr_id = pr_data["number"]
    if DownstreamSync.for_pr(git_gecko, git_wpt, pr_id):
        return
    wpt_base = "origin/%s" % pr_data["base"]["ref"]

    with SyncLock("downstream", str(pr_id)) as lock:
        sync = DownstreamSync.new(lock,
                                  git_gecko,
                                  git_wpt,
                                  wpt_base,
                                  pr_id,
                                  pr_data["title"],
                                  pr_data["body"] or "")
        with sync.as_mut(lock):
            try:
                sync.update_commits()
                sync.update_github_check()
            except Exception as e:
                sync.error = e
                if raise_on_error:
                    raise
                traceback.print_exc()
                logger.error(e)
            # Now wait for the status to change before we take any actions


@entry_point("downstream")
@mut("try_push", "sync")
def try_push_complete(git_gecko, git_wpt, try_push, sync):
    if not try_push.taskgroup_id:
        logger.error("No taskgroup id set for try push")
        return

    if not try_push.status == "complete":
        # Ensure we don't have some old set of tasks

        tasks = try_push.tasks()
        if not tasks.complete(allow_unscheduled=True):
            logger.info("Try push %s is not complete" % try_push.treeherder_url)
            return
        logger.info("Try push %s is complete" % try_push.treeherder_url)
        try:
            if not tasks.validate():
                try_push.infra_fail = True
                if len(sync.latest_busted_try_pushes()) > 5:
                    message = ("Too many busted try pushes. "
                               "Check the try results for infrastructure issues.")
                    sync.error = message
                    env.bz.comment(sync.bug, message)
                    try_push.status = "complete"
                    raise AbortError(message)
            elif len(tasks.failed_builds()):
                message = ("Try push had build failures")
                sync.error = message
                env.bz.comment(sync.bug, message)
                try_push.status = "complete"
                try_push.infra_fail = True
                raise AbortError(message)
            else:
                logger.info(f"Try push {try_push!r} for PR {sync.pr} complete")
                disabled = []
                if tasks.has_failures():
                    if sync.affected_tests():
                        log_files = []
                        wpt_tasks = try_push.download_logs(tasks.wpt_tasks)
                        for task in wpt_tasks:
                            for run in task.get("status", {}).get("runs", []):
                                log = run.get("_log_paths", {}).get("wptreport.json")
                                if log:
                                    log_files.append(log)
                        if not log_files:
                            raise ValueError("No log files found for try push %r" % try_push)
                        disabled = sync.update_metadata(log_files, stability=try_push.stability)
                    else:
                        env.bz.comment(sync.bug, ("The PR was not expected to affect any tests, "
                                                  "but the try push wasn't a success. "
                                                  "Check the try results for infrastructure "
                                                  "issues"))
                        # TODO: consider marking the push an error here so that we can't
                        # land without manual intervention

                if try_push.stability and disabled:
                    logger.info("The following tests were disabled:\n%s" % "\n".join(disabled))
                    # TODO notify relevant people about test expectation changes, stability
                    env.bz.comment(sync.bug, ("The following tests were disabled "
                                              "based on stability try push:\n %s" %
                                              "\n".join(disabled)))

            try_push.status = "complete"
            sync.next_try_push()
        finally:
            sync.update_github_check()
    else:
        sync.next_try_push()
        sync.update_github_check()

    if sync.metadata_commit is not None and len(sync.gecko_commits) == 1:
        # Apparently we only have a metadata commit and the actual change got rebased away
        # In this case the metadata commit is probably wrong,
        # and we just want to skip this sync.
        sync.skip = True

    if sync.landable_status == LandableStatus.ready:
        sync.try_notify()


@entry_point("downstream")
@mut("sync")
def update_pr(git_gecko: Repo,
              git_wpt: Repo,
              sync: DownstreamSync,
              action: str,
              merge_sha: str,
              base_sha: str,
              merged_by: str | None = None,
              ) -> None:
    try:
        if action == "closed" and not merge_sha:
            sync.pr_status = "closed"  # type: ignore
            if sync.bug:
                env.bz.set_status(sync.bug, "RESOLVED", "INVALID")
            sync.finish()
        elif action == "closed":
            # We are storing the wpt base as a reference
            sync.data["wpt-base"] = base_sha
            sync.next_try_push()
        elif action == "reopened" or action == "open":
            sync.status = "open"  # type: ignore
            sync.pr_status = "open"  # type: ignore
            sync.next_try_push()
            assert sync.bug is not None
            if sync.bug:
                status = env.bz.get_status(sync.bug)
                if status is not None and status[0] == "RESOLVED":
                    env.bz.set_status(sync.bug, "REOPENED")
        sync.update_github_check()
    except Exception as e:
        sync.error = e
        raise