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: 
  start_update_notifiers(DbName) ->
      lists:map(fun(#shard{node=Node, name=Name}) ->
          {Node, rexi:cast(Node, {?MODULE, start_update_notifier, [Name]})}
      end, mem3:shards(DbName)).
  
  % rexi endpoint
  
negative.code:         {state, Pid, State} -> State
      end.
  
  receive_results(Workers, Acc0, Timeout) ->
      Fun = fun handle_message/3,
      case rexi_utils:recv(Workers, #worker.ref, Fun, Acc0, infinity, Timeout) of
      {timeout, #acc{state=updated}=Acc} ->
          receive_results(Workers, Acc, Timeout);
      {timeout, #acc{state=waiting}=Acc} ->
          erlang:send(Acc#acc.parent, {state, self(), timeout}),
          receive_results(Workers, Acc#acc{state=unset}, Timeout);
      {timeout, Acc} ->
          receive_results(Workers, Acc#acc{state=timeout}, Timeout);
      {_, Acc} ->
          {ok, Acc}
      end.
  
  
  handle_message({rexi_DOWN, _, {_, Node}, _}, _Worker, _Acc) ->
      {error, {nodedown, Node}};
  handle_message({rexi_EXIT, _Reason}, Worker, _Acc) ->
      {error, {worker_exit, Worker}};
  handle_message({gen_event_EXIT, DbName, Reason}, _Worker, _Acc) ->
      {error, {gen_event_exit, DbName, Reason}};
  handle_message(db_updated, _Worker, #acc{state=waiting}=Acc) ->
      % propagate message to calling controller
      erlang:send(Acc#acc.parent, {state, self(), updated}),
      {ok, Acc#acc{state=unset}};
  handle_message(db_updated, _Worker, Acc) ->
      {ok, Acc#acc{state=updated}};
  handle_message(get_state, _Worker, #acc{state=unset}=Acc) ->
      {ok, Acc#acc{state=waiting}};
  handle_message(get_state, _Worker, Acc) ->
      erlang:send(Acc#acc.parent, {state, self(), Acc#acc.state}),
      {ok, Acc#acc{state=unset}};
  handle_message(done, _, _) ->
      {stop, ok}.
  
  