anchor.diff:      end,
       lists:foldl(FoldFun, LocalOps, CountByNode).
   
  -donate_fold(_Shard, {_, _, Moves, 0}) ->
  +donate_fold(_Shard, #gacc{limit = 0, moves = Moves}) ->
       throw({complete, Moves});
  -donate_fold(#shard{node = Node} = Shard, {Node, Nodes, Moves, DC}) ->
  +donate_fold(#shard{node = Node} = Shard, #gacc{node = Node} = Acc0) ->
  +     #gacc{
  +        targets = Nodes,
  +        moves = Moves,
  +        limit = DC,
  +        target_level = TargetLevel
  +    } = Acc0,
       Zone = mem3:node_info(Node, <<"zone">>),
       Shards = apply_shard_moves(mem3:shards(Shard#shard.dbname), Moves),
       InZone = filter_map_by_zone(shards_by_node(Shards, Nodes), Zone),
  
positive.code:     end, SortedByCount),
      case {lists:member(Shard, Shards), Candidates} of
          {false, _} ->
              {Node, Nodes, Moves, DC};
          {true, []} ->
              {Node, Nodes, Moves, DC};
          {true, [{Node, _} | _]} ->
              {Node, Nodes, Moves, DC};
          {true, [{Target, _} | _]} ->
              % Execute the move only if the target has fewer shards for this DB
              % than the source. Otherwise we'd generate a local imbalance.
              SourceCount = get_shard_count(Node, SortedByCount),
              TargetCount = get_shard_count(Target, SortedByCount),
              if TargetCount < SourceCount ->
                  print({move, Shard, Target}),
                  {Node, Nodes, [{move, Shard, Target} | Moves], DC - 1};
              true ->
                  {Node, Nodes, Moves, DC}
              end
      end;
  donate_fold(_Shard, Acc) ->
  
negative.code:             Acc;
          ({Node0, Count}, Acc) ->
              Node = list_to_existing_atom(binary_to_list(Node0)),
              InternalAcc0 = #gacc{
                  node = Node,
                  targets = TargetNodes0,
                  moves = Acc,
                  limit = erlang:min(Count - TargetLevel, Limit - length(Acc)),
                  target_level = TargetLevel
              },
              try mem3_shards:fold(fun donate_fold/2, InternalAcc0) of
                  #gacc{moves = Moves} ->
                      Moves
              catch
                  {complete, Moves} ->
  