anchor.diff:      work_on_phase(JobId, JobState),
       dht_node_state:set_mr_state(State, JobId, JobState);
   
  -on({mr, phase_result, JobId, {work_done, Data}}, State) ->
  +on({mr, phase_result, JobId, {work_done, Data}, Range}, State) ->
       ?TRACE("mr_~s on ~p: received phase results: ~p...~ndistributing...~n",
              [JobId, self(), lists:sublist(Data, 5)]),
       Ref = uid:get_global_uid(),
       NewMRState = mr_state:set_acked(dht_node_state:get_mr_state(State, JobId), {Ref, []}),
       case mr_state:is_last_phase(NewMRState) of
           false ->
  -            Reply = comm:reply_as(comm:this(), 4, {mr, next_phase_data_ack, {JobId,
  -                                                                              Ref}, '_'}),
  +            Reply = comm:reply_as(comm:this(), 4, {mr, next_phase_data_ack,
  +                                                   {JobId, Ref, Range}, '_'}),
               bulkowner:issue_bulk_distribute(Ref, dht_node,
                                               5, {mr, next_phase_data, JobId, Reply, '_'},
                                               Data);
  
positive.code:     State.
  
  work_on_phase(JobId, MRState) ->
      case mr_state:get_phase(MRState) of
          {_Round, _MoR, _FunTerm, []} ->
              case mr_state:is_last_phase(MRState) of
                  false ->
                      Master = mr_state:get(MRState, master),
                      MyRange = mr_state:get(MRState, my_range),
                      comm:send(Master, {mr, phase_completed, MyRange}),
                      ?TRACE("no data for phase...done...~p informs master~n", [self()]);
                  _ ->
  
negative.code:     comm:send(Source, mr_state:get(NewMRState, my_range)),
      dht_node_state:set_mr_state(State, JobId, NewMRState);
  
  on({mr, next_phase_data_ack, {JobId, Ref, Range}, Interval}, State) ->
      NewMRState = mr_state:set_acked(dht_node_state:get_mr_state(State, JobId),
                                      {Ref, Interval}),
      case mr_state:is_acked_complete(NewMRState) of
          true ->
              Master = mr_state:get(NewMRState, master),
              comm:send(Master, {mr, phase_completed, Range}),
              ?TRACE("Phase complete...~p informing master~n", [self()]);
          false ->
              ?TRACE("~p is still waiting for phase to complete~n", [self()])
  