anchor.diff:      State#state{batch_start=BatchStart,
                   batch=Batch,
                   batch_count=0,
  -                batch_skips=0}.
  +                batch_skips=0,
  +                riak=Riak}.
   
   %% @doc Extract a list of status information from a state record.
   -spec status_data(#state{}) -> [{atom(), term()}].
  
positive.code:         true ->
              NewState = State#state{batch=undefined};
          false ->
              %% this does not check out a worker from the riak
              %% connection pool; instead it creates a fresh new worker,
              %% the idea being that we don't want to delay deletion
              %% just because the normal request pool is empty; pool
              %% workers just happen to be literally the socket process,
              %% so "starting" one here is the same as opening a
              %% connection, and avoids duplicating the configuration
              %% lookup code
              {ok, Riak} = riak_moss_riakc_pool_worker:start_link([]),
              NewState = start_batch(State#state{riak=Riak})
      end,
      {reply, ok, fetching_next_fileset, NewState};
  idle(cancel_batch, _From, State) ->
  
negative.code: 
  %% @doc Actually kick off the batch.  After calling this function, you
  %% must advance the FSM state to `fetching_next_fileset'.
  %% Intentionally pattern match on an undefined Riak handle.
  start_batch(State=#state{riak=undefined}) ->
      %% this does not check out a worker from the riak
      %% connection pool; instead it creates a fresh new worker,
      %% the idea being that we don't want to delay deletion
      %% just because the normal request pool is empty; pool
      %% workers just happen to be literally the socket process,
      %% so "starting" one here is the same as opening a
      %% connection, and avoids duplicating the configuration
      %% lookup code
      {ok, Riak} = riak_moss_riakc_pool_worker:start_link([]),
      BatchStart = riak_cs_gc:timestamp(),
      Batch = fetch_eligible_manifest_keys(Riak, BatchStart),
      _ = lager:debug("Batch keys: ~p", [Batch]),
  