datastore/tests/ci/cherry_pick.py

#!/usr/bin/env python3
"""
A plan:
    - TODO: consider receiving GH objects cache from S3, but it's really a few
    of requests to API currently
    - Get all open release PRs (20.10, 21.8, 22.5, etc.)
    - Get all pull-requests between the date of the merge-base for the oldest PR with
    labels pr-must-backport and version-specific v21.8-must-backport, but without
    pr-backported
    - Iterate over gotten PRs:
        - for pr-must-backport:
            - check if all backport-PRs are created. If yes,
            set pr-backported label and finish
            - If not, create either cherrypick PRs or merge cherrypick (in the same
            stage, if mergable) and create backport-PRs
            - If successful, set pr-backported label on the PR

        - for version-specific labels (e.g. v25.12-must-backport):
            - the label marks the OLDEST release the PR must reach. Backport to
            that release AND to every newer active release branch, then the same
            check, cherry-pick, backport, pr-backported

Cherry-pick stage:
    - From time to time the cherry-pick fails, if it was done manually. In the
    case we check if it's even needed, and mark the release as done somehow.

The cross-repo synchronization is described in the KB article:
https://github.com/ClickHouse/internal-knowledge-base/issues/452
"""

import argparse
import logging
import os
from datetime import date, datetime, timedelta
from pathlib import Path
from subprocess import CalledProcessError
from typing import Iterable, List, Optional

from github.GithubException import GithubException

from cache_utils import GitHubCache
from cherry_pick_branches import (
    branch_version,
    label_version,
    select_backport_branches,
)
from ci_buddy import CIBuddy
from ci_utils import Shell
from env_helper import (
    GITHUB_REPOSITORY,
    GITHUB_SERVER_URL,
    GITHUB_UPSTREAM_REPOSITORY,
    IS_CI,
    TEMP_PATH,
)
from get_robot_token import get_best_robot_token
from git_helper import GIT_PREFIX, git_runner, is_shallow, stash
from github_helper import GitHub, PullRequest, PullRequests, Repository
from pr_info import Labels
from report import GITHUB_JOB_URL
from s3_helper import S3Helper
from ssh import SSHKey
from synchronizer_utils import SYNC_PR_PREFIX


class BackportException(Exception):
    pass


def recover_git_state() -> None:
    """
    Best-effort recovery of the working tree after a git command crashed
    (e.g. an internal assertion in `merge-ort`) and left `.git/index.lock`
    behind. In that state subsequent commands -- including
    `git merge --abort` -- fail with "Unable to create '.git/index.lock'",
    which would otherwise poison every later PR processed in the same run.
    """
    try:
        # `--absolute-git-dir` -- avoid a relative `.git` resolved against
        # Python's cwd (which is `tests/ci/`, not the repo root).
        git_dir = git_runner("git rev-parse --absolute-git-dir")
    except CalledProcessError:
        return
    lock = Path(git_dir) / "index.lock"
    if lock.exists():
        logging.warning(
            "Removing stale %s left by a crashed git process", lock
        )
        try:
            lock.unlink()
        except OSError as e:
            logging.error("Failed to remove %s: %s", lock, e)
            return
    # Best-effort cleanup of any in-progress merge / cherry-pick and the
    # working tree. None of these are required to succeed -- they only run
    # to bring the tree back to a usable state for the next PR.
    for cmd in (
        f"{GIT_PREFIX} merge --abort",
        f"{GIT_PREFIX} cherry-pick --abort",
        f"{GIT_PREFIX} reset --hard HEAD",
    ):
        try:
            git_runner(cmd)
        except CalledProcessError as e:
            logging.info("recover_git_state: %s -> %s (ignored)", cmd, e)


