async def METHOD_NAME(
    self, heartbeat_cutoff: datetime.datetime = None
) -> int:
    """
    Zombie tasks are tasks that claim to be Running, but haven't updated their heartbeat.
    This method either retries them or marks them as failed.
    Returns:
        - int: the number of zombie task runs that were handled
    """
    zombies = 0
    heartbeat_cutoff = heartbeat_cutoff or pendulum.now("utc").subtract(minutes=10)
    where_clause = await self.get_task_runs_where_clause(
        heartbeat_cutoff=heartbeat_cutoff
    )
    task_runs = await models.TaskRun.where(where_clause).get(
        selection_set={
            "id": True,
            "flow_run_id": True,
            "tenant_id": True,
            "name": True,
            # Information about the current flow run state
            "flow_run": {"state"},
            # get information about retries from task
            "task": {"max_retries", "retry_delay"},
            # count the number of retrying states for this task run
            with_args(
                "retry_count: states_aggregate",
                {"where": {"state": {"_eq": "Retrying"}}},
            ): {"aggregate": {"count"}},
        },
        limit=5000,
        order_by={"updated": EnumValue("desc")},
        apply_schema=False,
    )
    if task_runs:
        self.logger.info(f"Zombie killer found {len(task_runs)} task runs.")
    # Set task run states to failed
    for tr in task_runs:
        try:
            # if the flow run is running and retries are available, mark as retrying
            if (
                tr.flow_run.state == "Running"
                and tr.retry_count.aggregate.count < (tr.task.max_retries or 0)
            ):
                message = (
                    "No heartbeat detected from the remote task; retrying the run."
                    f"This will be retry {tr.retry_count.aggregate.count + 1} of {tr.task.max_retries}."
                )
                retry_delay = orm._as_timedelta(tr.task.retry_delay or "0")
                await prefect.api.states.set_task_run_state(
                    task_run_id=tr.id,
                    state=Retrying(
                        message=message,
                        run_count=tr.retry_count.aggregate.count + 1,
                        start_time=pendulum.now("UTC") + retry_delay,
                    ),
                )
            # mark failed
            else:
                message = (
                    "No heartbeat detected from the remote task; marking the "
                    f"task run {tr.id} ({tr.name}) as failed."
                )
                await prefect.api.states.set_task_run_state(
                    task_run_id=tr.id,
                    state=Failed(message=message),
                )
            # log the state change to the task run
            await prefect.api.logs.create_logs(
                [
                    dict(
                        tenant_id=tr.tenant_id,
                        flow_run_id=tr.flow_run_id,
                        task_run_id=tr.id,
                        name=f"{self.logger.name}.TaskRun",
                        message=message,
                        level="ERROR",
                    )
                ],
                defer_db_write=False,
            )
            zombies += 1
        except ValueError as exc:
            self.logger.error(exc, exc_info=True)
    if zombies:
        self.logger.info(f"Addressed {zombies} zombie task runs.")
    return zombies