anchor.diff:      handle_info/2
   ]).
   
  +-export([
  +    handle_config_change/5,
  +    handle_config_terminate/3
  +]).
  +
   -include_lib("mem3/include/mem3.hrl").
   
   -record(state, {
       nodes,
       shards,
  -    users
  +    users,
  +    delay,
  +    frequency,
  +    last_push,
  +    buckets
   }).
  +
  +%% Calling mem3_sync:push/2 on every update has a measurable performance cost,
  +%% so we'd like to coalesce multiple update messages from couch_event in to a
  +%% single push call. Doing this while ensuring both correctness (i.e., no lost
  +%% updates) and an even load profile is somewhat subtle. This implementation
  +%% groups updated shards in a list of "buckets" (see bucket_shard/2) and
  +%% guarantees that each shard is in no more than one bucket at a time - i.e.,
  +%% any update messages received before the shard's current bucket has been
  +%% pushed will be ignored - thereby reducing the frequency with which a single
  +%% shard will be pushed. mem3_sync:push/2 is called on all shards in the
  +%% *oldest* bucket roughly every mem3.sync_frequency milliseconds (see
  +%% maybe_push_shards/1) to even out the load on mem3_sync.
   
   start_link() ->
       couch_event_listener:start_link(?MODULE, [], [all_dbs]).
   
   init(_) ->
  -    State = #state{
  +    config:listen_for_changes(?MODULE, undefined),
  +    Delay = config:get_integer("mem3", "sync_delay", 5000),
  +    Frequency = config:get_integer("mem3", "sync_frequency", 500),
  +    Buckets = lists:duplicate(Delay div Frequency + 1, sets:new()),
  +    St = #state{
           nodes = mem3_sync:nodes_db(),
           shards = mem3_sync:shards_db(),
  -        users = mem3_sync:users_db()
  +        users = mem3_sync:users_db(),
  +        delay = Delay,
  +        frequency = Frequency,
  +        buckets = Buckets
       },
  -    {ok, State}.
  +    {ok, St}.
   
   terminate(_Reason, _State) ->
       ok.
  
positive.code:     Nodes = mem3:nodes(),
      Live = nodes(),
      [mem3_sync:push(NodesDb, N) || N <- Nodes, lists:member(N, Live)],
      {ok, St};
  handle_event(ShardsDb, updated, #state{shards = ShardsDb} = St) ->
      mem3_sync:push(ShardsDb, mem3_sync:find_next_node()),
      {ok, St};
  handle_event(UsersDb, updated, #state{users = UsersDb} = St) ->
      mem3_sync:push(UsersDb, mem3_sync:find_next_node()),
      {ok, St};
  handle_event(<<"shards/", _/binary>> = ShardName, updated, St) ->
      try mem3:shards(mem3:dbname(ShardName)) of
      Shards ->
          Targets = [S || #shard{node=N, name=Name} = S <- Shards,
  
negative.code:             lists:member(N, Live)]
      catch error:database_does_not_exist ->
          ok
      end.
  