from __future__ import annotations import enum import itertools import traceback from collections import defaultdict import git from . import bug from . import log from .base import ( BranchRefObject, CommitBuilder, IdentityMap, ProcessData, ProcessName, ProcessNameIndex, ) from .commit import GeckoCommit, WptCommit from .env import Environment from .errors import AbortError from .lock import MutGuard, mut, constructor from .repos import cinnabar from .worktree import Worktree from typing import (Any, Iterable, Iterator, Sequence, cast, TYPE_CHECKING) from git.repo.base import Repo from os import PathLike from typing_extensions import Literal if TYPE_CHECKING: from sync.commit import Commit from sync.index import Index from sync.lock import SyncLock from sync.trypush import TryPush try: from typing import (List, Tuple, Type, overload) except ImportError: def typing(f): return f env = Environment() logger = log.get_logger(__name__) class CommitFilter: """Filter of a range of commits""" def __init__(self) -> None: self._commits: dict[str, bool] = {} def path_filter(self) -> Sequence[PathLike]: """Path filter for the commit range, returning a list of paths that match. An empty list matches all paths.""" return [] def filter_commit(self, commit: Commit) -> bool: """Per-commit filter. :param commit: wpt_commit.Commit object :returns: A boolean indicating whether to include the commit""" if commit.sha1 not in self._commits: self._commits[commit.sha1] = self._filter_commit(commit) return self._commits[commit.sha1] def _filter_commit(self, commit: Commit) -> bool: return True def filter_commits( self, commits: Iterable[Commit], ) -> Sequence[Commit]: """Filter that applies to the set of commits that were selected by the per-commit filter. Useful for e.g. removing backouts from a set of commits. :param commits: List of wpt_commit.Commit objects :returns: List of commits to return""" return list(commits) class CommitRange: # This class should probably be generic in the commit subtype it returns # but trying that ended up with lots of errors, so for now it just specifies # Commit, meaning that users need to cast to use properies of subtypes """Range of commits in a specific repository. This is assumed to be between a tag and a branch sharing a single process_name i.e. the tag represents the base commit of the branch. TODO: Maybe just store the base branch name in the tag rather than making it an actual pointer since that works better with rebases. """ def __init__( self, repo: Repo, base: str | Commit, head_ref: BranchRefObject, commit_cls: type, commit_filter: CommitFilter, ) -> None: self.repo = repo # This ended up a little confused because these used to both be # VcsRefObjects, but now the base is stored as a ref not associated # with a process_name. This should be refactored. self._base_commit: Commit | None = None self._base = base self._head_ref = head_ref self.commit_cls = commit_cls self.commit_filter = commit_filter # Cache for the commits in this range self._commits: Sequence[Commit] = [] self._head_sha: str | None = None self._base_sha: str | None = None self._lock = None def as_mut(self, lock: SyncLock) -> MutGuard: return MutGuard(lock, self, [self._head_ref]) @property def lock_key(self) -> tuple[str, str]: return (self._head_ref.name.subtype, self._head_ref.name.obj_id) @overload def __getitem__( self, index: int ) -> Commit: pass @overload # noqa: F811 def __getitem__( self, index: slice ) -> Sequence[Commit]: pass def __getitem__( self, # noqa: F811 index: int | slice, ) -> Commit | Sequence[Commit]: return self.commits[index] def __iter__(self) -> Iterator[Commit]: yield from self.commits def __len__(self) -> int: return len(self.commits) def __contains__(self, other_commit: Any) -> bool: for commit in self: if commit == other_commit: return True return False @property def commits(self) -> Sequence[Commit]: if self._commits: if self.head.sha1 == self._head_sha and self.base.sha1 == self._base_sha: return self._commits revish = f"{self.base.sha1}..{self.head.sha1}" commits: list[Commit] = [] for git_commit in self.repo.iter_commits( revish, reverse=True, paths=self.commit_filter.path_filter() ): commit = self.commit_cls(self.repo, git_commit) if not self.commit_filter.filter_commit(commit): continue commits.append(commit) commits = self.commit_filter.filter_commits(commits) # type: ignore self._commits = commits self._head_sha = self.head.sha1 self._base_sha = self.base.sha1 return self._commits @property def files_changed(self) -> set[str]: # We avoid using diffs because that's harder to get right in the face of merges files = set() for commit in self.commits: commit_files = self.repo.git.show(commit.sha1, name_status=True, format="") for item in commit_files.splitlines(): parts = item.split("\t") for part in parts[1:]: files.add(part.strip()) return files @property def base(self) -> Commit: if self._base_commit is None: self._base_commit = self.commit_cls(self.repo, self._base) assert self._base_commit is not None return self._base_commit @base.setter # type: ignore @mut() def base(self, value: str) -> None: # Note that this doesn't actually update the stored value of the base # anywhere, unlike the head setter which will update the associated ref self._commits = [] self._base_sha = value self._base = self.commit_cls(self.repo, value) self._base_commit = None @property def head(self) -> Commit: head_commit = self._head_ref.commit assert head_commit is not None if TYPE_CHECKING: cast(Commit, head_commit) return head_commit @head.setter # type: ignore @mut() def head(self, value: Commit) -> None: self._head_ref.commit = value # type: ignore @enum.unique class LandableStatus(enum.Enum): ready = 0 no_pr = 1 upstream = 2 no_sync = 3 error = 4 missing_try_results = 5 skip = 6 def reason_str(self) -> str: return { LandableStatus.ready: "Ready", LandableStatus.no_pr: "No PR", LandableStatus.upstream: "From gecko", LandableStatus.no_sync: "No sync created", LandableStatus.error: "Error", LandableStatus.missing_try_results: "Incomplete try results", LandableStatus.skip: "Skip", }.get(self, "Unknown") class SyncPointName(metaclass=IdentityMap): """Like a process name but for pointers that aren't associated with a specific sync object, but with a general process e.g. the last update point for an upstream sync.""" def __init__(self, subtype: str, obj_id: str) -> None: self._obj_type = "sync" self._subtype = subtype self._obj_id = str(obj_id) self._lock = None @property def obj_type(self) -> str: return self._obj_type @property def subtype(self) -> str: return self._subtype @property def obj_id(self) -> str: return self._obj_id @classmethod def _cache_key(cls, subtype: str, obj_id: str) -> tuple[str, str]: return (subtype, str(obj_id)) def key(self) -> tuple[str, str]: return (self._subtype, self._obj_id) def __str__(self) -> str: return f"{self._obj_type}/{self._subtype}/{self._obj_id}" def path(self): return f"{self._obj_type}/{self._subtype}/{self._obj_id}" class SyncData(ProcessData): obj_type = "sync" class SyncProcess(metaclass=IdentityMap): obj_type: str = "sync" sync_type: str = "*" # Either "bug" or "pr" obj_id: str | None = None statuses: tuple[str, ...] = () status_transitions: list[tuple[str, str]] = [] # Can multiple syncs have the same obj_id multiple_syncs: bool = False def __init__(self, git_gecko: Repo, git_wpt: Repo, process_name: ProcessName) -> None: self._lock: SyncLock | None = None assert process_name.obj_type == self.obj_type assert process_name.subtype == self.sync_type self.git_gecko = git_gecko self.git_wpt = git_wpt self.process_name = process_name self.data = SyncData(git_gecko, process_name) self.gecko_commits = CommitRange( git_gecko, self.data["gecko-base"], BranchRefObject(git_gecko, self.process_name, commit_cls=GeckoCommit), commit_cls=GeckoCommit, commit_filter=self.gecko_commit_filter(), ) self.wpt_commits = CommitRange( git_wpt, self.data["wpt-base"], BranchRefObject(git_wpt, self.process_name, commit_cls=WptCommit), commit_cls=WptCommit, commit_filter=self.wpt_commit_filter(), ) self.gecko_worktree = Worktree(git_gecko, process_name) self.wpt_worktree = Worktree(git_wpt, process_name) # Hold onto indexes for the lifetime of the SyncProcess object self._indexes = {ProcessNameIndex(git_gecko)} @classmethod def _cache_key(cls, git_gecko: Repo, git_wpt: Repo, process_name: ProcessName) -> tuple[str, str, str, str]: return process_name.key() def as_mut(self, lock: SyncLock) -> MutGuard: return MutGuard( lock, self, [ self.data, self.gecko_commits, self.wpt_commits, self.gecko_worktree, self.wpt_worktree, ], ) @property def lock_key(self) -> tuple[str, str]: return (self.process_name.subtype, self.process_name.obj_id) def __repr__(self) -> str: return "<{} {} {}>".format( self.__class__.__name__, self.sync_type, self.process_name ) @classmethod def for_pr( cls, git_gecko: Repo, git_wpt: Repo, pr_id: str | int, ) -> SyncProcess | None: from . import index idx = index.PrIdIndex(git_gecko) process_name = idx.get((str(pr_id),)) if process_name and process_name.subtype == cls.sync_type: return cls(git_gecko, git_wpt, process_name) return None @overload # noqa: F811 @classmethod def for_bug( cls, git_gecko: Repo, git_wpt: Repo, bug: int, statuses: Iterable[str] | None, flat: Literal[True], ) -> list[SyncProcess]: pass @overload # noqa: F811 @classmethod def for_bug( cls, git_gecko: Repo, git_wpt: Repo, bug: int, statuses: Iterable[str] | None, flat: Literal[False], ) -> dict[str, set[SyncProcess]]: pass @classmethod # noqa: F811 def for_bug( cls, git_gecko: Repo, git_wpt: Repo, bug: int, statuses: Iterable[str] | None = None, flat: bool = False, ) -> dict[str, set[SyncProcess]] | list[SyncProcess]: """Get the syncs for a specific bug. :param bug: The bug number for which to find syncs. :param statuses: An optional list of sync statuses to include. Defaults to all statuses. :param flat: Return a flat list of syncs instead of a dictionary. :returns: By default a dictionary of {status: [syncs]}, but if flat is true, just returns a list of matching syncs. """ from . import index bug_str = str(bug) statuses = set(statuses) if statuses is not None else set(cls.statuses) rv = defaultdict(set) idx_key = (bug_str,) if len(statuses) == 1: idx_key == (bug_str, list(statuses)[0]) idx = index.BugIdIndex(git_gecko) process_names = idx.get(idx_key) for process_name in process_names: if process_name.subtype == cls.sync_type: sync = cls(git_gecko, git_wpt, process_name) if sync.status in statuses: rv[sync.status].add(sync) if flat: return list(itertools.chain.from_iterable(rv.values())) return rv @classmethod def load_by_obj(cls, git_gecko: Repo, git_wpt: Repo, obj_id: int, seq_id: int | None = None) -> set[SyncProcess]: process_names = ProcessNameIndex(git_gecko).get( cls.obj_type, cls.sync_type, str(obj_id) ) if seq_id is not None: process_names = {item for item in process_names if item.seq_id == seq_id} return {cls(git_gecko, git_wpt, process_name) for process_name in process_names} @classmethod def load_by_status(cls, git_gecko: Repo, git_wpt: Repo, status: str) -> set[SyncProcess]: from . import index idx = index.SyncIndex(git_gecko) key = (cls.obj_type, cls.sync_type, status) process_names = idx.get(key) rv = set() for process_name in process_names: rv.add(cls(git_gecko, git_wpt, process_name)) return rv # End of getter methods @classmethod def prev_gecko_commit( cls, git_gecko: Repo, repository_name: str, base_rev: str | None = None, ) -> tuple[BranchRefObject, GeckoCommit]: """Get the last gecko commit processed by a sync process. :param str repository_name: The name of the gecko branch being processed :param base_rev: The SHA1 for a commit to use as the previous commit, overriding the stored value :returns: Tuple of (LastSyncPoint, GeckoCommit) for the previous gecko commit. In case the base_rev override is passed in the LastSyncPoint may be pointing at a different commit to the returned gecko commit""" last_sync_point = cls.last_sync_point(git_gecko, repository_name) assert last_sync_point.commit is not None logger.info("Last sync point was %s" % last_sync_point.commit.sha1) if base_rev is None: prev_commit = last_sync_point.commit # TODO: Would be good if we could link the repo class to the commit type # in the type system assert isinstance(prev_commit, GeckoCommit) else: prev_commit = GeckoCommit(git_gecko, cinnabar(git_gecko).hg2git(base_rev)) return last_sync_point, prev_commit @classmethod def last_sync_point(cls, git_gecko: Repo, repository_name: str) -> BranchRefObject: assert "/" not in repository_name name = SyncPointName(cls.sync_type, repository_name) return BranchRefObject(git_gecko, name, commit_cls=GeckoCommit) @property def landable_status(self) -> LandableStatus: raise NotImplementedError def _output_data(self) -> list[str]: rv = [ "{}{}".format("*" if self.error else " ", self.process_name.path()), "gecko range: {}..{}".format( self.gecko_commits.base.sha1, self.gecko_commits.head.sha1 ), "wpt range: {}..{}".format( self.wpt_commits.base.sha1, self.wpt_commits.head.sha1 ), ] if self.error: rv.extend( ["ERROR:", self.error["message"] or "", self.error["stack"] or ""] ) landable_status = self.landable_status if landable_status: rv.append("Landable status: %s" % landable_status.reason_str()) for key, value in sorted(self.data.items()): if key != "error": rv.append(f"{key}: {value}") try_pushes = self.try_pushes() if try_pushes: try_pushes = sorted(try_pushes, key=lambda x: x.process_name.seq_id) rv.append("Try pushes:") for try_push in try_pushes: rv.append(f" {try_push.status} {try_push.treeherder_url}") return rv def output(self) -> str: return "\n".join(self._output_data()) def __ne__(self, other: Any) -> bool: return not self == other def set_wpt_base(self, ref: str) -> None: # This is kind of an appaling hack try: self.git_wpt.commit(ref) except Exception: raise ValueError self.data["wpt-base"] = ref self.wpt_commits._base = WptCommit(self.git_wpt, ref) @staticmethod def gecko_integration_branch() -> str: return env.config["gecko"]["refs"][env.config["gecko"]["landing"]] @staticmethod def gecko_landing_branch() -> str: return env.config["gecko"]["refs"]["central"] def gecko_commit_filter(self) -> CommitFilter: return CommitFilter() def wpt_commit_filter(self) -> CommitFilter: return CommitFilter() @property def branch_name(self) -> str: return self.process_name.path() @property def status(self) -> str: return self.data["status"] @status.setter # type: ignore @mut() def status(self, value: str) -> None: if value not in self.statuses: raise ValueError("Unrecognised status %s" % value) current = self.status if current == value: return if (current, value) not in self.status_transitions: raise ValueError( f"Tried to change status from {current} to {value}" ) from . import index index.SyncIndex(self.git_gecko).delete( index.SyncIndex.make_key(self), self.process_name ) index.BugIdIndex(self.git_gecko).delete( index.BugIdIndex.make_key(self), self.process_name ) self.data["status"] = value index.SyncIndex(self.git_gecko).insert( index.SyncIndex.make_key(self), self.process_name ) index.BugIdIndex(self.git_gecko).insert( index.BugIdIndex.make_key(self), self.process_name ) @property def bug(self) -> int | None: if self.obj_id == "bug": return int(self.process_name.obj_id) else: bug = self.data.get("bug") if bug: return int(bug) return None @bug.setter # type: ignore @mut() def bug(self, value: int) -> None: from . import index if self.obj_id == "bug": raise AttributeError("Can't set attribute") old_key = None if self.data.get("bug"): old_key = index.BugIdIndex.make_key(self) self.data["bug"] = value new_key = index.BugIdIndex.make_key(self) index.BugIdIndex(self.git_gecko).move(old_key, new_key, self.process_name) @property def pr(self) -> int | None: if self.obj_id == "pr": return int(self.process_name.obj_id) else: pr = self.data.get("pr") if pr is not None: return int(pr) return None @pr.setter # type: ignore @mut() def pr(self, value: int) -> None: from . import index if self.obj_id == "pr": raise AttributeError("Can't set attribute") old_key = None if self.data.get("pr"): old_key = index.PrIdIndex.make_key(self) self.data["pr"] = value new_key = index.PrIdIndex.make_key(self) index.PrIdIndex(self.git_gecko).move(old_key, new_key, self.process_name) @property def seq_id(self) -> int: return self.process_name.seq_id @property def last_pr_check(self) -> dict[str, str]: return self.data.get("last-pr-check", {}) @last_pr_check.setter # type: ignore @mut() def last_pr_check(self, value: dict[str, str]) -> None: if value is not None: self.data["last-pr-check"] = value else: del self.data["last-pr-check"] @property def error(self) -> dict[str, str | None] | None: return self.data.get("error") @error.setter # type: ignore @mut() def error(self, value: str | None) -> None: def encode(item: str | None) -> str | None: if item is None: return item if isinstance(item, str): return item if isinstance(item, str): return item.encode("utf8", "replace") return repr(item) if value is not None: if isinstance(value, (str, str)): message = value stack = None else: message = str(value) stack = traceback.format_exc() error = {"message": encode(message), "stack": encode(stack)} self.data["error"] = error self.set_bug_data("error") else: del self.data["error"] self.set_bug_data(None) def try_pushes(self, status: str | None = None) -> list[TryPush]: from . import trypush try_pushes = trypush.TryPush.load_by_obj( self.git_gecko, self.sync_type, int(self.process_name.obj_id) ) # I tried cast(Set[TryPush], try_pushes) here but it didn't work if status is not None: try_pushes_for_status: set[TryPush] = set() for item in try_pushes: assert isinstance(item, trypush.TryPush) if item.status == status: try_pushes_for_status.add(item) else: try_pushes_for_status = try_pushes # type: ignore return list(sorted(try_pushes_for_status, key=lambda x: x.process_name.seq_id)) def latest_busted_try_pushes(self) -> list[TryPush]: try_pushes = self.try_pushes(status="complete") busted = [] for push in reversed(try_pushes): if push.infra_fail: busted.append(push) else: break return busted @property def latest_try_push(self) -> TryPush | None: try_pushes = self.try_pushes() if try_pushes: try_pushes = sorted(try_pushes, key=lambda x: x.process_name.seq_id) return try_pushes[-1] return None def wpt_renames(self) -> dict[str, str]: renames = {} diff_blobs = self.wpt_commits.head.commit.diff( self.git_wpt.merge_base(self.data["wpt-base"], self.wpt_commits.head.sha1)[0] ) for item in diff_blobs: if item.rename_from: renames[item.rename_from] = item.rename_to return renames @classmethod @constructor( lambda args: ( args["cls"].sync_type, args["bug"] if args["cls"].obj_id == "bug" else str(args["pr"]), ) ) def new( cls, lock: SyncLock, git_gecko: Repo, git_wpt: Repo, gecko_base: str, gecko_head: str, wpt_base: str = "origin/master", wpt_head: str | None = None, bug: int | None = None, pr: int | None = None, status: str = "open", ): # type(...) -> SyncProcess # TODO: this object creation is extremely non-atomic :/ from . import index if cls.obj_id == "bug": assert bug is not None obj_id = str(bug) elif cls.obj_id == "pr": assert pr is not None obj_id = str(pr) else: raise ValueError("Invalid cls.obj_id: %s" % cls.obj_id) if wpt_head is None: wpt_head = wpt_base data = { "gecko-base": gecko_base, "wpt-base": wpt_base, "pr": pr, "bug": bug, "status": status, } # This is pretty ugly process_name = ProcessName.with_seq_id( git_gecko, cls.obj_type, cls.sync_type, obj_id ) if not cls.multiple_syncs and process_name.seq_id != 0: raise ValueError( "Tried to create new {} sync for {} {} but one already exists".format( cls.obj_id, cls.sync_type, obj_id ) ) commit_builder = CommitBuilder(git_gecko, f"Create sync {process_name}\n", env.config["sync"]["ref"]) refs = [BranchRefObject.create(lock, git_gecko, process_name, gecko_head, GeckoCommit, force=True), BranchRefObject.create(lock, git_wpt, process_name, wpt_head, WptCommit, force=True)] try: # This will commit all the data in a single commit when we exit the "with" block with commit_builder: SyncData.create(lock, git_gecko, process_name, data, commit_builder=commit_builder) sync_idx_key = (process_name, status) idxs: List[Tuple[Any, Type[Index]]] = [(sync_idx_key, index.SyncIndex)] if cls.obj_id == "bug": idxs.append(((process_name, status), index.BugIdIndex)) elif cls.obj_id == "pr": idxs.append((process_name, index.PrIdIndex)) for (key, idx_cls) in idxs: idx = idx_cls(git_gecko) idx.insert(idx.make_key(key), process_name).save(commit_builder) except Exception: for ref in refs: ref.delete() raise return cls(git_gecko, git_wpt, process_name) @mut() def finish(self, status: str = "complete") -> None: # TODO: cancel related try pushes &c. logger.info(f"Marking sync {self.process_name} as {status}") self.status = status # type: ignore self.error = None # type: ignore for worktree in [self.gecko_worktree, self.wpt_worktree]: worktree.delete() for repo in [self.git_gecko, self.git_wpt]: repo.git.worktree("prune") @mut() def gecko_rebase(self, new_base_ref: str, abort_on_fail: bool = False) -> None: new_base = GeckoCommit(self.git_gecko, new_base_ref) git_worktree = self.gecko_worktree.get() set_new_base = True try: git_worktree.git.rebase(new_base.sha1) except git.GitCommandError as e: if abort_on_fail: set_new_base = False try: git_worktree.git.rebase(abort=True) except git.GitCommandError: pass raise AbortError("Rebasing onto latest gecko failed:\n%s" % e) finally: if set_new_base: self.data["gecko-base"] = new_base_ref self.gecko_commits.base = new_base.sha1 # type: ignore @mut() def wpt_rebase(self, ref: str) -> None: assert ref in self.git_wpt.references git_worktree = self.wpt_worktree.get() git_worktree.git.rebase(ref) self.set_wpt_base(ref) @mut() def set_bug_data(self, status: str | None = None) -> None: if self.bug: whiteboard = env.bz.get_whiteboard(self.bug) if not whiteboard: return current_subtype, current_status = bug.get_sync_data(whiteboard) if current_subtype != self.sync_type or current_status != status: new_whiteboard = bug.set_sync_data(whiteboard, self.sync_type, status) env.bz.set_whiteboard(self.bug, new_whiteboard) @mut() def delete(self) -> None: from . import index for worktree in [self.gecko_worktree, self.wpt_worktree]: worktree.delete() assert self._lock is not None for try_push in self.try_pushes(): with try_push.as_mut(self._lock): try_push.delete() for git_repo, commit_cls in [ (self.git_wpt, WptCommit), (self.git_gecko, GeckoCommit), ]: BranchRefObject(git_repo, self.process_name, commit_cls=commit_cls).delete() for idx_cls in [index.SyncIndex, index.PrIdIndex, index.BugIdIndex]: key = idx_cls.make_key(self) idx = idx_cls(self.git_gecko) idx.delete(key, self.process_name).save() self.data.delete()