anchor.diff:  %%%===================================================================
   
   %% @doc Actually kick off the batch.  After calling this function, you
  -%% must advance the FSM state to `calculating'.
  +%% must advance the FSM state to `fetching_next_fileset'.
   start_batch(_Options, State) ->
       BatchStart = riak_cs_gc:timestamp(),
   
  
positive.code: fetching_next_fileset(status, _From, State) ->
      Props = [{interval, State#state.interval},
               {last, State#state.last},
               {current, State#state.current},
               {next, State#state.next},
               {elapsed, elapsed(State#state.batch_start)},
               {files_deleted, State#state.batch_count},
               {files_skipped, State#state.batch_skips},
               {files_left, length(State#state.batch)}],
      {reply, {ok, {calculating, Props}}, calculating, State};
  fetching_next_fileset({manual_batch, _Options}, _From, State) ->
      %% this is the manual user request to begin a batch
      {reply, {error, already_deleting}, deleting, State};
  fetching_next_fileset(pause_batch, _From, State) ->
      _ = lager:info("Pausing garbage collection"),
      {reply, ok, paused, State};
  fetching_next_fileset(cancel_batch, _From, #state{current=Current}=State) ->
      %% finished with this batch
      _ = lager:info("Canceled garbage collection batch after ~b seconds.",
                     [elapsed(State#state.batch_start)]),
      riak_moss_riakc_pool_worker:stop(State#state.riak),
      NewState = State#state{riak=undefined,
                             last=Current,
                             current=undefined,
                             batch=[]},
      {reply, ok, idle, NewState};
  fetching_next_fileset(_, _From, State) ->
      {reply, ok, deleting, State}.
  
  
  paused(status, From, State) ->
      {reply, {ok, {_, Status}}, _, State} = fetching_next_fileset(status, From, State),
      {reply, {ok, {paused, Status}}, paused, State};
  paused(resume_batch, _From, State) ->
      _ = lager:info("Resuming garbage collection"),
      gen_fsm:send_event(?SERVER, continue),
      {reply, ok, calculating, State};
  paused(cancel_batch, From, State) ->
      fetching_next_fileset(cancel_batch, From, State);
  paused(_, _From, State) ->
      {reply, ok, paused, State}.
  
  waiting_file_delete(done, _From, State) ->
      gen_fsm:send_event(?SERVER, continue),
      {reply, ok, initiating_file_delete, State};
  waiting_file_delete(_, _From, State) ->
      {reply, ok, waiting_file_delete, State}.
  
  %% @doc there are no all-state events for this fsm
  handle_event(_Event, StateName, State) ->
  
negative.code: 
  -record(state, {
            interval :: non_neg_integer(),
            last :: undefined | calendar:datetime(), % the last time a deletion was scheduled
            next :: undefined | calendar:datetime(), % the next scheduled gc time
            riak :: pid(), % Riak connection pid
            current_fileset :: [lfs_manifest()],
            batch_start :: undefined | calendar:datetime(), % start of the current gc interval
            batch_count=0 :: non_neg_integer(),
            batch_skips=0 :: non_neg_integer(),
            batch=[] :: [twop_set:twop_set()],
            pause_state :: atom(), % state of the fsm when a delete batch was paused
            delete_fsm_pid :: pid()
           }).
  
  %%%===================================================================
  