anchor.diff:  
   
   %% @doc Set up replication from the given source node to a list of
  -%% {VBucket, DstNode} pairs. Will silently refuse to start a new
  -%% replica if backfill is still starting on a source or destination
  -%% node. Returns true if it started any.
  +%% {VBucket, DstNode} pairs.
   set_replicas(SrcNode, Bucket, Replicas) ->
  -    GoodChildren = kill_runaway_children(SrcNode, Bucket, Replicas),
  -    %% Now filter out the replicas that still have children
  -    Actions = actions(GoodChildren),
  -    Sorted = lists:keysort(2, Replicas -- Actions),
  -    Grouped = misc:keygroup(2, Sorted),
  +    %% A dictionary mapping destination node to a sorted list of
  +    %% vbuckets to replicate there from SrcNode.
  +    DesiredReplicaDict =
  +        dict:map(
  +          fun (_, V) -> lists:usort(V) end,
  +          lists:foldl(
  +            fun ({VBucket, DstNode}, D) ->
  +                    dict:append(DstNode, VBucket, D)
  +            end, dict:new(), Replicas)),
  +    %% Remove any destination nodes from the dictionary that the
  +    %% replication is already correct for.
  +    NeededReplicas =
  +        dict:to_list(
  +          lists:foldl(
  +            fun (#child_id{dest_node=DstNode, vbuckets=VBuckets} = Id, D) ->
  +                    case dict:find(DstNode, DesiredReplicaDict) of
  +                        {ok, VBuckets} ->
  +                            %% Already running
  +                            dict:erase(DstNode, D);
  +                        _ ->
  +                            %% Either wrong vbuckets or not wanted at all
  +                            kill_child(SrcNode, Bucket, Id),
  +                            D
  +                    end
  +            end, DesiredReplicaDict, children(SrcNode, Bucket))),
       lists:foreach(
  -      fun ({DstNode, R}) ->
  -              VBuckets = [V || {V, _} <- R],
  -              start_replicas(SrcNode, Bucket, VBuckets,
  -                             DstNode)
  -      end, Grouped).
  +      fun ({DstNode, VBuckets}) ->
  +              start_child(SrcNode, Bucket, VBuckets, DstNode)
  +      end, NeededReplicas).
   
   
   -spec start_child(atom(), nonempty_string(), [non_neg_integer(),...], atom()) ->
                            {ok, pid()}.
   start_child(Node, Bucket, VBuckets, DstNode) ->
  -    PortServerArgs = args(Node, Bucket, VBuckets, DstNode, false),
  -    ?log_info("Args =~n~p",
  -              [PortServerArgs]),
  +    Args = args(Node, Bucket, VBuckets, DstNode, false),
  +    ?log_info("Starting replicator with args =~n~p",
  +              [Args]),
       ChildSpec = {#child_id{vbuckets=VBuckets, dest_node=DstNode},
  -                 {ebucketmigrator_srv, start_link, PortServerArgs},
  +                 {ebucketmigrator_srv, start_link, Args},
                    permanent, 10, worker, [ebucketmigrator_srv]},
       supervisor:start_child({server(Bucket), Node}, ChildSpec).
  -
  -
  -start_replicas(SrcNode, Bucket, VBuckets, DstNode) ->
  -    SortedVBuckets = lists:usort(VBuckets),
  -    ?log_info("Starting replicator for vbuckets ~w in bucket ~p from node ~p to node ~p",
  -              [SortedVBuckets, Bucket, SrcNode, DstNode]),
  -    kill_vbuckets(DstNode, Bucket, VBuckets),
  -    lists:foreach(
  -      fun (V) ->
  -              ns_memcached:set_vbucket(DstNode, Bucket, V, replica)
  -      end, VBuckets),
  -    %% Make sure the command line doesn't get too long
  -    {ok, _Pid} = start_child(SrcNode, Bucket, VBuckets, DstNode).
  -
  -%% @doc Generate a unique name suffix that's valid for a TAP queue.
  -unique_suffix(DstNode) ->
  -    {MegaSecs, Secs, MicroSecs} = now(),
  -    lists:flatten(io_lib:format("~s-~s-~B.~6.10.0B",
  -                                [DstNode, node(),
  -                                 MegaSecs * 1000000 + Secs, MicroSecs])).
  
positive.code:                   _:_ -> []
                end
        end, Nodes).
  
  actions(Children) ->
      [{VBucket, Dst} || #child_id{vbuckets=VBuckets, dest_node=Dst} <- Children,
                         VBucket <- VBuckets].
  
  kill_vbuckets(Node, Bucket, VBuckets) ->
      {ok, States} = ns_memcached:list_vbuckets(Node, Bucket),
      case [X || X = {V, S} <- States, lists:member(V, VBuckets), S /= replica] of
          [] ->
              ok;
          RemainingVBuckets ->
              lists:foreach(fun ({V, dead}) ->
                                    ns_memcached:delete_vbucket(Node, Bucket, V);
                                ({V, _}) ->
                                    ns_memcached:set_vbucket(Node, Bucket,
                                                             V, dead),
                                    ns_memcached:delete_vbucket(Node, Bucket, V)
                                end, RemainingVBuckets)
      end.
  
  
  set_replicas(Bucket, NodesReplicas) ->
      %% Replace with the empty list if replication is disabled
  
negative.code:                   [any(), ...].
  args(Node, Bucket, VBuckets, DstNode, TakeOver) ->
      {User, Pass} = ns_bucket:credentials(Bucket),
      Suffix = case TakeOver of
                   true ->
                       [VBucket] = VBuckets,
                       integer_to_list(VBucket);
                   false ->
                       %% We want to reuse names for replication.
                       atom_to_list(DstNode)
               end,
      [ns_memcached:host_port(Node), ns_memcached:host_port(DstNode),
       [{username, User},
        {password, Pass},
        {vbuckets, VBuckets},
        {takeover, TakeOver},
        {suffix, Suffix}]].
  
  -spec children(node(), nonempty_string()) -> [#child_id{}].
  children(Node, Bucket) ->
  