sync/landing.py

from __future__ import annotations
import os
import shutil
from collections import defaultdict

import enum
import git
from celery.exceptions import OperationalError

from . import bug
from . import bugcomponents
from . import commit as sync_commit
from . import downstream
from . import gitutils
from . import log
from . import tree
from . import load
from . import repos
from . import trypush
from . import upstream
from .base import entry_point
from .commit import GeckoCommit, first_non_merge
from .env import Environment
from .gitutils import update_repositories
from .lock import SyncLock, constructor, mut
from .errors import AbortError, RetryableError
from .projectutil import Mach
from .repos import cinnabar, pygit2_get
from .sync import LandableStatus, SyncProcess

from typing import Any, IO, List, Tuple, Union, cast, TYPE_CHECKING
from git.repo.base import Repo
if TYPE_CHECKING:
    from sync.base import ProcessName
    from sync.commit import Commit, WptCommit
    from sync.downstream import DownstreamSync
    from sync.gh import AttrDict
    from sync.trypush import TryPush, TryPushTasks
    from sync.upstream import UpstreamSync

    LandableCommits = List[Tuple[int, Union[DownstreamSync, UpstreamSync], List[WptCommit]]]

env = Environment()

logger = log.get_logger(__name__)


class SyncPoint:
    def __init__(self, data: Any | None = None) -> None:
        self._items: dict[str, str] = {}
        if data is not None:
            self._items.update(data)

    def __getitem__(self, key: str) -> str:
        return self._items[key]

    def __setitem__(self, key: str, value: str) -> None:
        self._items[key] = value

    def load(self, fp):
        with open(fp, "rb") as f:
            self.loads(f)

    def loads(self, data: bytes) -> None:
        for line in data.split(b"\n"):
            if line:
                key, value = (item.decode("utf8") for item in line.split(b": ", 1))
                self._items[key] = value

    def dump(self, fp: IO[bytes]) -> None:
        fp.write(self.dumps().encode("utf8") + b"\n")

    def dumps(self) -> str:
        return "\n".join(f"{key}: {value}" for key, value in self._items.items())


@enum.unique
class TryPushResult(enum.Enum):
    success = 0
    acceptable_failures = 1
    infra_fail = 2
    too_many_failures = 3
    pending = 4

    def is_failure(self) -> bool:
        return self in (TryPushResult.infra_fail, TryPushResult.too_many_failures)

    def is_ok(self) -> bool:
        return self in (TryPushResult.success, TryPushResult.acceptable_failures)


