**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
"""Autocomplete."""
        query = unquote(request.args.get('query', ''))
        if not query:
            return wwwutils.json_response([])
        
        dag_ids_query = session.query(DagModel.dag_id.label('item')).filter(  
            ~DagModel.is_subdag, DagModel.is_active, DagModel.dag_id.ilike('%' + query + '%')
        )  
        owners_query = session.query(func.distinct(DagModel.owners).label('item')).filter(
            ~DagModel.is_subdag, DagModel.is_active, DagModel.owners.ilike('%' + query + '%')
        )  
        
        status = flask_session.get(FILTER_STATUS_COOKIE)
        if status == 'active':
            dag_ids_query = dag_ids_query.filter(~DagModel.is_paused)
            owners_query = owners_query.filter(~DagModel.is_paused)
        elif status == 'paused':
            dag_ids_query = dag_ids_query.filter(DagModel.is_paused)
            owners_query = owners_query.filter(DagModel.is_paused)
        filter_dag_ids = current_app.appbuilder.sm.get_accessible_dag_ids(g.user)
        
        dag_ids_query = dag_ids_query.filter(DagModel.dag_id.in_(filter_dag_ids))
        owners_query = owners_query.filter(DagModel.dag_id.in_(filter_dag_ids))
        
        payload = [row[0] for row in dag_ids_query.union(owners_query).limit(10).all()]
        return wwwutils.json_response(payload)
class DagDependenciesView(AirflowBaseView):