anchor.diff:  -define(FAILED_TASK_PAUSE, 1000).
   -define(FAILED_MAX_PAUSE, 60 * 1000).
   -define(FAILED_PAUSE_RANDOMIZE, 30 * 1000).
  +
  +-type result() :: {node(), nonempty_string()}.
   
   % In theory we could keep the HTTP connection pending until the job
   % finishes but in practice long-living HTTP connections are a bad idea.
  
positive.code: 
          event_server:event(Name, "Reduce phase done", [], []),
          event_server:event(Name, "READY: Job finished in " ++
              disco_server:format_time(Started),
              [], {ready, RedResults});
      true ->
          event_server:event(Name, "READY: Job finished in " ++
              disco_server:format_time(Started),
              [], {ready, RedInputs})
      end,
      gen_server:cast(event_server, {job_done, Name}).
  
  
negative.code: 
  wait_workers(N, Results, Name, Mode) ->
      receive
          {{job_ok, Result}, Task, Node} ->
              event_server:event(Name,
              "Received results from ~s:~B @ ~s.",
                  [Task#task.mode, Task#task.taskid, Node],
                  {task_ready, Mode}),
              {N - 1, gb_trees:enter(Task#task.taskid,
                                     {disco:node(Node), list_to_binary(Result)},
                                     Results)};
  
          {{data_error, _Msg}, Task, Node} ->
              handle_data_error(Task, Node),
              {N, Results};
  
          {{job_error, _Error}, _Task, _Node} ->
              throw(logged_error);
  
          {{error, Error}, Task, Node} ->
              event_server:event(Name,
              "ERROR: Worker crashed in ~s:~B @ ~s: ~p",
                  [Task#task.mode, Task#task.taskid,
                  Node, Error], []),
              throw(logged_error);
  
          Error ->
              event_server:event(Name,
              "ERROR: Received an unknown error: ~p",
                  [Error], []),
              throw(logged_error)
      end.
  
  -spec submit_task(task()) -> _.
  submit_task(Task) ->
      case catch gen_server:call(disco_server, {new_task, Task}, 30000) of
          ok ->
              ok;
          _ ->
              event_server:event(Task#task.jobname,
              "ERROR: ~s:~B scheduling failed. "
              "Try again later.",
              [Task#task.mode, Task#task.taskid], []),
              throw(logged_error)
      end.
  
  % data_error signals that a task failed on an error that is not likely
  