class LandingSync(SyncProcess):
    sync_type = "landing"
    obj_id = "bug"
    statuses = ("open", "complete")
    status_transitions = [("open", "complete"), ("complete", "open")]

    def __init__(self, git_gecko: Repo, git_wpt: Repo, process_name: ProcessName) -> None:
        super().__init__(git_gecko, git_wpt, process_name)
        self._unlanded_gecko_commits = None

    @classmethod
    @constructor(lambda args: ("landing", None))
    def new(cls,
            lock: SyncLock,
            git_gecko: Repo,
            git_wpt: Repo,
            wpt_base: str,
            wpt_head: str,
            bug: int | None = None,
            ) -> LandingSync:
        # There is some chance here we create a bug but never create the branch.
        # Probably need something to clean up orphan bugs

        # The gecko branch is a new one based on master
        gecko_base = cls.gecko_integration_branch()
        gecko_head = cls.gecko_integration_branch()

        if bug is None:
            bug = env.bz.new("Update web-platform-tests to %s" % wpt_head,
                             "",
                             "Testing",
                             "web-platform-tests",
                             whiteboard="[wptsync landing]")

        return super().new(lock,
                           git_gecko,
                           git_wpt,
                           gecko_base,
                           gecko_head,
                           wpt_base=wpt_base,
                           wpt_head=wpt_head,
                           bug=bug)

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

    def unlanded_gecko_commits(self):
        """Get a list of gecko commits that correspond to commits which have
        landed on the gecko integration branch, but are not yet merged into the
        upstream commit we are updating to.

        There are two possible sources of such commits:
          * Unlanded PRs. These correspond to upstream syncs with status of "open"
          * Gecko PRs that landed between the wpt commit that we are syncing to
            and latest upstream master.

        :return: List of commits in the order in which they originally landed in gecko"""

        if self._unlanded_gecko_commits is None:
            commits = []

            def on_integration_branch(commit):
                # Calling this continually is O(N*M) where N is the number of unlanded commits
                # and M is the average depth of the commit in the gecko tree
                # If we need a faster implementation one approach would be to store all the
                # commits not on the integration branch and check if this commit is in that set
                return self.git_gecko.is_ancestor(commit,
                                                  self.git_gecko.rev_parse(
                                                      self.gecko_integration_branch()))

            # All the commits from unlanded upstream syncs that are reachable from the
            # integration branch
            unlanded_syncs = set()
            for status in ["open", "wpt-merged"]:
                unlanded_syncs |= set(upstream.UpstreamSync.load_by_status(self.git_gecko,
                                                                           self.git_wpt,
                                                                           status))

            for sync in unlanded_syncs:
                branch_commits = [commit.sha1 for commit in sync.gecko_commits if
                                  on_integration_branch(commit)]
                if branch_commits:
                    logger.info("Commits from unlanded sync for bug %s (PR %s) will be reapplied" %
                                (sync.bug, sync.pr))
                    commits.extend(branch_commits)

            # All the gecko commits that landed between the base sync point and master
            # We take the base here and then remove upstreamed commits that we are landing
            # as we reach them so that we can get the right diffs for the other PRs
            unlanded_commits = self.git_wpt.iter_commits("%s..origin/master" %
                                                         self.wpt_commits.base.sha1)
            seen_bugs = set()
            for commit in unlanded_commits:
                wpt_commit = sync_commit.WptCommit(self.git_wpt, commit)
                gecko_commit = wpt_commit.metadata.get("gecko-commit")
                if gecko_commit:
                    git_sha = cinnabar(self.git_gecko).hg2git(gecko_commit)
                    commit = sync_commit.GeckoCommit(self.git_gecko, git_sha)
                    bug_number = bug.bug_number_from_url(commit.metadata.get("bugzilla-url"))
                    if on_integration_branch(commit):
                        if bug_number and bug_number not in seen_bugs:
                            logger.info("Commits from landed sync for bug %s will be reapplied" %
                                        bug_number)
                            seen_bugs.add(bug_number)
                        commits.append(commit.sha1)

            commits = set(commits)

            # Order the commits according to the order in which they landed in gecko
            ordered_commits = []
            for commit in self.git_gecko.iter_commits(self.gecko_integration_branch(),
                                                      paths=env.config["gecko"]["path"]["wpt"]):
                if commit.hexsha in commits:
                    ordered_commits.append(commit.hexsha)
                    commits.remove(commit.hexsha)
                if not commits:
                    break

            self._unlanded_gecko_commits = list(reversed(
                [sync_commit.GeckoCommit(self.git_gecko, item) for item in ordered_commits]))
        return self._unlanded_gecko_commits

    def has_metadata_for_sync(self, sync: DownstreamSync) -> bool:
        for item in reversed(list(self.gecko_commits)):
            if (item.metadata.get("wpt-pr") == sync.pr and
                item.metadata.get("wpt-type") == "metadata"):
                return True
        return False

    @property
    def landing_commit(self) -> Any | None:
        head = self.gecko_commits.head
        if (head.metadata.get("wpt-type") == "landing" and
            head.metadata.get("wpt-head") == self.wpt_commits.head.sha1):
            return head
        return None

    @mut()
    def add_pr(self,
               pr_id: int,
               sync: DownstreamSync | UpstreamSync,
               wpt_commits: list[WptCommit],
               copy: bool = True,
               prev_wpt_head: str | None = None,
               ) -> Commit | None:
        if len(wpt_commits) > 1:
            assert all(item.pr() == pr_id for item in wpt_commits)

        # Assume we can always use the author of the first commit
        author = first_non_merge(wpt_commits).author

        git_work_wpt = self.wpt_worktree.get()
        git_work_gecko = self.gecko_worktree.get()

        pr = env.gh_wpt.get_pull(int(pr_id))

        metadata = {
            "wpt-pr": pr_id,
            "wpt-commits": ", ".join(item.sha1 for item in wpt_commits)
        }

        message = b"""Bug %d [wpt PR %d] - %s, a=testonly

Automatic update from web-platform-tests\n%s
"""
        message = message % ((sync and sync.bug) or self.bug,
                             pr.number,
                             pr.title.encode("utf8"),
                             b"\n--\n".join(item.msg for item in wpt_commits) + b"\n--")

        message = sync_commit.try_filter(message)

        upstream_changed = set()
        diffs = wpt_commits[-1].commit.diff(wpt_commits[0].commit.parents[0])
        for diff in diffs:
            new_path = diff.b_path
            if new_path:
                upstream_changed.add(new_path)

        logger.info("Upstream files changed:\n%s" % "\n".join(sorted(upstream_changed)))

        # If this is originally an UpstreamSync and no new changes were introduced to the GH PR
        # then we can safely skip and not need to re-apply these changes. Compare the hash of
        # the upstreamed gecko commits against the final hash in the PR.
        if isinstance(sync, upstream.UpstreamSync):
            commit_is_local = False
            pr_head = sync.pr_head
            if sync.wpt_commits.head.sha1 == pr_head:
                commit_is_local = True
            else:
                # Check if we rebased locally without pushing the rebase;
                # this is a thing we used to do to check the PR would merge
                try:
                    ref_log = sync.git_wpt.references[sync.branch_name].log()
                except Exception:
                    # If we can't read the reflog just skip this
                    pass
                else:
                    commit_is_local = any(entry.newhexsha == pr_head for entry in ref_log)
            if commit_is_local:
                logger.info("Upstream sync doesn't introduce any gecko changes")
                return None

        if copy:
            commit = self.copy_pr(git_work_gecko, git_work_wpt, pr, wpt_commits,
                                  message, author, metadata)
        else:
            commit = self.move_pr(git_work_gecko, git_work_wpt, pr, wpt_commits,
                                  message, author, prev_wpt_head, metadata)

        if commit is not None:
            self.gecko_commits.head = commit  # type: ignore

        return commit

    @mut()
    def copy_pr(self, git_work_gecko, git_work_wpt, pr, wpt_commits, message, author, metadata):
        # Ensure we have anything in a wpt submodule
        git_work_wpt.git.submodule("update", "--init", "--recursive")

        dest_path = os.path.join(git_work_gecko.working_dir,
                                 env.config["gecko"]["path"]["wpt"])
        src_path = git_work_wpt.working_dir

        # Specific paths that should be re-checked out
        keep_paths = {"LICENSE", "resources/testdriver_vendor.js"}
        # file names that are ignored in any part of the tree
        ignore_files = {".git"}

        logger.info("Setting wpt HEAD to %s" % wpt_commits[-1].sha1)
        git_work_wpt.head.reference = wpt_commits[-1].commit
        git_work_wpt.head.reset(index=True, working_tree=True)

        # First remove all files so we handle deletion correctly
        shutil.rmtree(dest_path)

        ignore_paths = defaultdict(set)
        for name in keep_paths:
            src, name = os.path.split(os.path.join(src_path, name))
            ignore_paths[src].add(name)

        def ignore_names(src, names):
            rv = []
            for item in names:
                if item in ignore_files:
                    rv.append(item)
            if src in ignore_paths:
                rv.extend(ignore_paths[src])
            return rv

        shutil.copytree(src_path,
                        dest_path,
                        ignore=ignore_names)

        # Now re-checkout the files we don't want to change
        # checkout-index allows us to ignore files that don't exist
        git_work_gecko.git.checkout_index(*(os.path.join(env.config["gecko"]["path"]["wpt"], item)
                                            for item in keep_paths), force=True, quiet=True)

        allow_empty = False
        if not git_work_gecko.is_dirty(untracked_files=True):
            logger.info("PR %s didn't add any changes" % pr.number)
            allow_empty = True

        git_work_gecko.git.add(env.config["gecko"]["path"]["wpt"],
                               no_ignore_removal=True)

        message = sync_commit.Commit.make_commit_msg(message, metadata)

        commit = git_work_gecko.index.commit(message=message,
                                             author=git.Actor._from_string(author))
        logger.debug("Gecko files changed: \n%s" % "\n".join(list(commit.stats.files.keys())))
        gecko_commit = sync_commit.GeckoCommit(self.git_gecko,
                                               commit.hexsha,
                                               allow_empty=allow_empty)

        return gecko_commit

    @mut()
    def move_pr(self,
                git_work_gecko: Repo,
                git_work_wpt: Repo,
                pr: AttrDict,
                wpt_commits: list[WptCommit],
                message: bytes,
                author: str,
                prev_wpt_head: str,
                metadata: dict[str, str],
                ) -> Commit | None:
        if prev_wpt_head is None:
            if wpt_commits[-1].is_merge:
                base = wpt_commits[-1].sha1 + "^"
            else:
                base = wpt_commits[0].sha1 + "^"
        else:
            base = self.git_wpt.git.merge_base(prev_wpt_head, wpt_commits[-1].sha1)

        head = sync_commit.GeckoCommit(self.git_gecko, git_work_gecko.head.commit)
        if (head.is_downstream and
            head.metadata.get("wpt-pr") == str(pr.number)):
            return None

        revish = f"{base}..{wpt_commits[-1].sha1}"
        logger.info("Moving wpt commits %s" % revish)

        return sync_commit.move_commits(self.git_wpt,
                                        revish,
                                        message,
                                        git_work_gecko,
                                        dest_prefix=env.config["gecko"]["path"]["wpt"],
                                        amend=False,
                                        metadata=metadata,
                                        rev_name="pr-%s" % pr.number,
                                        author=first_non_merge(wpt_commits).author,
                                        exclude={"resources/testdriver_vendor.js"},
                                        allow_empty=True)

    @mut()
    def reapply_local_commits(self, gecko_commits_landed):
        # The local commits to apply are everything that hasn't been landed at this
        # point in the process
        commits = [item for item in self.unlanded_gecko_commits()
                   if item.canonical_rev not in gecko_commits_landed]

        landing_commit = self.gecko_commits[-1]
        git_work_gecko = self.gecko_worktree.get()

        logger.debug("Reapplying commits: %s" % " ".join(item.canonical_rev for item in commits))

        if not commits:
            return

        already_applied = landing_commit.metadata.get("reapplied-commits")
        if already_applied:
            already_applied = [item.strip() for item in already_applied.split(",")]
        else:
            already_applied = []
        already_applied_set = set(already_applied)

        unapplied_gecko_commits = [item for item in commits if item.canonical_rev
                                   not in already_applied_set]

        try:
            for i, commit in enumerate(unapplied_gecko_commits):
                def msg_filter(_):
                    msg = landing_commit.msg
                    reapplied_commits = (already_applied +
                                         [commit.canonical_rev for commit in commits[:i + 1]])
                    metadata = {"reapplied-commits": ", ".join(reapplied_commits)}
                    return msg, metadata
                logger.info(f"Reapplying {commit.sha1} - {commit.msg}")
                # Passing in a src_prefix here means that we only generate a patch for the
                # part of the commit that affects wpt, but then we need to undo it by adding
                # the same dest prefix
                commit = commit.move(git_work_gecko,
                                     msg_filter=msg_filter,
                                     src_prefix=env.config["gecko"]["path"]["wpt"],
                                     dest_prefix=env.config["gecko"]["path"]["wpt"],
                                     three_way=True,
                                     amend=True)
                if commit is None:
                    break

        except AbortError as e:
            err_msg = (
                f"Landing wpt failed because reapplying commits failed:\n{e.message}"
            )
            env.bz.comment(self.bug, err_msg)
            raise AbortError(err_msg)

    @mut()
    def add_metadata(self, sync: DownstreamSync) -> None:
        logger.info("Adding metadata from downstream sync")

        if self.has_metadata_for_sync(sync):
            logger.info("Metadata already applied for PR %s" % sync.pr)
            return

        if not sync.metadata_commit or sync.metadata_commit.is_empty():
            logger.info("No metadata commit available for PR %s" % sync.pr)
            return

        worktree = self.gecko_worktree.get()

        success = gitutils.cherry_pick(worktree, sync.metadata_commit.sha1)

        if not success:
            logger.info("Cherry-pick failed, trying again with only test-related changes")
            # Try to reset all metadata files that aren't related to an affected test.
            affected_metadata = {os.path.join(env.config["gecko"]["path"]["meta"], item) + ".ini"
                                 for items in sync.affected_tests_readonly.values()
                                 for item in items}
            checkout = []
            status = gitutils.status(worktree)
            for head_path, data in status.items():
                if data["code"] not in {"DD", "AU", "UD", "UA", "DU", "AA", "UU"}:
                    # Only try to reset merge conflicts
                    continue
                path = data["rename"] if data["rename"] else head_path
                if path not in affected_metadata:
                    logger.debug("Resetting changes to %s" % head_path)
                    if data["code"] == "DU":
                        # Files that were deleted in master should just be removed
                        worktree.git.rm(head_path)
                    else:
                        checkout.append(head_path)
            logger.debug("Resetting changes to %s" % " ".join(checkout))
            try:
                worktree.git.checkout("HEAD", "--", *checkout)
                # Now try to commit again
                worktree.git.commit(c=sync.metadata_commit.sha1, no_edit=True)
                success = True
            except git.GitCommandError as e:
                if gitutils.handle_empty_commit(worktree, e):
                    return
                if sync.skip:
                    return
                success = False

        if not success:
            try:
                logger.info("Cherry-pick had merge conflicts trying to automatically resolve")
                status = gitutils.status(worktree)
                for head_path, data in status.items():
                    if data["code"] in {"DD", "UD", "DU"}:
                        # Deleted by remote or local
                        # Could do better here and have the mergetool handle this case
                        logger.info("Removing %s which was deleted somewhere" % head_path)
                        worktree.git.rm(head_path)
                    if data["code"] in {"UA", "AU"}:
                        logger.info("Adding %s which was added somewhere" % head_path)
                        worktree.git.add(head_path)
                logger.info("Running mergetool")
                worktree.git.mergetool(tool="metamerge",
                                       env={"MOZBUILD_STATE_PATH":
                                            repos.Gecko.get_state_path(env.config,
                                                                       worktree.working_dir)})
                worktree.git.commit(c=sync.metadata_commit.sha1, no_edit=True)
                worktree.git.clean(f=True)
                success = True
            except git.GitCommandError as e:
                if gitutils.handle_empty_commit(worktree, e):
                    return
                if sync.skip:
                    return
                logger.error("Failed trying to use mergetool to resolve conflicts")
                raise

        metadata_commit = sync_commit.GeckoCommit(worktree, worktree.head.commit)
        if metadata_commit.msg.startswith(b"Bug None"):
            # If the metadata commit didn't get a valid bug number for some reason,
            # we want to replace the placeholder bug number with the
            # either the sync or landing bug number, otherwise the push will be
            # rejected
            bug_number = sync.bug or self.bug
            new_message = b"Bug %s%s" % (str(bug_number).encode("utf8"),
                                         metadata_commit.msg[len(b"Bug None"):])
            sync_commit.create_commit(worktree, new_message, amend=True)

    @mut()
    def apply_prs(self, prev_wpt_head: str, landable_commits: LandableCommits) -> None:
        """Main entry point to setting the commits for landing.

        For each upstream PR we want to create a separate commit in the
        gecko repository so that we are preserving a useful subset of the history.
        We also want to prevent divergence from upstream. So for each PR that landed
        upstream since our last sync, we take the following steps:
        1) Copy the state of upstream at the commit where the PR landed over to
           the gecko repo
        2) Reapply any commits that have been made to gecko on the integration branch
           but which are not yet landed upstream on top of the PR
        3) Apply any updated metadata from the downstream sync for the PR.
        """

        last_pr = None
        has_metadata = False
        have_prs = set()
        # Find the last gecko commit containing a PR
        if len(self.gecko_commits):
            head_commit = self.gecko_commits.head
            if TYPE_CHECKING:
                head = cast(GeckoCommit, head_commit)
            else:
                head = head_commit
            if head.is_landing:
                return

            for commit in list(self.gecko_commits):
                assert isinstance(commit, GeckoCommit)
                if commit.metadata.get("wpt-pr") is not None:
                    last_pr = int(commit.metadata["wpt-pr"])
                    has_metadata = commit.metadata.get("wpt-type") == "metadata"
                    have_prs.add(last_pr)

        pr_count_applied = len(have_prs)

        gecko_commits_landed = set()

        def update_gecko_landed(sync: DownstreamSync | UpstreamSync,
                                commits: list[WptCommit]) -> None:
            if isinstance(sync, upstream.UpstreamSync):
                for commit in commits:
                    gecko_commit = commit.metadata.get("gecko-commit")
                    if gecko_commit:
                        gecko_commits_landed.add(gecko_commit)

        unapplied_commits = []
        pr_count_upstream_empty = 0
        last_applied_seen = last_pr is None
        for i, (pr, sync, commits) in enumerate(landable_commits):
            if last_applied_seen:
                unapplied_commits.append((i, (pr, sync, commits, False)))
            else:
                prev_wpt_head = commits[-1].sha1
                try:
                    have_prs.remove(pr)
                except KeyError:
                    if isinstance(sync, downstream.DownstreamSync):
                        # This could be wrong if the changes already landed in gecko for some reason
                        raise AbortError("Expected an existing gecko commit for PR %s, "
                                         "but not found" % (pr,))
                    pr_count_upstream_empty += 1
                    continue
                if pr == last_pr:
                    last_applied_seen = True
                    if not has_metadata:
                        unapplied_commits.append((i, (pr, sync, commits, True)))
            update_gecko_landed(sync, commits)

        if have_prs:
            raise AbortError("Found unexpected gecko commit for PRs %s"
                             % (", ".join(str(item) for item in have_prs),))

        pr_count_unapplied = len(unapplied_commits)
        if pr_count_applied and not has_metadata:
            # If we have seen the commit but not the metadata it will both be in
            # have_prs and unapplied_commits, so avoid double counting
            pr_count_unapplied -= 1

        if (pr_count_applied + pr_count_upstream_empty + pr_count_unapplied !=
            len(landable_commits)):
            raise AbortError("PR counts don't match; got %d applied, %d unapplied %d upstream"
                             "(total %s), expected total %d" %
                             (pr_count_applied,
                              pr_count_unapplied,
                              pr_count_upstream_empty,
                              pr_count_unapplied + pr_count_applied + pr_count_upstream_empty,
                              len(landable_commits)))

        for i, (pr, sync, commits, meta_only) in unapplied_commits:
            logger.info("Applying PR %i of %i" % (i + 1, len(landable_commits)))
            update_gecko_landed(sync, commits)

            # If copy is set then we copy the commits and reapply in-progress upstream
            # syncs. This is currently always disabled, but the intent was to do this for
            # the first commit to ensure that the possible drift from upstream was limited.
            # However there were some difficulties reapplying all the right commits, so it's
            # disabled until this is worked out.
            # To reenable it change the below line to
            # copy = i == 0
            copy = False
            pr_commit: Commit | None = None
            if not meta_only:
                # If we haven't applied it before then create the initial commit
                pr_commit = self.add_pr(pr, sync, commits, prev_wpt_head=prev_wpt_head,
                                        copy=copy)
            prev_wpt_head = commits[-1].sha1
            if pr_commit:
                if copy:
                    self.reapply_local_commits(gecko_commits_landed)
            if isinstance(sync, downstream.DownstreamSync):
                self.add_metadata(sync)

    @mut()
    def update_landing_commit(self) -> GeckoCommit:
        git_work = self.gecko_worktree.get()
        if not self.landing_commit:
            metadata = {
                "wpt-type": "landing",
                "wpt-head": self.wpt_commits.head.sha1
            }
            msg = sync_commit.Commit.make_commit_msg(
                b"""Bug %s - [wpt-sync] Update web-platform-tests to %s, a=testonly

MANUAL PUSH: wpt sync bot
                """ %
                (str(self.bug).encode("utf8"), self.wpt_commits.head.sha1.encode("utf8")),
                metadata)
            sync_commit.create_commit(git_work, msg, allow_empty=True)
        else:
            sync_commit.create_commit(git_work,
                                      self.landing_commit.msg,
                                      allow_empty=True,
                                      amend=True,
                                      no_edit=True)
        rv = self.gecko_commits[-1]
        assert isinstance(rv, GeckoCommit)
        return rv

    @mut()
    def update_bug_components(self) -> None:
        renames = self.wpt_renames()
        if renames is None:
            return

        gecko_work = self.gecko_worktree.get()
        mozbuild_path = bugcomponents.mozbuild_path(gecko_work)
        if not os.path.exists(mozbuild_path):
            return

        bugcomponents.update(gecko_work, renames)

        if gecko_work.is_dirty(path=mozbuild_path):
            gecko_work.git.add(mozbuild_path, all=True)
            self.update_landing_commit()

    @mut()
    def update_metadata(self, log_files: list, update_intermittents: bool = False) -> None:
        """Update the web-platform-tests metadata based on the logs
        generated in a try run.

        :param log_files: List of paths to the raw logs from the try run
        """
        # TODO: this shares a lot of code with downstreaming
        meta_path = env.config["gecko"]["path"]["meta"]

        gecko_work = self.gecko_worktree.get()
        mach = Mach(gecko_work.working_dir)
        logger.info("Updating metadata from %s logs" % len(log_files))
        args = ["--full"]
        if update_intermittents:
            args.append("--update-intermittent")
        args.extend(log_files)
        mach.wpt_update(*args)

        if gecko_work.is_dirty(untracked_files=True, path=meta_path):
            gecko_work.git.add(meta_path, all=True)
            self.update_landing_commit()
            gecko_work.git.reset(hard=True)

    @mut()
    def update_sync_point(self, sync_point: SyncPoint) -> None:
        """Update the in-tree record of the last sync point."""
        new_sha1 = self.wpt_commits.head.sha1
        if sync_point["upstream"] == new_sha1:
            return
        sync_point["upstream"] = new_sha1
        gecko_work = self.gecko_worktree.get()
        with open(os.path.join(gecko_work.working_dir,
                               env.config["gecko"]["path"]["meta"],
                               "mozilla-sync"), "wb") as f:
            sync_point.dump(f)
        if gecko_work.is_dirty():
            gecko_work.index.add([os.path.join(env.config["gecko"]["path"]["meta"],
                                               "mozilla-sync")])
            self.update_landing_commit()

    @mut()
    def next_try_push(self, retry: bool = False) -> TryPush | None:
        if self.status != "open":
            return None

        latest_try_push = self.latest_try_push
        stability = False

        if latest_try_push:
            if latest_try_push.status != "complete":
                return None
            elif latest_try_push.stability and not retry:
                return None

        if retry:
            stability = latest_try_push.stability if latest_try_push is not None else False
        else:
            stability = (latest_try_push is not None and
                         not latest_try_push.infra_fail)

        return trypush.TryPush.create(
            self._lock,
            self,
            hacks=False,
            stability=stability,
            rebuild_count=0,
            try_cls=trypush.TryFuzzyCommit,
            disable_target_task_filter=True,
            artifact=not stability,
            queries=["web-platform-tests !ccov !shippable",
                     "web-platform-tests linux-32 shippable",
                     "web-platform-tests mac !debug shippable"])

    def try_result(self, try_push: TryPush | None = None,
                   tasks: TryPushTasks | None = None) -> TryPushResult:
        """Determine whether a try push has infra failures, or an acceptable
        level of test passes for the current build"""
        if try_push is None:
            try_push = self.latest_try_push
        if try_push is None:
            raise ValueError("No try push found")
        target_success_rate = 0.5 if not try_push.stability else 0.8

        if try_push.infra_fail and not try_push.accept_failures:
            return TryPushResult.infra_fail
        if tasks is None:
            tasks = try_push.tasks()
        if tasks is None:
            # This can happen if the taskgroup_id is not yet set
            return TryPushResult.pending
        if not tasks.complete(allow_unscheduled=True):
            return TryPushResult.pending
        if tasks.success():
            return TryPushResult.success
        if tasks.failed_builds() and not try_push.accept_failures:
            return TryPushResult.infra_fail
        if (tasks.failure_limit_exceeded(target_success_rate) and
            not try_push.accept_failures):
            return TryPushResult.too_many_failures
        return TryPushResult.acceptable_failures


