anchor.diff:      #task_spec{stage = Stage} = TaskSpec,
       case jc_utils:new_task_permitted_for_stage(P, Stage, SI) of
           false ->
  -            do_submit_tasks(Mode, Rest, S#state{pending = [{TaskId,
  -                            Mode}|Pending]}, NFailuresAllowed);
  +            do_submit_tasks(Mode, Rest,
  +                            S#state{pending = gb_sets:add_element({TaskId, Mode}, Pending)},
  +                            NFailuresAllowed);
           true ->
  -            S1 = S#state{pending = Pending -- [{TaskId, Mode}]},
  +            NewPending = case gb_sets:is_element({TaskId, Mode}, Pending) of
  +                true ->
  +                    gb_sets:del_element({TaskId, Mode}, Pending);
  +                false -> Pending
  +            end,
  +            S1 = S#state{pending = NewPending},
               do_submit_tasks_in(Mode, [TaskId|Rest], S1, NFailuresAllowed)
       end.
   
  
positive.code:     S3.
  
  -spec maybe_start_pending(state()) -> state().
  maybe_start_pending(#state{pending = AllPending} = S) ->
      lists:foldl(
          fun({TaskId, Mode}, S1) ->
              #state{tasks = Tasks, pipeline = P, stage_info = SI} = S1,
              #task_info{spec = TaskSpec} = jc_utils:task_info(TaskId, Tasks),
  
negative.code:             TInfo1 = TInfo#task_info{failed_count = FC,
                                       failed_hosts = Blacklist},
              S#state{tasks = jc_utils:update_task_info(TaskId, TInfo1, Tasks),
                      pending = gb_sets:add_element({TaskId, re_run}, Pending)}
      end.
  
  -spec do_kill_job(term(), state()) -> ok.
  