anchor.diff:      end.
   
   stop_update_notifiers(Notifiers) ->
  -    [rexi:kill(Node, Ref) || {Node, Ref} <- Notifiers].
  +    [rexi:kill(Node, Ref) || #worker{node=Node, ref=Ref} <- Notifiers].
   
   stop({Pid, Ref}) ->
       erlang:send(Pid, {Ref, done}).
  
positive.code:         {state, Pid, State} -> State
      end.
  
  receive_results(Workers, State, Timeout) ->
      case rexi_utils:recv(Workers, 2, fun handle_message/3, State,
              infinity, Timeout) of
      {timeout, {NewWorkers, Parent, waiting}} ->
          erlang:send(Parent, {state, self(), timeout}),
          receive_results(NewWorkers, {NewWorkers, Parent, unset}, Timeout);
      {timeout, {NewWorkers, Parent, _State}} ->
          receive_results(NewWorkers, {NewWorkers, Parent, timeout}, Timeout);
      {_, NewState} ->
          {ok, NewState}
      end.
  
  
  handle_message({rexi_DOWN, _, {_,NodeRef},_}, _Worker, {Workers, Parent, State}) ->
      NewWorkers = lists:filter(fun({_Node, Ref}) -> NodeRef =/= Ref end, Workers),
      case NewWorkers of
      [] ->
          {error, {nodedown, <<"progress not possible">>}};
      _ ->
          {ok, {NewWorkers, Parent, State}}
      end;
  handle_message({rexi_EXIT, Reason}, Worker, {Workers, Parent, State}) ->
      NewWorkers = lists:delete(Worker,Workers),
      case NewWorkers of
      [] ->
          {error, Reason};
      _ ->
          {ok, {NewWorkers, Parent, State}}
      end;
  handle_message(db_updated, {_Worker, _From}, {Workers, Parent, waiting}) ->
      % propagate message to calling controller
      erlang:send(Parent, {state, self(), updated}),
      {ok, {Workers, Parent, unset}};
  handle_message(db_updated, _Worker, {Workers, Parent, _State}) ->
      {ok, {Workers, Parent, updated}};
  handle_message(get_state, {_Worker, _From}, {Workers, Parent, unset}) ->
      {ok, {Workers, Parent, waiting}};
  handle_message(get_state, {_Worker, _From}, {Workers, Parent, State}) ->
      erlang:send(Parent, {state, self(), State}),
      {ok, {Workers, Parent, unset}};
  handle_message(done, _, _) ->
      {stop, ok}.
  
  
negative.code: -include("fabric.hrl").
  -include_lib("mem3/include/mem3.hrl").
  
  -record(worker, {
      ref,
      node,
      pid
  }).
  
  -record(acc, {
      parent,
      state
  }).
  
  go(Parent, ParentRef, DbName, Timeout) ->
      Notifiers = start_update_notifiers(DbName),
      MonRefs = lists:usort([{rexi_server, Node} || {Node, _Ref} <- Notifiers]),
      RexiMon = rexi_monitor:start(MonRefs),
      MonPid = start_cleanup_monitor(self(), Notifiers),
      %% This is not a common pattern for rexi but to enable the calling
      %% process to communicate via handle_message/3 we "fake" it as a
      %% a spawned worker.
      Workers = [#worker{ref=ParentRef, pid=Parent} | Notifiers],
      try
          receive_results(Workers, #acc{parent=Parent, state=unset}, Timeout)
      after
          rexi_monitor:stop(RexiMon),
          stop_cleanup_monitor(MonPid)
  