def push(landing: LandingSync) -> None:
    """Push from git_work_gecko to inbound."""
    success = False

    landing_tree = env.config["gecko"]["landing"]

    old_head = None
    err = None
    assert landing.bug is not None
    while not success:
        try:
            logger.info("Rebasing onto %s" % landing.gecko_integration_branch())
            landing.gecko_rebase(landing.gecko_integration_branch())
        except AbortError as e:
            logger.error(e)
            env.bz.comment(landing.bug, str(e))
            raise e

        if old_head == landing.gecko_commits.head.sha1:
            err = ("Landing push failed and rebase didn't change head:%s" %
                   ("\n%s" % err if err else ""))
            logger.error(err)
            env.bz.comment(landing.bug, err)
            raise AbortError(err)
        old_head = landing.gecko_commits.head.sha1

        if not tree.is_open(landing_tree):
            logger.info("%s is closed" % landing_tree)
            raise RetryableError(AbortError("Tree is closed"))

        try:
            logger.info("Pushing landing")
            push_info = landing.git_gecko.remotes.mozilla.push(
                "{}:{}".format(landing.branch_name,
                               landing.gecko_integration_branch().split("/", 1)[1])
            )
            for item in push_info:
                if item.flags & item.ERROR:
                    raise AbortError(item.summary)
        except git.GitCommandError as e:
            changes = landing.git_gecko.remotes.mozilla.fetch()
            err = "Pushing update to remote failed:\n%s" % e
            if not changes:
                logger.error(err)
                env.bz.comment(landing.bug, err)
                raise AbortError(err)
        else:
            success = True
    # The landing is marked as finished when it reaches central


