**Description**

For a DAG that looks like this:

```
A >> B
```

If A fails, B goes into "upstream_failed" state. If a user then marks A "success", B will remain in "upstream_failed" state. It will not automatically start running. This scenario often happens if the failure of A is dealt with outside of Airflow and the user does not want Airflow to run A again. But he usually expect Airflow to run B after A is marked "success".

**Use case / motivation**

After A is marked "success", its downstream tasks B should be cleared automatically and get ready to be scheduled. To avoid changing this behaviour completely, there can be a toggle button next to "Success" that lets the user decide if he wants Airflow to automatically clear downstream tasks that are in "upstream_failed" state.
airflow/models/dag.py
"""This method is deprecated in favor of bulk_write_to_db"""
        warnings.warn(
            "This method is deprecated and will be removed in a future version. Please use bulk_write_to_db",
            DeprecationWarning,
            stacklevel=2,
        )
        return cls.bulk_write_to_db(dags, session)
    @classmethod
    @provide_session
    def bulk_write_to_db(cls, dags: Collection["DAG"], session=None):
        """
        Ensure the DagModel rows for the given dags are up-to-date in the dag table in the DB, including
        calculated fields.
        Note that this method can be called for both DAGs and SubDAGs. A SubDag is actually a SubDagOperator.
        :param dags: the DAG objects to save to the DB
        :type dags: List[airflow.models.dag.DAG]
        :return: None
        """
        if not dags:
            return
        log.info("Sync %s DAGs", len(dags))
        dag_by_ids = {dag.dag_id: dag for dag in dags}
        dag_ids = set(dag_by_ids.keys())
        query = (
            session.query(DagModel)