anchor.diff:  handle_cast({stage_done, Stage}, S) ->
       {noreply, do_stage_done(Stage, S)};
   handle_cast({submit_tasks, Mode, Tasks}, S) ->
  -    {noreply, do_submit_tasks(Mode, Tasks, S)};
  +    {noreply, do_submit_tasks(Mode, Tasks, S, ?FAILURES_ALLOWED)};
  +handle_cast({submit_tasks, Mode, Tasks, NFailuresAllowed}, S) ->
  +    {noreply, do_submit_tasks(Mode, Tasks, S, NFailuresAllowed)};
   handle_cast({task_done, TaskId, Host, Results}, S) ->
       {noreply, do_task_done(TaskId, Host, Results, S)};
   handle_cast(pipeline_done, #state{jobinfo = #jobinfo{jobname = JobName}} = S) ->
  
positive.code:             Event = {stage_start, Stage, NTasks},
              event_server:event(JobName, "Stage ~s scheduled with ~p tasks",
                                 [Stage, NTasks], Event),
              do_submit_tasks(first_run, Tasks, S1)
      end.
  
  -spec setup_stage_tasks(stage_name(), stage_name(), label_grouping(),
  
negative.code:     make_stage_tasks(Stage, Grouping, SaveOutputs, SaveInfo, Rest, S1,
                       {TaskNum + 1, [NextTaskId | Acc]}).
  
  
   -spec do_submit_tasks(submit_mode(), [task_id()], state(),
       non_neg_integer()) -> state().
  do_submit_tasks(_Mode, [], S, _) -> S;
  do_submit_tasks(Mode, [TaskId | Rest], #state{stage_info = SI,
                                                data_map   = DataMap,
                                                next_runid = RunId,
                                                hosts      = Hosts,
                                                tasks      = Tasks} = S,
                                        NFailuresAllowed) ->
      #task_info{spec = TaskSpec, failed_hosts = FailedHosts}
          = jc_utils:task_info(TaskId, Tasks),
      #task_spec{stage = Stage, group = {_L, H}, input = Input} = TaskSpec,
  