def unlanded_with_type(git_gecko, git_wpt, wpt_head, prev_wpt_head):
    pr_commits = unlanded_wpt_commits_by_pr(git_gecko,
                                            git_wpt,
                                            wpt_head or prev_wpt_head,
                                            "origin/master")
    for pr, commits in pr_commits:
        if pr is None:
            status = LandableStatus.no_pr
        else:
            sync = load.get_pr_sync(git_gecko, git_wpt, pr, log=False)
            if sync is None:
                status = LandableStatus.no_sync
            elif isinstance(sync, upstream.UpstreamSync):
                status = LandableStatus.upstream
            else:
                assert isinstance(sync, downstream.DownstreamSync)
                status = sync.landable_status
        yield (pr, commits, status)


def load_sync_point(git_gecko: Repo, git_wpt: Repo) -> SyncPoint:
    """Read the last sync point from the batch sync process"""
    pygit2_repo = pygit2_get(git_gecko)
    integration_sha = pygit2_repo.revparse_single(LandingSync.gecko_integration_branch()).id
    blob_id = pygit2_repo[integration_sha].tree["testing/web-platform/meta/mozilla-sync"].id
    mozilla_data = pygit2_repo[blob_id].data
    sync_point = SyncPoint()
    sync_point.loads(mozilla_data)
    return sync_point


