anchor.diff:      dht_node_state:set_mr_state(State, JobId, NewMRState);
   
   on({mr, next_phase, JobId, Round, _DeliveryInterval}, State) ->
  +    %% master started next round.
       ?TRACE("master initiated phase ~p in ~p ~p~n",
                 [Round, JobId, self()]),
       work_on_phase(JobId, State, Round),
       State;
   
   on({mr, terminate_job, JobId, _DeliveryInterval}, State) ->
  +    %% master wants to terminate job.
       MRState = dht_node_state:get_mr_state(State, JobId),
       _ = mr_state:clean_up(MRState),
       dht_node_state:delete_mr_state(State, JobId);
  
positive.code:     dht_node_state:set_mr_state(State, JobId, NewMRState);
  
  on({mr, next_phase_data_ack, AckInterval, JobId, DeliveryInterval}, State) ->
      NewMRState = mr_state:set_acked(dht_node_state:get_mr_state(State, JobId),
                                      AckInterval),
      case mr_state:is_acked_complete(NewMRState) of
  
negative.code: 
  on({bulk_distribute, _Id, Interval,
     {mr, next_phase_data, JobId, AckRange, Data, Round}, _Parents}, State) ->
      %% processing of data for next phase.
      %% save data and send ack
      ?TRACE("mr_~s on ~p: received next phase data for ~p: ~p~n",
             [JobId, self(), Interval, Data]),
      NewMRState = mr_state:add_data_to_phase(dht_node_state:get_mr_state(State,
  