def __call__()

in sync/handlers.py [0:0]

• 63 lines of code
• 18 McCabe index (conditional complexity)

    def __call__(self, git_gecko: Repo, git_wpt: Repo, body: Dict[str, Any]) -> None:
        newrelic.agent.set_transaction_name("DecisionTaskHandler")
        task_id = body["status"]["taskId"]
        taskgroup_id = body["status"]["taskGroupId"]

        msg = "Expected kind decision-task, got %s" % body["task"]["tags"]["kind"]
        assert body["task"]["tags"]["kind"] == "decision-task", msg

        newrelic.agent.add_custom_parameter("tc_task", task_id)
        newrelic.agent.add_custom_parameter("tc_taskgroup", taskgroup_id)

        state = body["status"]["state"]
        newrelic.agent.add_custom_parameter("state", state)

        # Enforce the invariant that the taskgroup id is not set until
        # the decision task is complete. This allows us to determine if a
        # try push should have the expected wpt tasks just by checking if
        # this is set
        if state not in self.complete_states:
            logger.info("Decision task is not yet complete, status %s" % state)
            return

        task = tc.get_task(task_id)
        if task is None:
            raise ValueError("Failed to get task for task_id %s" % task_id)

        sha1 = task.get("payload", {}).get("env", {}).get("GECKO_HEAD_REV")

        if sha1 is None:
            raise ValueError("Failed to get commit sha1 from task message")

        if state == "exception":
            run_id = body["runId"]
            runs = body.get("status", {}).get("runs", [])
            if 0 <= run_id < len(runs):
                reason = runs[run_id].get("reasonResolved")
                if reason in ["superseded",
                              "claim-expired",
                              "worker-shutdown",
                              "intermittent-task"]:
                    logger.info("Task %s had an exception for reason %s, "
                                "assuming taskcluster will retry" %
                                (task_id, reason))
                    return

        try_push = trypush.TryPush.for_commit(git_gecko, sha1)
        if not try_push:
            logger.debug(f"No try push for SHA1 {sha1} taskId {task_id}")
            # This could be a race condition if the decision task completes before this
            # task is in the index
            raise RetryableError("Got a wptsync task with no corresponding try push")

        with SyncLock.for_process(try_push.process_name) as lock:
            assert isinstance(lock, SyncLock)
            with try_push.as_mut(lock):
                # If we retrigger, we create a new taskgroup, with id equal to the new task_id.
                # But the retriggered decision task itself is still in the original taskgroup
                if state == "completed":
                    logger.info("Setting taskgroup id for try push %r to %s" %
                                (try_push, taskgroup_id))
                    try_push.taskgroup_id = taskgroup_id
                elif state in ("failed", "exception"):
                    sync = try_push.sync(git_gecko, git_wpt)
                    message = ("Decision task got status %s for task %s%s" %
                               (state, sha1, " PR %s" % sync.pr if sync and sync.pr else ""))
                    logger.error(message)
                    taskgroup = tc.TaskGroup(task["taskGroupId"])
                    if len(taskgroup.view(
                            lambda x: x["task"]["metadata"]["name"] == "Gecko Decision Task")) > 5:
                        try_push.status = "complete"
                        try_push.infra_fail = True
                        try_push.taskgroup_id = taskgroup_id
                        if sync and sync.bug:
                            env.bz.comment(
                                sync.bug,
                                "Try push failed: decision task %s returned error" % task_id)
                    else:
                        logger.info("Retriggering decision task for sync %s" %
                                    (sync.process_name,))
                        client = tc.TaskclusterClient()
                        client.retrigger(task_id)