if (not unfinished_tasks and
                    any(r.state in (State.FAILED, State.UPSTREAM_FAILED) for r in roots)):
                self.log.info('Marking run %s failed', self)
                self.state = State.FAILED
                dag.handle_callback(self, success=False, reason='task_failure',