Repository: spotify/luigi
File: luigi/scheduler.py (Python)

Code before:
        self.update(worker_id)

    def _upstream_status(self, task_id, upstream_status_table):
        if task_id in upstream_status_table:
            return upstream_status_table[task_id]
        elif self._state.has_task(task_id):
            task_stack = [task_id]

            while task_stack:
                dep_id = task_stack.pop()
                dep = self._state.get_task(dep_id)
                if dep:
                    if dep.status == DONE:
                        continue
                    if dep_id not in upstream_status_table:
                        if dep.status == PENDING and dep.deps:
                            task_stack += [dep_id] + list(dep.deps)
                            upstream_status_table[dep_id] = ''  # will be updated postorder
                        else:
                            dep_status = STATUS_TO_UPSTREAM_MAP.get(dep.status, '')
                            upstream_status_table[dep_id] = dep_status
                    elif upstream_status_table[dep_id] == '' and dep.deps:
                        # This is the postorder update step when we set the
                        # status based on the previously calculated child elements
                        status = max((upstream_status_table.get(a_task_id, '')
                                      for a_task_id in dep.deps),
                                     key=UPSTREAM_SEVERITY_KEY)
                        upstream_status_table[dep_id] = status
            return upstream_status_table[dep_id]

    def _serialize_task(self, task_id, include_deps=True, deps=None):
        task = self._state.get_task(task_id)
        ret = {
            'display_name': task.pretty_id,
            'status': task.status,
            'workers': list(task.workers),
            'worker_running': task.worker_running,
            'time_running': getattr(task, "time_running", None),
            'start_time': task.time,
            'last_updated': getattr(task, "updated", task.time),
            'params': task.params,
            'name': task.family,
            'priority': task.priority,
            'resources': task.resources,
            'tracking_url': getattr(task, "tracking_url", None),
            'status_message': getattr(task, "status_message", None)
        }
        if task.status == DISABLED:
            ret['re_enable_able'] = task.scheduler_disable_time is not None
        if include_deps:
            ret['deps'] = list(task.deps if deps is None else deps)

Review comment (bug):
@Tarrasch Here is the case;

When any `SUCCESS` task exists among upstream tasks, `upstream_status_table.get(a_task_id, '')` results `''` empty string event all other upstream tasks are `FAILED` or `DISABLED`. Because there is no record in `upstream_status_table` for `SUCCESS` tasks. This will cause the wrapper is as PENDING all the time even there is no PENDING task among its upstream tasks. This will also let worker runs forever. 

Here are some SS showing two case;

### Case which is just swtiching max to min;

![screen shot 2016-07-28 at 11 11 53](https://cloud.githubusercontent.com/assets/1279644/17205761/01b303ba-54b5-11e6-99f6-c1f261ca7a67.png)

As you can se the wrapper task which is `TaskLevelRetryPolicy` runs all the time, even its dependencies `SUCCESS` and `DISABLED`. This is happenning because there is at least one `SUCCESS` task.

### The case what is in my commit;

![screen shot 2016-07-28 at 11 14 51](https://cloud.githubusercontent.com/assets/1279644/17205813/5356489e-54b5-11e6-9ee7-90f4d678ade8.png)

In this case, the empty string upstream severity is ignored. First, it tries to find min among elements not includes empty string. Even there is SUCCESS which causes empty string return,
elmination will be done among severities. You can see it on the SS.

So all I mean is; max works fine, but min doesn't work like it.


Code after:
        self.update(worker_id)

    def _upstream_status(self, task_id, upstream_status_table):
        if task_id in upstream_status_table:
            return upstream_status_table[task_id]
        elif self._state.has_task(task_id):
            task_stack = [task_id]

            while task_stack:
                dep_id = task_stack.pop()
                dep = self._state.get_task(dep_id)
                if dep:
                    if dep.status == DONE:
                        continue
                    if dep_id not in upstream_status_table:
                        if dep.status == PENDING and dep.deps:
                            task_stack += [dep_id] + list(dep.deps)
                            upstream_status_table[dep_id] = ''  # will be updated postorder
                        else:
                            dep_status = STATUS_TO_UPSTREAM_MAP.get(dep.status, '')
                            upstream_status_table[dep_id] = dep_status
                    elif upstream_status_table[dep_id] == '' and dep.deps:
                        # This is the postorder update step when we set the
                        # status based on the previously calculated child elements
                        upstream_severities = list(upstream_status_table.get(a_task_id) for a_task_id in dep.deps if a_task_id in upstream_status_table) or ['']
                        status = min(upstream_severities, key=UPSTREAM_SEVERITY_KEY)
                        upstream_status_table[dep_id] = status
            return upstream_status_table[dep_id]

    def _serialize_task(self, task_id, include_deps=True, deps=None):
        task = self._state.get_task(task_id)
        ret = {
            'display_name': task.pretty_id,
            'status': task.status,
            'workers': list(task.workers),
            'worker_running': task.worker_running,
            'time_running': getattr(task, "time_running", None),
            'start_time': task.time,
            'last_updated': getattr(task, "updated", task.time),
            'params': task.params,
            'name': task.family,
            'priority': task.priority,
            'resources': task.resources,
            'tracking_url': getattr(task, "tracking_url", None),
            'status_message': getattr(task, "status_message", None)
        }
        if task.status == DISABLED:
            ret['re_enable_able'] = task.scheduler_disable_time is not None
        if include_deps:
            ret['deps'] = list(task.deps if deps is None else deps)
        return ret