anchor.diff:  -define(CAPABILITIES, [async_fold]).
   
   -record(state, {ref :: reference(),
  -                link_file :: string(),
  -                data_file :: string(),
  +                data_dir :: string(),
                   opts :: [{atom(), term()}],
  +                partition :: integer(),
                   root :: string(),
                   async_folds :: boolean()}).
   
  
positive.code:     end.
  
  %% @private
  check_symlink(LinkFile, DeleteExisting) ->
      case file:read_link(LinkFile) of
          {ok, DataFile} ->
              case DeleteExisting of
                  true ->
                      file:delete(LinkFile),
                      make_symlink(LinkFile);
                  false ->
                      {ok, DataFile}
              end;
          {error, enoent} ->
              make_symlink(LinkFile);
          Error ->
              Error
      end.
  
  %% @private
  make_symlink(LinkFile) ->
      {MegaSecs, Secs, MicroSecs} = erlang:now(),
      LinkAbsPath = filename:absname(LinkFile),
      DataFile = [filename:basename(LinkFile),
                  "-",
                  integer_to_list(MegaSecs),
                  integer_to_list(Secs),
                  integer_to_list(MicroSecs)],
      case filelib:ensure_dir(LinkAbsPath) of
          ok ->
              case file:make_symlink(DataFile,
                                     LinkAbsPath) of
                  ok ->
                      {ok, DataFile};
                  {error, Reason} ->
                      lager:error("Failed to create bitcask dir ~s: ~p",
                                  [DataFile, Reason]),
                      {error, Reason}
              end;
          {error, Reason1} ->
              lager:error("Failed to create bitcask dir ~s: ~p",
                          [DataFile, Reason1]),
              {error, Reason1}
      end.
  
  %% @private
  drop_data_cleanup(DataRoot, LinkFile, DataFile) ->
      fun() ->
              %% List all the directories in data root
              case file:list_dir(DataRoot) of
                  {ok, Dirs} ->
                      LinkBase = filename:basename(LinkFile),
                      %% Delete the contents of each directory and
                      %% the directory itself excluding the
                      %% current data directory and the symlink
                      %% pointing to it.
                      [data_directory_cleanup(filename:join(DataRoot, Dir)) ||
                          Dir <- Dirs,
                          Dir /= lists:flatten(DataFile),
                          Dir /= LinkBase,
                          string:left(Dir, length(LinkBase) + 1) =:= (LinkBase ++ "-")];
                  {error, _} ->
                      ignore
              end
  
negative.code: %%  of its own, use that instead.
  -spec drop(state()) -> {ok, state()} | {error, term(), state()}.
  drop(#state{ref=Ref,
              partition=Partition,
              root=DataRoot,
              opts=BitcaskOpts}=State) ->
      %% Close the bitcask reference
      bitcask:close(Ref),
  
      case make_data_dir(filename:join([DataRoot,
                                        integer_to_list(Partition)])) of
          {ok, DataDir} ->
              %% Spawn a process to cleanup the old data files.
              %% The use of spawn is intentional. We do not
              %% care if this process dies since any lingering
              %% files will be cleaned up on the next drop.
              %% The worst case is that the files hang
              %% around and take up some disk space.
              spawn(drop_data_cleanup(DataRoot, Partition, DataDir)),
  
              %% Now open the bitcask and return an updated state
              %% so this backend can continue processing.
              case bitcask:open(filename:join(DataRoot, DataDir), BitcaskOpts) of
                  Ref1 when is_reference(Ref1) ->
                      {ok, State#state{data_dir=DataDir,
                                       ref=Ref1}};
                  {error, Reason} ->
                      {error, Reason, State#state{data_dir=DataDir}}
              end;
          {error, Reason1} ->
              {error, Reason1, State}
  