anchor.diff:  %% @doc Initialize the fsm.
   -spec init([binary() | timeout()]) ->
                     {ok, initialize, state(), 0} |
  -                  {ok, waiting_file_info, state()}.
  -init([Bucket, Name, Timeout]) ->
  -    %% @TODO Get rid of this once sure that we can
  -    %% guarantee file name will be passed in as a binary.
  -    case is_binary(Name) of
  -        true ->
  -            FileName = Name;
  -        false ->
  -            FileName = list_to_binary(Name)
  -    end,
  +                  {ok, write_root, state()}.
  +init([_Bucket, _Key, _Options]) ->
  +    %% Set up our state record
  +    %% and try to register
  +    %% ourselves with gproc,
  +    %% failing if we can't
  +    {ok, prepare, #state{}, 0}.
   
  -    State = #state{bucket=Bucket,
  -                   filename=FileName,
  -                   timeout=Timeout},
  -    {ok, initialize, State, 0};
  -init({test, Args, StateProps}) ->
  -    {ok, initialize, State, 0} = init(Args),
  -    %% Update the state with entries from StateProps
  -    Fields = record_info(fields, state),
  -    FieldPos = lists:zip(Fields, lists:seq(2, length(Fields)+1)),
  -    ModStateFun = fun({Field, Value}, State0) ->
  -                          Pos = proplists:get_value(Field, FieldPos),
  -                          setelement(Pos, State0, Value)
  -                  end,
  -    TestState = lists:foldl(ModStateFun, State, StateProps),
  -    {ok, waiting_file_info, TestState}.
  +prepare(timeout, State) ->
  +    %% Get the latest version
  +    %% of the manifest we're going
  +    %% to delete.
  +    %% TODO: a get specific manifest
  +    %% function needs to be written
  +    %% for the manifest fsm
   
  -%% @doc First state of the delete fsm
  --spec initialize(timeout, state()) ->
  -                        {next_state, waiting_file_info, state(), timeout()} |
  -                        {stop, term(), state()}.
  -initialize(timeout, State=#state{bucket=Bucket,
  -                                 filename=FileName,
  -                                 timeout=Timeout}) ->
  -    %% Start the worker to perform the writing
  -    case start_deleter() of
  -        {ok, DeleterPid} ->
  -            link(DeleterPid),
  -            %% Provide the deleter with the file details
  -            riak_moss_deleter:initialize(DeleterPid,
  -                                         self(),
  -                                         Bucket,
  -                                         FileName),
  -            UpdState = State#state{deleter_pid=DeleterPid},
  -            {next_state, waiting_file_info, UpdState, Timeout};
  -        {error, Reason} ->
  -            _ = lager:error("Failed to start the delete fsm deleter process. Reason: ",
  -                            [Reason]),
  -            {stop, Reason, State}
  -    end.
  +    %% Based on the manifest,
  +    %% fill in the delete_blocks_remaining
  +    %% set and then launch N delete
  +    %% workers (ie. riak_moss_block_server)
   
  -%% @doc State to receive information about the file or object to be
  -%% deleted from the deleter process.
  --spec waiting_file_info({deleter_ready, object |
  -                         {file, non_neg_integer()} |
  -                         {error, term()}},
  -                        state()) ->
  -                               {next_state,
  -                                waiting_root_delete | waiting_blocks_delete,
  -                                state(),
  -                                timeout()}.
  -waiting_file_info({deleter_ready, object}, State=#state{deleter_pid=DeleterPid,
  -                                                        timeout=Timeout}) ->
  -    %% Send request to the deleter to delete the object
  -    riak_moss_deleter:delete_root(DeleterPid),
  -    {next_state, waiting_root_delete, State, Timeout};
  -waiting_file_info({deleter_ready, {file, BlockCount}},
  -                  State=#state{deleter_pid=DeleterPid,
  -                               timeout=Timeout}) ->
  -    riak_moss_deleter:update_root(DeleterPid, set_inactive),
  -    UpdState = State#state{block_count=BlockCount,
  -                           blocks_remaining=BlockCount},
  -    {next_state, waiting_root_update, UpdState, Timeout};
  -waiting_file_info({deleter_ready, {error, Reason}},
  -                  State=#state{deleter_pid=DeleterPid}) ->
  -    _ = lager:warning("The deletion process encountered an error. Reason: ~p", [Reason]),
  -    riak_moss_deleter:stop(DeleterPid),
  -    {stop, normal, State}.
  +    %% start the save_manifest
  +    %% timer
  +    {next_state, deleting, State}.
   
  -%% @doc State to receive notifcation when the root file object has
  -%% been updated to mark the file as inactive.
  --spec waiting_root_update(root_inactive, state()) ->
  -                                 {next_state,
  -                                  waiting_blocks_delete,
  -                                  state(),
  -                                  timeout()}.
  -waiting_root_update(root_inactive, State=#state{block_count=BlockCount,
  -                                                deleter_pid=DeleterPid,
  -                                                timeout=Timeout}) ->
  +deleting(_Event, State) ->
  +    %% We just received a delete
  +    %% acknowledgement, so depending
  +    %% on the value of delete_blocks_remaining
  +    %% (after removing this element,
  +    %% stay in the deleting state,
  +    %% or stop.
  +    %% If stopping, we should also
  +    %% sync wait to update the manifest.
  +    %% Cancel the timer
   
  -    %% Send request to the deleter to delete the file blocks
  -    %% @TODO Backpressure or concurrency of deletion needed. Currently
  -    %% all block delete requests will serialize at the `riak_moss_deleter'.
  -    case BlockCount of
  -        0 ->
  -            _ = riak_moss_deleter:delete_root(DeleterPid),
  -            {next_state, waiting_root_delete, State, Timeout};
  -        _ ->
  -            _ = [riak_moss_deleter:delete_block(DeleterPid, BlockID) ||
  -                    BlockID <- lists:seq(0, BlockCount-1)],
  -            {next_state, waiting_blocks_delete, State, Timeout}
  -    end.
  +    %% or
  +    %% {stop, normal, State}.
  +    {next_state, deleting, State}.
   
  -%% @doc State for waiting for responses about file data blocks
  -%% being deleted.
  --spec waiting_blocks_delete({block_deleted, non_neg_integer()}, state()) ->
  -                                   {next_state,
  -                                    waiting_root_delete | waiting_blocks_delete,
  -                                    state(),
  -                                    non_neg_integer()}.
  -waiting_blocks_delete({block_deleted, _},
  -                      State=#state{blocks_remaining=BlocksRemaining,
  -                                   deleter_pid=DeleterPid,
  -                                   timeout=Timeout}) ->
  -    UpdBlocksRemaining = BlocksRemaining-1,
  -    UpdState = State#state{blocks_remaining=UpdBlocksRemaining},
  -    case UpdBlocksRemaining of
  -        0 ->
  -            %% All blocks deleted, transition to `waiting_root_delete'.
  -            %% @TODO The method for tracking compelted blocks could stand
  -            %% to be more robust.
  -            riak_moss_deleter:delete_root(DeleterPid),
  -            {next_state, waiting_root_delete, UpdState, Timeout};
  -        _ ->
  -            {next_state, waiting_blocks_delete, UpdState, Timeout}
  -    end.
  +handle_event(_Event, StateName, State) ->
  +    {next_state, StateName, State}.
  +>>>>>>> Skeleton for new delete fsm
   
  -%% @doc State for waiting for response about object or file root
  -%% being deleted.
  --spec waiting_root_delete(root_deleted, state()) ->
  -                                 {stop, normal, state()}.
  -waiting_root_delete(root_deleted, State=#state{deleter_pid=DeleterPid}) ->
  -    riak_moss_deleter:stop(DeleterPid),
  -    {stop, normal, State}.
  -
  -%% @doc Handle events that should be handled
  -%% the same regardless of the current state.
  --spec handle_event(term(), atom(), state()) ->
  -                          {stop, badmsg, state()}.
  -handle_event(_Event, _StateName, State) ->
  -    {stop, badmsg, State}.
  -
  -%% @doc Handle synchronous events that should be handled
  -%% the same regardless of the current state.
  --spec handle_sync_event(term(), term(), atom(), state()) ->
  -                               {reply, term(), atom(), state()} |
  -                               {next_state, atom(), state()}.
  -handle_sync_event(current_state, _From, StateName, State) ->
  -    {reply, StateName, StateName, State};
   handle_sync_event(_Event, _From, StateName, State) ->
       {next_state, StateName, State}.
   
  -%% @doc @TODO
  --spec handle_info(term(), atom(), state()) ->
  -                         {next_state, atom(), state(), timeout()} |
  -                         {stop, badmsg, state()}.
  -handle_info({'EXIT', _Pid, _Reason}, StateName, State=#state{timeout=Timeout}) ->
  -    {next_state, StateName, State, Timeout};
  -handle_info({_ReqId, {ok, _Pid}},
  -            StateName,
  -            State=#state{timeout=Timeout}) ->
  -    {next_state, StateName, State, Timeout};
  -handle_info(_Info, _StateName, State) ->
  -    {stop, badmsg, State}.
  +handle_info(save_manifest, StateName, State=#state{mani_pid=ManiPid,
  +                                                   manifest=Manifest}) ->
  +    %% 1. save the manifest
   
  -%% @doc Unused.
  --spec terminate(term(), atom(), state()) -> ok.
  -terminate(Reason, _StateName, _State) ->
  -    Reason.
  +    %% TODO:
  +    %% are there any times where
  +    %% we should be cancelling the
  +    %% timer here, depending on the
  +    %% state we're in?
  +    riak_moss_manifest_fsm:update_manifest(ManiPid, Manifest),
  +    {next_state, StateName, State};
  +handle_info(_Info, StateName, State) ->
  +    {next_state, StateName, State}.
   
  -%% @doc Unused.
  --spec code_change(term(), atom(), state(), term()) ->
  -                         {ok, atom(), state()}.
  +terminate(_Reason, _StateName, _State) ->
  +    ok.
  +
   code_change(_OldVsn, StateName, State, _Extra) ->
       {ok, StateName, State}.
   
  
positive.code: %% Internal functions
  %% ====================================================================
  
  %% @private
  %% @doc Start a `riak_moss_deleter' process to perform the actual work
  %% of writing data to Riak.
  -spec start_deleter() -> {ok, pid()} | {error, term()}.
  start_deleter() ->
      riak_moss_deleter_sup:start_deleter(node(), []).
  
  %% ===================================================================
  %% Test API
  %% ===================================================================
  
  -ifdef(TEST).
  
  %% @doc Start a `riak_moss_delete_fsm' for testing.
  -spec test_link([{atom(), term()}],
                  binary(),
                  string(),
                  timeout()) ->
                         {ok, pid()} | ignore | {error, term()}.
  test_link(StateProps, Bucket, Name, Timeout) ->
      Args = [Bucket, Name, Timeout],
      gen_fsm:start_link(?MODULE, {test, Args, StateProps}, []).
  
  %% @doc Get the current state of the fsm for testing inspection
  -spec current_state(pid()) -> atom() | {error, term()}.
  current_state(Pid) ->
      gen_fsm:sync_send_all_state_event(Pid, current_state, infinity).
  
  -endif.
  
negative.code: %% ===================================================================
  
  %% @doc Start a `riak_moss_delete_fsm'.
  start_link(Bucket, Key, Options) ->
      Args = [Bucket, Key, Options],
      gen_fsm:start_link(?MODULE, Args, []).
  
  %% ====================================================================
  %% gen_fsm callbacks
  