https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
_log_state(task_instance=task_instance, lead_msg="Immediate failure requested. " if force_fail else "")
    if (
        failure_context["task"]
        and failure_context["email_for_state"](failure_context["task"])
        and failure_context["task"].email
    ):
        try:
            task_instance.email_alert(error, failure_context["task"])
        except Exception:
            log.exception("Failed to send email to: %s", failure_context["task"].email)
    if failure_context["callbacks"] and failure_context["context"]:
        _run_finished_callback(
            callbacks=failure_context["callbacks"],
            context=failure_context["context"],
        )
    if not test_mode:
        TaskInstance.save_to_db(failure_context["ti"], session)
def _get_try_number(*, task_instance: TaskInstance | TaskInstancePydantic):
    """
    Return the try number that a task number will be when it is actually run.
    If the TaskInstance is currently running, this will match the column in the
    database, in all other cases this will be incremented.
    This is designed so that task logs end up in the right file.
    :param task_instance: the task instance
    :meta private:
    """
    if task_instance.state == TaskInstanceState.RUNNING.RUNNING:
        return task_instance._try_number
    return task_instance._try_number + 1
def _set_try_number(*, task_instance: TaskInstance | TaskInstancePydantic, value: int) -> None: