**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
session.query(DagRun.dag_id, func.count('*'))
            .filter(
                DagRun.dag_id.in_(existing_dag_ids),
                DagRun.state == State.RUNNING,  
                DagRun.external_trigger == expression.false(),
            )
            .group_by(DagRun.dag_id)
            .all()
        )
        for orm_dag in sorted(orm_dags, key=lambda d: d.dag_id):
            dag = dag_by_ids[orm_dag.dag_id]
            if dag.is_subdag:
                orm_dag.is_subdag = True
                orm_dag.fileloc = dag.parent_dag.fileloc  
                orm_dag.root_dag_id = dag.parent_dag.dag_id  
                orm_dag.owners = dag.parent_dag.owner  
            else:
                orm_dag.is_subdag = False
                orm_dag.fileloc = dag.fileloc
                orm_dag.owners = dag.owner
            orm_dag.is_active = True
            orm_dag.last_parsed_time = timezone.utcnow()
            orm_dag.default_view = dag.default_view
            orm_dag.description = dag.description
            orm_dag.schedule_interval = dag.schedule_interval
            orm_dag.concurrency = dag.concurrency
            orm_dag.has_task_concurrency_limits = any(t.task_concurrency is not None for t in dag.tasks)
            orm_dag.calculate_dagrun_date_fields(
                dag,
                most_recent_dag_runs.get(dag.dag_id),