def unlanded_wpt_commits_by_pr(git_gecko: Repo,
                               git_wpt: Repo,
                               prev_wpt_head: str,
                               wpt_head: str = "origin/master",
                               ) -> list[tuple[int | None, list[WptCommit]]]:
    revish = f"{prev_wpt_head}..{wpt_head}"

    commits_by_pr: list[tuple[int | None, list[WptCommit]]] = []
    index_by_pr: dict[int | None, int] = {}

    for commit in git_wpt.iter_commits(revish,
                                       reverse=True,
                                       first_parent=True):
        wpt_commit = sync_commit.WptCommit(git_wpt, commit.hexsha)
        pr = wpt_commit.pr()
        extra_commits = []
        if pr not in index_by_pr:
            pr_data: tuple[int | None, list[WptCommit]] = (pr, [])
            # If we have a merge commit, also get the commits merged in
            if len(commit.parents) > 1:
                merged_revish = f"{commit.parents[0].hexsha}..{commit.hexsha}"
                for merged_commit in git_wpt.iter_commits(merged_revish,
                                                          reverse=True):
                    if merged_commit.hexsha != commit.hexsha:
                        wpt_commit = sync_commit.WptCommit(git_wpt, merged_commit.hexsha)
                        if wpt_commit.pr() == pr:
                            extra_commits.append(wpt_commit)
        else:
            idx = index_by_pr[pr]
            pr_data = commits_by_pr.pop(idx)
            assert pr_data[0] == pr
            index_by_pr = {key: (value if value < idx else value - 1)
                           for key, value in index_by_pr.items()}
        for c in extra_commits + [wpt_commit]:
            pr_data[1].append(c)
        commits_by_pr.append(pr_data)
        index_by_pr[pr] = len(commits_by_pr) - 1

    return commits_by_pr