class ReleaseBranch:
    CHERRYPICK_DESCRIPTION = f"""## Do not merge this PR manually

This pull-request is a first step of an automated backporting.
It contains changes similar to calling `git cherry-pick` locally.
If you intend to continue backporting the changes, then resolve all conflicts if any.
Otherwise, if you do not want to backport them, then just close this pull-request.

The check results does not matter at this step - you can safely ignore them.

### Troubleshooting

#### If the conflicts were resolved in a wrong way

If this cherry-pick PR is completely screwed by a wrong conflicts resolution, and you \
want to recreate it:

- delete the `{Labels.PR_CHERRYPICK}` label from the PR
- delete this branch from the repository

You also need to check the **Original pull-request** for `{Labels.PR_BACKPORTS_CREATED}` \
label, and  delete if it's presented there
"""
    BACKPORT_DESCRIPTION = """This pull-request is a last step of an automated \
backporting.
Treat it as a standard pull-request: look at the checks and resolve conflicts.
Merge it only if you intend to backport changes to the target branch, otherwise just \
close it.
"""
    PR_SOURCE_DESCRIPTION = ""
    REMOTE = ""

    @property
    def pr_source(self) -> str:
        if self.PR_SOURCE_DESCRIPTION:
            return self.PR_SOURCE_DESCRIPTION
        header = "\n\n### The PR source\n"
        if not IS_CI:
            self.PR_SOURCE_DESCRIPTION = (
                f"{header}The PR is created manually outside of the CI"
            )
        else:
            self.PR_SOURCE_DESCRIPTION = (
                f"{header}The PR is created in the [CI job]({GITHUB_JOB_URL()})"
            )

        return self.PR_SOURCE_DESCRIPTION

    def __init__(
        self,
        name: str,
        pr: PullRequest,
        repo: Repository,
    ):
        self.name = name
        self.pr = pr
        self.repo = repo

        self.cherrypick_branch = self.cp_branch(name, pr.number)
        self.backport_branch = self.bp_branch(name, pr.number)
        self.cherrypick_pr = None  # type: Optional[PullRequest]
        self.backport_pr = None  # type: Optional[PullRequest]
        self._backported = False

        self.pre_check()

    @staticmethod
    def cp_branch(name: str, pr_number: int) -> str:
        """
        Returns the name of the cherry-pick branch for the given release branch and PR
        number.
        """
        return f"cherrypick/{name}/{pr_number}"

    @staticmethod
    def bp_branch(name: str, pr_number: int) -> str:
        """
        Returns the name of the backport branch for the given release branch and PR
        number.
        """
        return f"backport/{name}/{pr_number}"

    def pre_check(self):
        self._backported = Shell.check(
            f"git merge-base --is-ancestor {self.pr.merge_commit_sha} {self.REMOTE}/{self.name}",
            verbose=True,
        )
        if self._backported:
            print(
                f"WARNING: Backport for PR [{self.pr}] is already present on {self.name}"
            )

    def pop_prs(self, prs: PullRequests) -> PullRequests:
        """the method processes all prs and pops the ReleaseBranch related prs"""
        to_pop = []  # type: List[int]
        for i, pr in enumerate(prs):
            if self.name not in pr.head.ref:
                # this pr is not for the current branch
                continue
            if pr.head.ref.startswith(f"cherrypick/{self.name}"):
                self.cherrypick_pr = pr
                to_pop.append(i)
            elif pr.head.ref.startswith(f"backport/{self.name}"):
                self.backport_pr = pr
                self._backported = True
                to_pop.append(i)
            else:
                assert False, f"BUG! Invalid PR's branch [{pr.head.ref}]"

        for i in reversed(to_pop):
            # Going from the tail to keep the order and pop greater index first
            prs.pop(i)
        return prs

    def process(  # pylint: disable=too-many-return-statements
        self, dry_run: bool
    ) -> None:
        if self.backported:
            return

        if not self.cherrypick_pr:
            if dry_run:
                logging.info(
                    "DRY RUN: Would create cherry-pick or backport PR for #%s",
                    self.pr.number,
                )
                return
            self.create_cherrypick()

        if self.backported:
            # The `backported` can be set to True if the changes are already applied
            return
        assert self.cherrypick_pr, "Unable to create cherry-pick PR"

        if self.cherrypick_pr.mergeable and self.cherrypick_pr.state != "closed":
            if dry_run:
                logging.info(
                    "DRY RUN: Would merge cherry-pick PR for #%s", self.pr.number
                )
                return
            self.cherrypick_pr.merge()
            # The PR needs update, since PR.merge doesn't update the object
            self.cherrypick_pr.update()
        if self.cherrypick_pr.merged:
            if dry_run:
                logging.info(
                    "DRY RUN: Would create backport PR for #%s", self.pr.number
                )
                return
            self.create_backport()
            return
        if self.cherrypick_pr.state == "closed":
            logging.info(
                "The cherry-pick PR #%s for PR #%s is discarded",
                self.cherrypick_pr.number,
                self.pr.number,
            )
            self._backported = True
            return
        logging.info(
            "Cherry-pick PR #%s for PR #%s has conflicts and unable to be merged",
            self.cherrypick_pr.number,
            self.pr.number,
        )
        # Assign to engineer if not already assigned (only for PRs with conflicts)
        if not self.cherrypick_pr.assignees:
            if dry_run:
                logging.info(
                    "DRY RUN: Would assign cherry-pick PR #%s to engineers",
                    self.cherrypick_pr.number,
                )
            else:
                self._assign_new_pr(self.cherrypick_pr)
                self.cherrypick_pr.update()
        self.ping_cherry_pick_assignees(dry_run)

    def create_cherrypick(self):
        # First, create backport branch:
        # Checkout release branch with discarding every change
        git_runner(f"{GIT_PREFIX} checkout -f {self.name}")
        # Create or reset backport branch
        git_runner(f"{GIT_PREFIX} checkout -B {self.backport_branch}")
        # Merge all changes from PR's the first parent commit w/o applying anything
        # It will allow to create a merge commit like it would be a cherry-pick
        first_parent = git_runner(f"git rev-parse {self.pr.merge_commit_sha}^1")
        git_runner(f"{GIT_PREFIX} merge -s ours --no-edit {first_parent}")

        # Second step, create cherrypick branch
        git_runner(
            f"{GIT_PREFIX} checkout --no-track -B "
            f"{self.cherrypick_branch} {self.pr.merge_commit_sha}"
        )

        # Merge backport_branch into cherrypick_branch locally to find out
        # whether the cherry-pick applies cleanly. The rename limit is raised
        # so git does not silently disable rename detection on large diffs
        # (files renamed between the release branch and master would otherwise
        # show up as spurious conflicts).
        #
        # - No conflicts: cherrypick_branch now holds the fully resolved tree,
        #   so the backport PR can be created directly from it and the
        #   intermediate cherry-pick PR is skipped (see
        #   create_backport_from_resolved_tree).
        # - Conflicts: `merge --abort` restores cherrypick_branch to
        #   pr.merge_commit_sha and we fall back to opening a cherry-pick PR
        #   for manual resolution by the assigned engineer.
        merged_cleanly = False
        try:
            git_runner(
                f"{GIT_PREFIX} -c merge.renameLimit=999999 "
                f"merge --no-ff --no-edit {self.backport_branch}"
            )
            merged_cleanly = True
        except CalledProcessError:
            try:
                git_runner(f"{GIT_PREFIX} merge --abort")
            except CalledProcessError:
                # `merge --abort` itself can fail when the merge process
                # crashed (e.g. merge-ort assertion) and left
                # `.git/index.lock` behind -- the lock blocks any further
                # git command in this checkout. Clean it up so subsequent
                # PRs in the same run are not poisoned.
                recover_git_state()

        if merged_cleanly:
            # If the merge produced no tree change vs backport_branch, the PR
            # is effectively already backported to the release branch - either
            # "Already up to date" (no merge commit at all) or an empty merge
            # commit whose resolution collapsed onto backport_branch's tree
            # (e.g. the PR was manually applied with equivalent content). In
            # either case, skip creating an empty PR.
            if not git_runner(
                f"{GIT_PREFIX} diff --name-only "
                f"{self.backport_branch} {self.cherrypick_branch}"
            ):
                logging.info(
                    "Release branch %s already contain changes from %s",
                    self.name,
                    self.pr.number,
                )
                self._backported = True
                return
            # A clean cherry-pick needs no manual conflict resolution, so the
            # intermediate cherry-pick PR carries no value. Create the backport
            # PR directly from the already resolved tree and skip it entirely.
            self.create_backport_from_resolved_tree()
            return

        # There are conflicts: cherrypick_branch stays at pr.merge_commit_sha
        # (merge --abort restored HEAD). Push both branches and open the
        # cherry-pick PR so the conflicts are surfaced on GitHub for manual
        # resolution by the assigned engineer.
        for branch in [self.cherrypick_branch, self.backport_branch]:
            git_runner(f"{GIT_PREFIX} push -f {self.REMOTE} {branch}:{branch}")

        title = f"Cherry pick #{self.pr.number} to {self.name}: {self.pr.title}"
        self.cherrypick_pr = self.repo.create_pull(
            title=title,
            body=self.body_header() + self.CHERRYPICK_DESCRIPTION + self.pr_source,
            base=self.backport_branch,
            head=self.cherrypick_branch,
        )
        self.cherrypick_pr.add_to_labels(Labels.PR_CHERRYPICK)
        self.cherrypick_pr.add_to_labels(Labels.DO_NOT_TEST)
        if Labels.PR_CRITICAL_BUGFIX in [label.name for label in self.pr.labels]:
            self.cherrypick_pr.add_to_labels(Labels.PR_CRITICAL_BUGFIX)
        elif Labels.PR_BUGFIX in [label.name for label in self.pr.labels]:
            self.cherrypick_pr.add_to_labels(Labels.PR_BUGFIX)
        # Do not assign yet - will assign only if there are conflicts
        # update cherrypick PR to get the state for PR.mergable
        self.cherrypick_pr.update()

    def create_backport(self):
        assert self.cherrypick_pr is not None
        # Checkout the backport branch from the remote and make all changes to
        # apply like they are only one cherry-pick commit on top of release
        logging.info("Creating backport for PR #%s", self.pr.number)
        git_runner(f"{GIT_PREFIX} checkout -f {self.backport_branch}")
        git_runner(f"{GIT_PREFIX} pull --ff-only {self.REMOTE} {self.backport_branch}")
        merge_base = git_runner(
            f"{GIT_PREFIX} merge-base "
            f"{self.REMOTE}/{self.name} {self.backport_branch}"
        )
        git_runner(f"{GIT_PREFIX} reset --soft {merge_base}")
        title = f"Backport #{self.pr.number} to {self.name}: {self.pr.title}"
        git_runner(f"{GIT_PREFIX} commit --allow-empty -F -", input=title)

        # Push with force, create the backport PR, label and assign it
        git_runner(
            f"{GIT_PREFIX} push -f {self.REMOTE} "
            f"{self.backport_branch}:{self.backport_branch}"
        )
        self._finalize_backport_pr(title)

    def create_backport_from_resolved_tree(self):
        # Fast path for a conflict-free cherry-pick: the merge done in
        # create_cherrypick already produced the fully resolved tree in
        # cherrypick_branch. Materialize it as a single commit on top of the
        # release branch tip - exactly what create_backport produces after a
        # cherry-pick PR is merged - and open the backport PR directly, without
        # the intermediate cherry-pick PR and the extra run it needs (the
        # cherry-pick PR is created in one run and only merged in a later one,
        # once GitHub has computed its mergeable state).
        logging.info(
            "Creating backport directly for PR #%s (no conflicts)", self.pr.number
        )
        resolved_tree = git_runner(f"git rev-parse {self.cherrypick_branch}^{{tree}}")
        title = f"Backport #{self.pr.number} to {self.name}: {self.pr.title}"
        commit = git_runner(
            f"{GIT_PREFIX} commit-tree {resolved_tree} "
            f"-p {self.REMOTE}/{self.name} -F -",
            input=title,
        )
        git_runner(f"{GIT_PREFIX} branch -f {self.backport_branch} {commit}")
        git_runner(
            f"{GIT_PREFIX} push -f {self.REMOTE} "
            f"{self.backport_branch}:{self.backport_branch}"
        )
        self._finalize_backport_pr(title)
        # A backport PR now exists, so the original PR is fully processed.
        self._backported = True

    def _finalize_backport_pr(self, title: str) -> None:
        try:
            self.backport_pr = self.repo.create_pull(
                title=title,
                body=self.body_header() + self.BACKPORT_DESCRIPTION + self.pr_source,
                base=self.name,
                head=self.backport_branch,
            )
        except GithubException as e:
            if e.status != 422 or "already exists" not in str(e):
                raise
            # The backport PR was created in a previous run but left without the
            # `pr-backport` label (e.g. the run was interrupted after `create_pull`
            # but before `add_to_labels`). Find and reuse it.
            existing = list(
                self.repo.get_pulls(
                    head=f"{self.repo.owner.login}:{self.backport_branch}",
                    base=self.name,
                    state="open",
                )
            )
            if not existing:
                raise
            self.backport_pr = existing[0]
            logging.warning(
                "Backport PR #%s for PR #%s already exists without label, reusing it",
                self.backport_pr.number,
                self.pr.number,
            )
        self.backport_pr.add_to_labels(Labels.PR_BACKPORT)
        if Labels.PR_CRITICAL_BUGFIX in [label.name for label in self.pr.labels]:
            self.backport_pr.add_to_labels(Labels.PR_CRITICAL_BUGFIX)
        elif Labels.PR_BUGFIX in [label.name for label in self.pr.labels]:
            self.backport_pr.add_to_labels(Labels.PR_BUGFIX)
        self._assign_new_pr(self.backport_pr)

    def body_header(self) -> str:
        """
        Returns the description of the original PR, which is used in the
        cherry-pick and backport PRs.
        """
        upstream_pr = ""
        if self.pr.head.ref.startswith(SYNC_PR_PREFIX):
            try:
                upstream_pr_number = int(self.pr.head.ref.rsplit("/", maxsplit=1)[-1])
                upstream_pr = (
                    f"Upstream pull-request {GITHUB_SERVER_URL}/"
                    f"{GITHUB_UPSTREAM_REPOSITORY}/pull/{upstream_pr_number}\n"
                )
            except ValueError:
                logging.error(
                    "Sync PR #%s has an invalid head ref: %s",
                    self.pr.number,
                    self.pr.head.ref,
                )
        original_pr = f"Original pull-request {self.pr.html_url}\n"
        cherrypick_pr = ""
        if self.cherrypick_pr is not None:
            cherrypick_pr = f"Cherry-pick pull-request {self.cherrypick_pr.html_url}\n"
        return f"{upstream_pr}{original_pr}{cherrypick_pr}\n"

    def ping_cherry_pick_assignees(self, dry_run: bool) -> None:
        assert self.cherrypick_pr is not None
        logging.info(
            "Checking if cherry-pick PR #%s needs to be pinged or closed",
            self.cherrypick_pr.number,
        )
        # The `updated_at` is Optional[datetime]
        cherrypick_updated_ts = (
            self.cherrypick_pr.updated_at or datetime.now()
        ).timestamp()
        since_updated = int(datetime.now().timestamp() - cherrypick_updated_ts)
        since_updated_str = (
            f"{since_updated // 86400}d{since_updated // 3600 % 24}h"
            f"{since_updated // 60 % 60}m{since_updated % 60}s"
        )

        PING_THRESHOLD = 3 * 24 * 3600  # 3 days
        CLOSE_THRESHOLD = 7 * 24 * 3600  # 7 days

        if since_updated < PING_THRESHOLD:
            logging.info(
                "The cherry-pick PR was updated %s ago, waiting for the next run",
                since_updated_str,
            )
            return

        if since_updated >= CLOSE_THRESHOLD:
            # Close the PR after 7 days
            if self.cherrypick_pr.assignees:
                assignees = ", ".join(
                    f"@{user.login}" for user in self.cherrypick_pr.assignees
                )
                comment_body = (
                    f"Dear {assignees}, this cherry-pick PR has not been updated for {since_updated_str}. "
                    f"Closing automatically. If you still want to backport #{self.pr.number}, "
                    "please resolve the conflicts and reopen this PR."
                )
            else:
                logging.warning(
                    "Cherry-pick PR #%s has no assignees when closing",
                    self.cherrypick_pr.number,
                )
                comment_body = (
                    f"This cherry-pick PR has not been updated for {since_updated_str}. "
                    f"Closing automatically. If you still want to backport #{self.pr.number}, "
                    "please resolve the conflicts and reopen this PR."
                )
            if dry_run:
                logging.info(
                    "DRY RUN: would close cherry-pick PR #%s with comment:\n%s",
                    self.cherrypick_pr.number,
                    comment_body,
                )
                return
            self.cherrypick_pr.create_issue_comment(comment_body)
            logging.info(
                "Posted closing comment to cherry-pick PR #%s",
                self.cherrypick_pr.number,
            )
            self.cherrypick_pr.edit(state="closed")
            logging.info(
                "Closed cherry-pick PR #%s after %s of inactivity",
                self.cherrypick_pr.number,
                since_updated_str,
            )
            return

        # Ping after 3 days
        # Check if we've already pinged to avoid spamming
        comments = self.cherrypick_pr.get_issue_comments()
        for comment in comments:
            if (
                "has not been updated for" in comment.body
                and "resolve the conflicts" in comment.body
            ):
                # We've already pinged, don't ping again
                logging.info(
                    "Already pinged cherry-pick PR #%s, waiting for update or closure threshold",
                    self.cherrypick_pr.number,
                )
                return

        if self.cherrypick_pr.assignees:
            assignees = ", ".join(
                f"@{user.login}" for user in self.cherrypick_pr.assignees
            )
            comment_body = (
                f"Dear {assignees}, this cherry-pick PR has not been updated for {since_updated_str}. "
                f"Please resolve the conflicts to backport #{self.pr.number}, "
                "or close this PR if the backport is no longer needed. "
                f"This PR will be automatically closed after {CLOSE_THRESHOLD // 86400} days of inactivity."
            )
        else:
            logging.warning(
                "Cherry-pick PR #%s has no assignees when pinging",
                self.cherrypick_pr.number,
            )
            comment_body = (
                f"This cherry-pick PR has not been updated for {since_updated_str}. "
                f"Please resolve the conflicts to backport #{self.pr.number}, "
                "or close this PR if the backport is no longer needed. "
                f"This PR will be automatically closed after {CLOSE_THRESHOLD // 86400} days of inactivity."
            )
        if dry_run:
            logging.info(
                "DRY RUN: would comment on cherry-pick PR #%s:\n%s",
                self.cherrypick_pr.number,
                comment_body,
            )
            return

        self.cherrypick_pr.create_issue_comment(comment_body)
        logging.info(
            "Posted ping comment to cherry-pick PR #%s after %s of inactivity",
            self.cherrypick_pr.number,
            since_updated_str,
        )

    def _assign_new_pr(self, new_pr: PullRequest) -> None:
        """Assign `new_pr` to author, merger and assignees of an original PR"""
        # It looks there some race when multiple .add_to_assignees are executed,
        # so we'll add all at once
        assignees = [self.pr.user, self.pr.merged_by]
        if self.pr.assignees:
            assignees.extend(self.pr.assignees)
        assignees = [
            a
            for a in assignees
            if "robot-clickhouse" not in str(a) and "clickhouse-gh" not in str(a)
        ]
        logging.info(
            "Assing #%s to author and assignees of the original PR: %s",
            new_pr.number,
            ", ".join(user.login for user in assignees),
        )
        new_pr.add_to_assignees(*assignees)

    @property
    def backported(self) -> bool:
        return self._backported

    def __repr__(self):
        return self.name


