**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/www/views.py
"""Refresh DAG."""
        dag_id = request.values.get('dag_id')
        orm_dag = session.query(DagModel).filter(DagModel.dag_id == dag_id).first()
        if orm_dag:
            orm_dag.last_expired = timezone.utcnow()
            session.merge(orm_dag)
        session.commit()
        dag = current_app.dag_bag.get_dag(dag_id)
        
        current_app.appbuilder.sm.sync_perm_for_dag(dag_id, dag.access_control)
        flash(f"DAG [{dag_id}] is now fresh as a daisy")
        return redirect(request.referrer)
    @expose('/refresh_all', methods=['POST'])
    @auth.has_access(
        [
            (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_DAG),
        ]
    )
    @action_logging
    def refresh_all(self):
        """Refresh everything"""
        current_app.dag_bag.collect_dags_from_db()
        
        for dag_id, dag in current_app.dag_bag.dags.items():
            current_app.appbuilder.sm.sync_perm_for_dag(dag_id, dag.access_control)
        flash("All DAGs are now up to date")
        return redirect(url_for('Airflow.index'))
    @expose('/gantt')
    @auth.has_access(
        [