def landable_commits(git_gecko: Repo,
                     git_wpt: Repo,
                     prev_wpt_head: str,
                     wpt_head: str | None = None,
                     include_incomplete: bool = False
                     ) -> tuple[str, LandableCommits] | None:
    """Get the list of commits that are able to land.

    :param prev_wpt_head: The sha1 of the previous wpt commit landed to gecko.
    :param wpt_head: The sha1 of the latest possible commit to land to gecko,
                     or None to use the head of the master branch
    :param include_incomplete: By default we don't attempt to land anything that
                               hasn't completed a metadata update. This flag disables
                               that and just lands everything up to the specified commit."""
    if wpt_head is None:
        wpt_head = "origin/master"
    pr_commits = unlanded_wpt_commits_by_pr(git_gecko, git_wpt, prev_wpt_head, wpt_head)
    landable_commits = []
    for pr, commits in pr_commits:
        last = False
        if not pr:
            # Assume this was some trivial fixup:
            continue

        first_commit = first_non_merge(commits)
        if not first_commit:
            # If we only have a merge commit just use that; it doesn't come from gecko anyway
            first_commit = commits[-1]

        def upstream_sync(bug_number):
            syncs = upstream.UpstreamSync.for_bug(git_gecko,
                                                  git_wpt,
                                                  bug_number,
                                                  flat=True)
            for sync in syncs:
                if sync.merge_sha == commits[-1].sha1 and not sync.wpt_commits:
                    # TODO: this shouldn't be mutating here
                    with SyncLock("upstream", None) as lock:
                        assert isinstance(lock, SyncLock)
                        with sync.as_mut(lock):
                            # If we merged with a merge commit, the set of commits
                            # here will be empty
                            sync.set_wpt_base(sync_commit.WptCommit(git_wpt,
                                                                    commits[0].sha1 + "~").sha1)

                # Only check the first commit since later ones could be added in the PR
                sync_revs = {item.canonical_rev for item in sync.upstreamed_gecko_commits}
                if any(commit.metadata.get("gecko-commit") in sync_revs for commit in commits):
                    break
            else:
                sync = None
            return sync

        sync = None
        sync = load.get_pr_sync(git_gecko, git_wpt, pr)
        if isinstance(sync, downstream.DownstreamSync):
            if sync and "affected-tests" in sync.data and sync.data["affected-tests"] is None:
                del sync.data["affected-tests"]
        if not include_incomplete:
            if not sync:
                # TODO: schedule a downstream sync for this pr
                logger.info("PR %s has no corresponding sync" % pr)
                last = True
            elif (isinstance(sync, downstream.DownstreamSync) and
                  sync.landable_status not in (LandableStatus.ready, LandableStatus.skip)):
                logger.info(f"PR {pr}: {sync.landable_status.reason_str()}")
                last = True
            if last:
                break
        assert isinstance(sync, (upstream.UpstreamSync, downstream.DownstreamSync))
        landable_commits.append((pr, sync, commits))

    if not landable_commits:
        logger.info("No new commits are landable")
        return None

    wpt_head = landable_commits[-1][2][-1].sha1
    logger.info("Landing up to commit %s" % wpt_head)

    return wpt_head, landable_commits


def current(git_gecko: Repo, git_wpt: Repo) -> LandingSync | None:
    landings = LandingSync.load_by_status(git_gecko, git_wpt, "open")
    if len(landings) > 1:
        raise ValueError("Multiple open landing branches")
    if landings:
        landing = landings.pop()
        assert isinstance(landing, LandingSync)
        return landing
    return None


@entry_point("landing")
def wpt_push(git_gecko: Repo, git_wpt: Repo, commits: list[str],
             create_missing: bool = True) -> None:
    prs = set()
    for commit_sha in commits:
        # This causes the PR to be recorded as a note
        commit = sync_commit.WptCommit(git_wpt, commit_sha)
        pr = commit.pr()
        if pr is not None and not upstream.UpstreamSync.has_metadata(commit.msg):
            prs.add(pr)
    if create_missing:
        for pr in prs:
            from . import update
            sync = load.get_pr_sync(git_gecko, git_wpt, pr)
            if not sync:
                # If we don't have a sync for this PR create one
                # It's easiest just to go via the GH API here
                pr_data = env.gh_wpt.get_pull(pr)
                update.update_pr(git_gecko, git_wpt, pr_data)