class BackportPRs:
    def __init__(
        self,
        gh: GitHub,
        repo: str,
        dry_run: bool,
    ):
        self.gh = gh
        self._repo_name = repo
        self.dry_run = dry_run

        self._remote = ""
        self._remote_line = ""

        self._repo = None  # type: Optional[Repository]
        self.release_prs = []  # type: PullRequests
        self.release_branches = []  # type: List[str]
        self.labels_to_backport = []  # type: List[str]
        self.prs_for_backport = []  # type: PullRequests
        self.error = None  # type: Optional[Exception]

    @property
    def remote_line(self) -> str:
        if not self._remote_line:
            # lines of "origin	[email protected]:ClickHouse/ClickHouse.git (fetch)"
            remotes = git_runner("git remote -v").split("\n")
            # We need the first word from the first matching result
            self._remote_line = next(
                iter(
                    remote
                    for remote in remotes
                    if f"github.com/{self._repo_name}" in remote  # https
                    or f"github.com:{self._repo_name}" in remote  # ssh
                )
            )

        return self._remote_line

    @property
    def remote(self) -> str:
        if not self._remote:
            self._remote = self.remote_line.split(maxsplit=1)[0]
            git_runner(f"git fetch {self._remote}")
            ReleaseBranch.REMOTE = self._remote
        return self._remote

    @property
    def is_remote_ssh(self) -> bool:
        return "github.com:" in self.remote_line

    def receive_release_prs(self):
        logging.info("Getting release PRs")
        self.release_prs = self.gh.get_release_pulls(self._repo_name)
        self.release_branches = [pr.head.ref for pr in self.release_prs]

        # A version-specific label `vX.Y-must-backport` is backported to X.Y and
        # to every newer active release (see `select_backport_branches`). The
        # named release need not be active itself, so the search must also pick
        # up PRs labelled for an end-of-life release as long as a newer release
        # is still active. Include every version-specific label that exists in
        # the repo whose version is not newer than the newest active release --
        # a newer label could not expand to any active branch.
        newest_active = max(branch_version(branch) for branch in self.release_branches)
        self.labels_to_backport = sorted(
            label.name
            for label in self.repo.get_labels()
            if label_version(label.name) is not None
            and label_version(label.name) <= newest_active
        )

        logging.info("Active releases: %s", ", ".join(self.release_branches))
        logging.info("Labels to backport: %s", ", ".join(self.labels_to_backport))

    def update_local_release_branches(self):
        logging.info("Update local release branches")
        branches = git_runner("git branch").split()
        for branch in self.release_branches:
            if branch not in branches:
                # the local branch is not exist, so continue
                continue
            local_ref = git_runner(f"git rev-parse {branch}")
            remote_ref = git_runner(f"git rev-parse {self.remote}/{branch}")
            if local_ref == remote_ref:
                # Do not need to update, continue
                continue
            logging.info("Resetting %s to %s/%s", branch, self.remote, branch)
            git_runner(f"git branch -f {branch} {self.remote}/{branch}")

    def oldest_commit_date(self) -> date:
        # The dates of every commit in each release branche
        commit_dates = [
            commit
            for branch in self.release_branches
            for commit in git_runner(
                "git log --no-merges --format=format:%cs --reverse "
                f"{self.remote}/{self.default_branch}..{self.remote}/{branch}"
            ).split("\n")
        ]
        return min(date.fromisoformat(c_date) for c_date in commit_dates)

    def receive_prs_for_backport(
        self,
        since_date: Optional[date] = None,
        # The following arguments are used for a cross-repo labels synchronization
        labels_to_backport: Optional[Iterable[str]] = None,
        backport_created_label: str = Labels.PR_BACKPORTS_CREATED,
        repo_name: str = "",
    ) -> None:

        since_date = since_date or self.oldest_commit_date()
        labels_to_backport = labels_to_backport or (
            self.labels_to_backport + [Labels.MUST_BACKPORT, Labels.MUST_BACKPORT_FORCE]
        )
        repo_name = repo_name or self.repo.full_name
        # To not have a possible TZ issues
        tomorrow = date.today() + timedelta(days=1)

        # The search API struggles to serve the heavy queries, so we limit the
        # updated date to 90 days ago. It improves the response quality by an order of
        # magnitude
        updated = (date.today() - timedelta(days=90)).isoformat() + "..*"

        query_args = {
            "query": f"type:pr repo:{repo_name} -label:{backport_created_label}",
            "label": ",".join(labels_to_backport),
            "updated": updated,
            "merged": [since_date, tomorrow],
        }
        logging.info("Query to find the backport PRs:\n %s", query_args)

        self.prs_for_backport = self.gh.get_pulls_from_search(**query_args)
        logging.info(
            "PRs to be backported:\n %s",
            "\n ".join([pr.html_url for pr in self.prs_for_backport]),
        )

    def process_backports(self):
        for pr in self.prs_for_backport:
            try:
                self.process_pr(pr)
            except Exception as e:
                logging.error(
                    "During processing the PR #%s error occurred: %s", pr.number, e
                )
                self.error = e
                # Whatever went wrong, make sure the next PR starts from a
                # clean working tree -- a leftover `.git/index.lock` from a
                # crashed git process would otherwise break every later PR.
                recover_git_state()

    def _rolling_out_branches(self) -> List[str]:
        """
        Returns release branch names whose corresponding release PR has the
        `rolling-out` label.  Used to skip general backports (`pr-must-backport`
        or `pr-critical-bugfix`) to branches that are currently being rolled out.
        Direct version-specific labels (e.g. `v25.10-must-backport`) always
        override this and proceed as normal.
        """
        return [
            release_pr.head.ref
            for release_pr in self.release_prs
            if Labels.ROLLING_OUT in {label.name for label in release_pr.labels}
        ]

    def _close_prs_for_rolling_out_branch(self, pr: PullRequest, branch: str) -> None:
        """
        Close any open cherry-pick or backport PRs that were previously created
        for a release branch that is now marked `rolling-out`.
        """
        cp_branch = ReleaseBranch.cp_branch(branch, pr.number)
        bp_branch = ReleaseBranch.bp_branch(branch, pr.number)
        # Search separately: label:A,B in GitHub search is AND (must have both),
        # so we issue two queries — one per label — to find either type.
        open_prs = []
        for head_branch, label in (
            (cp_branch, Labels.PR_CHERRYPICK),
            (bp_branch, Labels.PR_BACKPORT),
        ):
            open_prs += self.gh.get_pulls_from_search(
                query=f"type:pr repo:{self._repo_name} head:{head_branch}",
                state="open",
                label=label,
            )
        for open_pr in open_prs:
            logging.info(
                "PR #%s: closing PR #%s because release branch %s is rolling-out",
                pr.number,
                open_pr.number,
                branch,
            )
            if self.dry_run:
                logging.info(
                    "DRY RUN: would close PR #%s for rolling-out branch %s",
                    open_pr.number,
                    branch,
                )
                continue
            version = branch.replace("release/", "")
            open_pr.create_issue_comment(
                f"Closing this PR because the target release branch `{branch}` "
                "is currently being rolled out. Backporting is skipped for "
                "rolling-out branches when the original PR carries only the "
                "generic `pr-must-backport` or `pr-critical-bugfix` label.\n\n"
                f"If you still want to backport this change, add the "
                f"`v{version}-must-backport` label to the original PR "
                f"#{pr.number} — that overrides the rolling-out skip."
            )
            open_pr.edit(state="closed")

    def process_pr(self, pr: PullRequest) -> None:
        pr_labels = [label.name for label in pr.labels]

        # Decide the target release branches (pure logic, unit-tested in
        # `test_cherry_pick_branches.py`). A version-specific label
        # (`vX.Y-must-backport`) marks the OLDEST release the PR must reach, so
        # the PR is backported to that release and every newer active release
        # branch; the lowest such label wins. `skipped` are rolling-out branches
        # excluded for a general backport that no version-specific label covers.
        rolling_out = set(self._rolling_out_branches())
        branch_names, skipped = select_backport_branches(
            pr_labels,
            self.release_branches,
            rolling_out,
            general_backport_labels={Labels.MUST_BACKPORT, Labels.MUST_BACKPORT_FORCE}
            | Labels.AUTO_BACKPORT,
            force_backport_label=Labels.MUST_BACKPORT_FORCE,
        )

        if skipped:
            logging.info(
                "PR #%s: skipping rolling-out release branches for general "
                "backport: %s",
                pr.number,
                ", ".join(skipped),
            )
            for br in skipped:
                self._close_prs_for_rolling_out_branch(pr, br)

        if not branch_names:
            logging.info(
                "PR #%s: all candidate release branches are rolling-out, "
                "skipping backport",
                pr.number,
            )
            return

        branches = [
            ReleaseBranch(br, pr, self.repo) for br in branch_names
        ]  # type: List[ReleaseBranch]

        logging.info(
            "  PR #%s is supposed to be backported to %s",
            pr.number,
            ", ".join(map(str, branches)),
        )
        # All PRs for cherry-pick and backport branches as heads
        query_suffix = " ".join(
            [
                f"head:{branch.backport_branch} head:{branch.cherrypick_branch}"
                for branch in branches
            ]
        )

        # Backport and cherry-pick PRs
        bp_cp_prs = self.gh.get_pulls_from_search(
            query=f"type:pr repo:{self._repo_name} {query_suffix}",
            label=f"{Labels.PR_BACKPORT},{Labels.PR_CHERRYPICK}",
        )
        # Check that all
        for br in branches:
            bp_cp_prs = br.pop_prs(bp_cp_prs)
        assert not bp_cp_prs, f"Some PRs are not processed by backporting: {bp_cp_prs}"

        for br in branches:
            br.process(self.dry_run)

        if all(br.backported for br in branches):
            # And check it after the running
            self.mark_pr_backported(pr)

    def mark_pr_backported(self, pr: PullRequest) -> None:
        if self.dry_run:
            logging.info("DRY RUN: would mark PR #%s as done", pr.number)
            return
        pr.add_to_labels(Labels.PR_BACKPORTS_CREATED)
        logging.info(
            "PR #%s is successfully labeled with `%s`",
            pr.number,
            Labels.PR_BACKPORTS_CREATED,
        )

    @property
    def repo(self) -> Repository:
        if self._repo is None:
            self._repo = self.gh.get_repo(self._repo_name)
        return self._repo

    @property
    def default_branch(self) -> str:
        return self.repo.default_branch


