**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
num_active_runs.get(dag.dag_id, 0),
            )
            for orm_tag in list(orm_dag.tags):
                if orm_tag.name not in orm_dag.tags:
                    session.delete(orm_tag)
                orm_dag.tags.remove(orm_tag)
            if dag.tags:
                orm_tag_names = [t.name for t in orm_dag.tags]
                for dag_tag in list(dag.tags):
                    if dag_tag not in orm_tag_names:
                        dag_tag_orm = DagTag(name=dag_tag, dag_id=dag.dag_id)
                        orm_dag.tags.append(dag_tag_orm)
                        session.add(dag_tag_orm)
        if settings.STORE_DAG_CODE:
            DagCode.bulk_sync_to_db([dag.fileloc for dag in orm_dags])
        
        
        session.flush()
        for dag in dags:
            cls.bulk_write_to_db(dag.subdags, session=session)
    @provide_session
    def sync_to_db(self, session=None):
        """
        Save attributes about this DAG to the DB. Note that this method
        can be called for both DAGs and SubDAGs. A SubDag is actually a
        SubDagOperator.
        :return: None
        """
        self.bulk_write_to_db([self], session)
    def get_default_view(self):