@entry_point("landing")
def update_landing(git_gecko: Repo,
                   git_wpt: Repo,
                   prev_wpt_head: Any | None = None,
                   new_wpt_head: Any | None = None,
                   include_incomplete: bool = False,
                   retry: bool = False,
                   allow_push: bool = True,
                   accept_failures: bool = False,
                   ) -> LandingSync | None:
    """Create or continue a landing of wpt commits to gecko.

    :param prev_wpt_head: The sha1 of the previous wpt commit landed to gecko.
    :param wpt_head: The sha1 of the latest possible commit to land to gecko,
                     or None to use the head of the master branch"
    :param include_incomplete: By default we don't attempt to land anything that
                               hasn't completed a metadata update. This flag disables
                               that and just lands everything up to the specified commit.
    :param retry: Create a new try push for the landing even if there's an existing one
    :param allow_push: Allow pushing to gecko if try is complete
    :param accept_failures: Don't fail if an existing try push has too many failures """
    landing = current(git_gecko, git_wpt)
    sync_point = load_sync_point(git_gecko, git_wpt)

    with SyncLock("landing", None) as lock:
        assert isinstance(lock, SyncLock)
        if landing is None:
            update_repositories(git_gecko, git_wpt)
            if prev_wpt_head is None:
                prev_wpt_head = sync_point["upstream"]

            landable = landable_commits(git_gecko, git_wpt,
                                        prev_wpt_head,
                                        wpt_head=new_wpt_head,
                                        include_incomplete=include_incomplete)
            if landable is None:
                return None
            wpt_head, commits = landable
            landing = LandingSync.new(lock, git_gecko, git_wpt, prev_wpt_head, wpt_head)

            # Set the landing to block all the bugs that will land with it
            blocks = [sync.bug for (pr_, sync, commits_) in commits
                      if isinstance(sync, downstream.DownstreamSync) and
                      sync.bug is not None]
            assert landing.bug is not None
            with env.bz.bug_ctx(landing.bug) as bug:
                for bug_id in blocks:
                    bug.add_blocks(bug_id)
        else:
            if prev_wpt_head and landing.wpt_commits.base.sha1 != prev_wpt_head:
                raise AbortError("Existing landing base commit %s doesn't match"
                                 "supplied previous wpt head %s" % (landing.wpt_commits.base.sha1,
                                                                    prev_wpt_head))
            elif new_wpt_head and landing.wpt_commits.head.sha1 != new_wpt_head:
                raise AbortError("Existing landing head commit %s doesn't match"
                                 "supplied wpt head %s" % (landing.wpt_commits.head.sha1,
                                                           new_wpt_head))
            head = landing.gecko_commits.head
            if git_gecko.is_ancestor(head.commit,
                                     git_gecko.rev_parse(
                                         env.config["gecko"]["refs"]["central"])):
                logger.info("Landing reached central")
                with landing.as_mut(lock):
                    landing.finish()
                return None
            elif git_gecko.is_ancestor(head.commit,
                                       git_gecko.rev_parse(landing.gecko_integration_branch())):
                logger.info("Landing is on inbound but not yet on central")
                return None

            landable = landable_commits(git_gecko,
                                        git_wpt,
                                        landing.wpt_commits.base.sha1,
                                        landing.wpt_commits.head.sha1,
                                        include_incomplete=include_incomplete)
            if landable is None:
                raise AbortError("No new commits are landable")
            wpt_head, commits = landable
            assert wpt_head == landing.wpt_commits.head.sha1

        pushed = False

        with landing.as_mut(lock):
            if landing.latest_try_push is None:
                landing.apply_prs(prev_wpt_head, commits)

                landing.update_bug_components()

                landing.update_sync_point(sync_point)

                landing.next_try_push()
            elif retry:
                try:
                    landing.gecko_rebase(landing.gecko_landing_branch())
                except AbortError:
                    message = record_rebase_failure(landing)
                    raise AbortError(message)

                with landing.latest_try_push.as_mut(lock):
                    landing.latest_try_push.status = "complete"  # type: ignore
                landing.next_try_push(retry=True)
            else:
                try_push = landing.latest_try_push
                try_result = landing.try_result()
                if try_push.status == "complete" and (try_result.is_ok() or
                                                      accept_failures):
                    try:
                        landing.gecko_rebase(landing.gecko_landing_branch())
                    except AbortError:
                        message = record_rebase_failure(landing)
                        raise AbortError(message)

                    if landing.next_try_push() is None:
                        push_to_gecko(git_gecko, git_wpt, landing, allow_push)
                        pushed = True
                elif try_result == TryPushResult.pending:
                    logger.info("Existing try push %s is waiting for try results" %
                                try_push.treeherder_url)
                else:
                    logger.info("Existing try push %s requires manual fixup" %
                                try_push.treeherder_url)

        try_notify_downstream(commits, landing_is_complete=pushed)

        if pushed:
            retrigger()
    return landing


def retrigger() -> None:
    try:
        # Avoid circular import
        from . import tasks
        tasks.retrigger.apply_async()
    except OperationalError:
        logger.warning("Failed to retrigger blocked syncs")


@entry_point("landing")
@mut('try_push', 'sync')
def try_push_complete(git_gecko: Repo,
                      git_wpt: Repo,
                      try_push: TryPush,
                      sync: LandingSync,
                      allow_push: bool = True,
                      accept_failures: bool = False,
                      tasks: Any | None = None,
                      ) -> None:
    """Run after all jobs in a try push are complete.

    This function handles updating the metadata based on the try push, or scheduling
    more jobs. In the case that the metadata has been updated successfully, the try
    push is marked as complete. If there's an error e.g. an infrastructure failure
    the try push is not marked as complete; user action is required to complete the
    handling of the try push (either by passing in accept_failures=True to indicate
    that the failure is not significant or by retyring the try push in which case the
    existing one will be marked as complete)."""

    if try_push.status == "complete":
        logger.warning("Called try_push_complete on a completed try push")
        return None

    if accept_failures:
        try_push.accept_failures = True  # type: ignore

    if tasks is None:
        tasks = try_push.tasks()

    if tasks is None:
        logger.error("Taskgroup id is not yet set")
        return None

    try_result = sync.try_result(tasks=tasks)

    if try_result == TryPushResult.pending:
        logger.info("Try push results are pending")
        return None

    if not try_result == TryPushResult.success:
        if try_result.is_failure():
            if try_result == TryPushResult.infra_fail:
                message = record_build_failures(sync, try_push)
                try_push.infra_fail = True  # type: ignore
                raise AbortError(message)
            elif try_result == TryPushResult.too_many_failures and not try_push.stability:
                message = record_too_many_failures(sync, try_push)
                raise AbortError(message)

        if not try_push.stability:
            update_metadata(sync, try_push)
        else:
            retriggered = tasks.retriggered_wpt_states()
            if not retriggered:
                if try_result == TryPushResult.too_many_failures:
                    record_too_many_failures(sync, try_push)
                    try_push.status = "complete"  # type: ignore
                    return None
                num_new_jobs = tasks.retrigger_failures()
                logger.info(f"{num_new_jobs} new tasks scheduled on try for {sync.bug}")
                if num_new_jobs:
                    assert sync.bug is not None
                    env.bz.comment(sync.bug,
                                   ("Retriggered failing web-platform-test tasks on "
                                    "try before final metadata update."))
                    return None

            update_metadata(sync, try_push, tasks)

    try_push.status = "complete"  # type: ignore

    if try_result == TryPushResult.infra_fail:
        record_infra_fail(sync, try_push)
        return None

    update_landing(git_gecko, git_wpt, allow_push=allow_push)


def needinfo_users() -> list[str]:
    needinfo_users = [item.strip() for item in
                      (env.config["gecko"]["needinfo"]
                       .get("landing", "")
                       .split(","))]
    return [item for item in needinfo_users if item]


def record_failure(sync: LandingSync, log_msg: str, bug_msg: str,
                   fixup_msg: Any | None = None) -> str:
    if fixup_msg is None:
        fixup_msg = "Run `wptsync landing` with either --accept-failures or --retry"
    logger.error(f"Bug {sync.bug}:{log_msg}\n{fixup_msg}")
    sync.error = log_msg  # type: ignore
    assert sync.bug is not None
    with env.bz.bug_ctx(sync.bug) as bug:
        bug.add_comment(f"{bug_msg}\nThis requires fixup from a wpt sync admin.")
        bug.needinfo(*needinfo_users())
    return log_msg