class CherryPickPRs:
    def __init__(self, gh: GitHub, repo: str, dry_run: bool):
        self.gh = gh
        self.repo = gh.get_repo(repo)
        self.dry_run = dry_run
        self.error = None  # type: Optional[Exception]

        self.release_prs = gh.get_release_pulls(repo)
        logging.info(f"Release PRs: {self.release_prs}")

    def get_open_cherry_pick_prs(self) -> PullRequests:
        """
        Get all open cherry-pick PRs in the repository.
        """
        query_args = {
            "query": f"type:pr repo:{self.repo.full_name} label:{Labels.PR_CHERRYPICK}",
            "state": "open",
        }
        logging.info("Query to find the cherry-pick PRs:\n %s", query_args)
        return self.gh.get_pulls_from_search(**query_args)

    def check_open_prs(self) -> None:
        """
        After the cherry-pick PRs are closed, the original PRs are marked as
        `pr-backports-created`. If the cherry-pick PR is reopened, we remove this label
        """
        try:
            prs = self.get_open_cherry_pick_prs()
            logging.info("Found %d open cherry-pick PRs", len(prs))
        except Exception as e:
            logging.error("Error while getting open cherry-pick PRs: %s", e)
            self.error = e
            return

        # We need to check if there is an open release branch for each cherry-pick PR
        for pr in prs.copy():
            # We need to copy the list, since we will modify it
            #
            try:
                if not self._check_opened_release(pr):
                    # The cherry-pick PR is not opened in any of the release branches,
                    # so we can skip it
                    prs.remove(pr)
                    continue
            except Exception as e:
                logging.error(
                    "Error while checking opened release branch for cherry-pick PR #%s: %s",
                    pr.number,
                    e,
                )
                self.error = e
                continue

        # And then, we need to check if the original PR is marked as backported for any
        # open cherry-pick PR
        for pr in prs:
            try:
                self._remove_backported_label(pr)
            except Exception as e:
                logging.error(
                    "Error while pinging stale cherry-pick PR #%s: %s", pr.number, e
                )
                self.error = e
                continue

    def _check_opened_release(self, cpp: PullRequest) -> bool:
        """
        Check if the original PR is opened in any of the release branches.
        """
        # The cherry-pick's head ref is like cherrypick/{release_name}/12345,
        # so we can extract the release name from it, and then try to find it in the
        # self.release_prs
        original_pr_number = int(cpp.head.ref.rsplit("/", maxsplit=1)[-1])
        if cpp.head.ref in [
            ReleaseBranch.cp_branch(r.head.ref, original_pr_number)
            for r in self.release_prs
        ]:
            # The release branch is opened, so we can continue
            return True
        release_name = cpp.head.ref.split("/", maxsplit=1)[1].rsplit("/", maxsplit=1)[0]
        logging.info(
            "An opened release PR `%s` for cherry-pick PR %s is not found, going to close it",
            release_name,
            cpp.html_url,
        )
        if self.dry_run:
            logging.info(
                "DRY RUN: would close and leave a comment in the cherry-pick PR #%s",
                cpp.number,
            )
            return False
        cpp.create_issue_comment(
            f"The release branch `{release_name}` for the cherry-pick doesn't have "
            "an opened PR, closing this PR."
        )
        cpp.edit(state="closed")
        return False

    def _remove_backported_label(self, pr: PullRequest) -> None:
        # The `updated_at` is Optional[datetime]
        try:
            original_pr_number = int(pr.head.ref.rsplit("/")[-1])
        except ValueError:
            logging.error(
                "Cherry-pick PR #%s has an invalid head ref: %s",
                pr.number,
                pr.head.ref,
            )
            raise

        original_pr = self.gh.get_pull_cached(self.repo, original_pr_number)
        if not any(l.name == Labels.PR_BACKPORTS_CREATED for l in original_pr.labels):
            # The original PR is not marked as backported, so nothing to do
            return

        if pr.assignees:
            assignees = ", ".join(f"@{user.login}" for user in pr.assignees)
            comment_body = (
                f"Dear {assignees}, this PR is opened while #{original_pr.number} was "
                f"marked as backported. The `{Labels.PR_BACKPORTS_CREATED}` is removed, so "
                "the original PR can be processed again.\n\n"
                "If the cherry-pick is not needed anymore, then just close this PR."
            )
        else:
            logging.warning(
                "Cherry-pick PR #%s has no assignees when removing backported label",
                pr.number,
            )
            comment_body = (
                f"This PR is opened while #{original_pr.number} was "
                f"marked as backported. The `{Labels.PR_BACKPORTS_CREATED}` is removed, so "
                "the original PR can be processed again.\n\n"
                "If the cherry-pick is not needed anymore, then just close this PR."
            )
        logging.info(
            "Label %s should be removed from from #%s due opened cherry-pick PR #%s",
            Labels.PR_BACKPORTS_CREATED,
            original_pr.number,
            pr.number,
        )
        if self.dry_run:
            logging.info(
                "DRY RUN: would remove label and comment the cherry-pick PR #%s:\n%s",
                pr.number,
                comment_body,
            )
            return

        try:
            original_pr.remove_from_labels(Labels.PR_BACKPORTS_CREATED)
        except GithubException as e:
            if e.status == 404:
                logging.info(
                    "Label %s is already removed from PR #%s",
                    Labels.PR_BACKPORTS_CREATED,
                    original_pr.number,
                )
            else:
                raise
        pr.create_issue_comment(comment_body)
        logging.info(
            "Removed label %s from PR #%s and posted comment to cherry-pick PR #%s",
            Labels.PR_BACKPORTS_CREATED,
            original_pr.number,
            pr.number,
        )


