anchor.diff:              do_submit_tasks_in(Mode, [TaskId|Rest], S1, NFailuresAllowed)
       end.
   
  +-spec do_submit_tasks_in(submit_mode(), [task_id()], state(), non_neg_integer()) -> state().
   do_submit_tasks_in(Mode, [TaskId | Rest], #state{stage_info = SI,
                                                 data_map   = DataMap,
                                                 next_runid = RunId,
  
positive.code:     S2 = do_next_stage(Stage, S1),
      maybe_start_pending(S2).
  
  maybe_event_stage_done(Stage, #state{pipeline = P, stage_info = SI} = S) ->
      case can_finish(P, Stage, SI) of
          true -> event_stage_done(Stage, S);
  
negative.code:     make_stage_tasks(Stage, Grouping, Rest, S2,
                       {TaskNum + 1, [NextTaskId | Acc]}).
  
  -spec add_inputs_to_data_map(state(), [{input_id(), data_input()}]) -> state().
  add_inputs_to_data_map(#state{data_map = OldDataMap} = S, Inputs) ->
      DataMap = lists:foldl(
                  fun({InputId, DataInput}, DM) ->
  