def record_build_failures(sync, try_push):
    log_msg = f"build failures in try push {try_push.treeherder_url}"
    bug_msg = f"Landing failed due to build failures in try push {try_push.treeherder_url}"
    return record_failure(sync, log_msg, bug_msg)


def record_too_many_failures(sync: LandingSync, try_push: TryPush) -> str:
    log_msg = f"too many test failures in try push {try_push.treeherder_url}"
    bug_msg = "Landing failed due to too many test failures in try push {}".format(
        try_push.treeherder_url)
    return record_failure(sync, log_msg, bug_msg)


def record_infra_fail(sync, try_push):
    log_msg = "infra failures in try push %s. " % (try_push.treeherder_url)
    bug_msg = "Landing failed due to infra failures in try push {}.".format(
        try_push.treeherder_url)
    return record_failure(sync, log_msg, bug_msg)


def record_rebase_failure(sync):
    log_msg = "rebase failed"
    bug_msg = "Landing failed due to conficts during rebase"
    fixup_msg = "Resolve the conflicts in the worktree and run `wptsync landing`"
    return record_failure(sync, log_msg, bug_msg, fixup_msg)


def update_metadata(sync: LandingSync, try_push: TryPush,
                    tasks: TryPushTasks | None = None) -> None:
    if tasks is None:
        tasks = try_push.tasks()
    if tasks is None:
        raise AbortError("Try push has no taskgroup id set")
    wpt_tasks = try_push.download_logs(tasks.wpt_tasks)
    log_files = []
    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:
        logger.warning("No log files found for try push %r" % try_push)
    sync.update_metadata(log_files, update_intermittents=True)


def push_to_gecko(git_gecko: Repo, git_wpt: Repo, sync: LandingSync,
                  allow_push: bool = True) -> None:
    if not allow_push:
        logger.info("Landing in bug %s is ready for push.\n"
                    "Working copy is in %s" % (sync.bug,
                                               sync.gecko_worktree.get().working_dir))
        return

    update_repositories(git_gecko, git_wpt)
    push(sync)


def try_notify_downstream(commits: Any, landing_is_complete: bool = False) -> None:
    for _, sync, _ in commits:
        if sync is not None:
            if isinstance(sync, downstream.DownstreamSync):
                with SyncLock.for_process(sync.process_name) as lock:
                    assert isinstance(lock, SyncLock)
                    with sync.as_mut(lock):
                        try:
                            if not sync.skip:
                                sync.try_notify()
                        except Exception as e:
                            logger.error(str(e))
                        finally:
                            if landing_is_complete:
                                sync.finish()
                                if sync.bug is not None and not sync.results_notified:
                                    env.bz.comment(sync.bug,
                                                   "Test result changes from PR not available.")


@entry_point("landing")
def gecko_push(git_gecko: Repo,
               git_wpt: Repo,
               repository_name: str,
               hg_rev: str,
               raise_on_error: bool = False,
               base_rev: Any | None = None,
               ) -> None:
    rev = git_gecko.rev_parse(cinnabar(git_gecko).hg2git(hg_rev))
    last_sync_point, base_commit = LandingSync.prev_gecko_commit(git_gecko,
                                                                 repository_name)

    if base_rev is None and git_gecko.is_ancestor(rev, base_commit.commit):
        logger.info("Last sync point moved past commit")
        return

    landed_central = repository_name == "mozilla-central"
    revish = f"{base_commit.sha1}..{rev.hexsha}"

    landing_sync = current(git_gecko, git_wpt)
    for commit in git_gecko.iter_commits(revish,
                                         reverse=True):
        gecko_commit = sync_commit.GeckoCommit(git_gecko, commit.hexsha)
        logger.debug("Processing commit %s" % gecko_commit.sha1)
        if landed_central and gecko_commit.is_landing:
            logger.info("Found wptsync landing in commit %s" % gecko_commit.sha1)
            if gecko_commit.bug is None:
                logger.error("Commit %s looked link a landing, but had no bug" %
                             gecko_commit.sha1)
                continue
            syncs = LandingSync.for_bug(git_gecko,
                                        git_wpt,
                                        gecko_commit.bug,
                                        statuses=None,
                                        flat=True)
            if syncs:
                sync = syncs[0]
                logger.info("Found sync %s" % sync.process_name)
                with SyncLock("landing", None) as lock:
                    assert isinstance(lock, SyncLock)
                    with syncs[0].as_mut(lock):
                        sync.finish()
            else:
                logger.error("Failed to find sync for commit")
        elif gecko_commit.is_backout:
            backed_out, _ = gecko_commit.landing_commits_backed_out()
            if backed_out:
                logger.info("Commit %s backs out wpt sync landings" % gecko_commit.sha1)
            for backed_out_commit in backed_out:
                bug = backed_out_commit.bug
                syncs = []
                if bug is not None:
                    syncs = LandingSync.for_bug(git_gecko, git_wpt, bug,
                                                statuses=None, flat=True)
                if syncs:
                    # TODO: should really check if commit is actually part of the sync if there's >1
                    # TODO: reopen landing? But that affects the invariant that there is only one
                    sync = syncs[0]
                    logger.info("Found sync %s" % sync.process_name)
                    with SyncLock("landing", None) as lock:
                        assert isinstance(lock, SyncLock)
                        with sync.as_mut(lock):
                            sync.error = "Landing was backed out"  # type: ignore
                else:
                    logger.error("Failed to find sync for commit")
        elif gecko_commit.is_downstream:
            syncs = []
            bug = gecko_commit.bug
            if bug is not None:
                syncs = LandingSync.for_bug(git_gecko, git_wpt, bug,
                                            statuses=None, flat=True)
            for sync in syncs:
                sync = syncs[0]
                with SyncLock("landing", None) as lock:
                    assert isinstance(lock, SyncLock)
                    with sync.as_mut(lock):
                        sync.finish()

    # TODO: Locking here
    with SyncLock("landing", None) as lock:
        assert isinstance(lock, SyncLock)
        with last_sync_point.as_mut(lock):
            assert last_sync_point.commit is not None
            if not git_gecko.is_ancestor(rev, last_sync_point.commit.commit):
                last_sync_point.commit = rev.hexsha  # type: ignore

    if landing_sync and landing_sync.status == "complete":
        start_next_landing()


def start_next_landing():
    from . import tasks
    tasks.land.apply_async()