def parse_args():
    parser = argparse.ArgumentParser(
        "Create cherry-pick and backport PRs",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument("--token", help="github token, if not set, used from smm")
    parser.add_argument("--repo", default=GITHUB_REPOSITORY, help="repo owner/name")
    parser.add_argument("--dry-run", action="store_true", help="do not create anything")

    parser.add_argument(
        "--debug-helpers",
        action="store_true",
        help="add debug logging for git_helper and github_helper",
    )
    return parser.parse_args()


def main():
    temp_path = Path(TEMP_PATH)
    temp_path.mkdir(parents=True, exist_ok=True)

    args = parse_args()
    if args.debug_helpers:
        logging.getLogger("github_helper").setLevel(logging.DEBUG)
        logging.getLogger("git_helper").setLevel(logging.DEBUG)
    token = args.token or get_best_robot_token()

    gh = GitHub(token)
    temp_path = Path(TEMP_PATH)
    gh_cache = GitHubCache(gh.cache_path, temp_path, S3Helper())
    gh_cache.download()

    # First, check if some cherry-pick PRs are reopened and original PRs are mared as
    # done
    cpp = CherryPickPRs(gh, args.repo, args.dry_run)
    cpp.check_open_prs()

    bpp = BackportPRs(gh, args.repo, args.dry_run)

    bpp.gh.cache_path = temp_path / "gh_cache"
    bpp.receive_release_prs()
    bpp.update_local_release_branches()
    bpp.receive_prs_for_backport()
    bpp.process_backports()
    gh_cache.upload()

    errors = [e for e in (bpp.error, cpp.error) if e is not None]
    if any(errors):
        logging.error("Finished successfully, but %s errors occurred!", len(errors))
        raise BackportException(
            "Errors occurred during backport process: "
            + "; ".join(str(e) for e in errors)
        )


if __name__ == "__main__":
    logging.getLogger().setLevel(level=logging.INFO)

    assert not is_shallow()
    try:
        with stash():
            if os.getenv("ROBOT_CLICKHOUSE_SSH_KEY", ""):
                with SSHKey("ROBOT_CLICKHOUSE_SSH_KEY"):
                    main()
            else:
                main()

    except Exception as e:
        if IS_CI:
            ci_buddy = CIBuddy()
            ci_buddy.post_job_error(
                f"The backport process finished with errors: {e}",
                with_instance_info=True,
                with_wf_link=True,
                critical=True,
            )