anchor.diff:  cleanup(Workers) ->
       [rexi:kill(Node, Ref) || #shard{node=Node, ref=Ref} <- Workers].
   
  -stream_start(Workers0, Keypos) ->
  +stream_start(Workers, Keypos) ->
  +    stream_start(Workers, Keypos, undefined, undefined).
  +
  +stream_start(Workers0, Keypos, StartFun, Replacements) ->
       Fun = fun handle_stream_start/3,
  -    Acc = fabric_dict:init(Workers0, waiting),
  +    Acc = #stream_acc{
  +        workers = fabric_dict:init(Workers0, waiting),
  +        start_fun = StartFun,
  +        replacements = Replacements
  +    },
       Timeout = request_timeout(),
       case rexi_utils:recv(Workers0, Keypos, Fun, Acc, Timeout, infinity) of
  -        {ok, Workers} ->
  +        {ok, #stream_acc{workers=Workers}} ->
               true = fabric_view:is_progress_possible(Workers),
               AckedWorkers = fabric_dict:fold(fun(Worker, From, WorkerAcc) ->
                   rexi:stream_start(From),
  
positive.code:         update_counter/3, remove_ancestors/2, create_monitors/1, kv/2,
          remove_down_workers/2]).
  -export([request_timeout/0]).
  -export([stream_start/2]).
  
  -include_lib("fabric/include/fabric.hrl").
  -include_lib("mem3/include/mem3.hrl").
  -include_lib("couch/include/couch_db.hrl").
  -include_lib("eunit/include/eunit.hrl").
  
  remove_down_workers(Workers, BadNode) ->
      Filter = fun(#shard{node = Node}, _) -> Node =/= BadNode end,
  
negative.code:             Else
      end.
  
  handle_stream_start({rexi_DOWN, _, {_, NodeRef}, _}, _, St) ->
      case fabric_util:remove_down_workers(St#stream_acc.workers, NodeRef) of
      {ok, Workers} ->
          {ok, St#stream_acc{workers=Workers}};
      error ->
          Reason = {nodedown, <<"progress not possible">>},
          {error, Reason}
      end;
  handle_stream_start({rexi_EXIT, Reason}, Worker, St) ->
      Workers = fabric_dict:erase(Worker, St#stream_acc.workers),
      case {fabric_view:is_progress_possible(Workers), Reason} of
      {true, _} ->
          {ok, St#stream_acc{workers=Workers}};
      {false, {maintenance_mode, _Node}} ->
          % Check if we have replacements for this range
          % and start the new workers if so.
          case lists:keytake(Worker#shard.range, 1, St#stream_acc.replacements) of
              {value, {_Range, WorkerReplacements}, NewReplacements} ->
                  FinalWorkers = lists:foldl(fun(Repl, NewWorkers) ->
                      NewWorker = (St#stream_acc.start_fun)(Repl),
                      fabric_dict:store(NewWorker, waiting, NewWorkers)
                  end, Workers, WorkerReplacements),
                  % Assert that our replaced worker provides us
                  % the oppurtunity to make progress.
                  true = fabric_view:is_progress_possible(FinalWorkers),
                  NewRefs = fabric_dict:fetch_keys(FinalWorkers),
                  {new_refs, NewRefs, St#stream_acc{
                      workers=FinalWorkers,
                      replacements=NewReplacements
                  }};
              false ->
                  % If we progress isn't possible and we don't have any
                  % replacements then we're dead in the water.
                  Error = {nodedown, <<"progress not possible">>},
                  {error, Error}
          end;
      {false, _} ->
          {error, fabric_util:error_info(Reason)}
      end;
  handle_stream_start(rexi_STREAM_INIT, {Worker, From}, St) ->
      case fabric_dict:lookup_element(Worker, St#stream_acc.workers) of
      undefined ->
          % This worker lost the race with other partition copies, terminate
          rexi:stream_cancel(From),
          {ok, St};
      waiting ->
          % Don't ack the worker yet so they don't start sending us
          % rows until we're ready
          Workers0 = fabric_dict:store(Worker, From, St#stream_acc.workers),
          Workers1 = fabric_view:remove_overlapping_shards(Worker, Workers0),
          case fabric_dict:any(waiting, Workers1) of
              true ->
                  {ok, St#stream_acc{workers=Workers1}};
              false ->
                  {stop, St#stream_acc{workers=Workers1}}
          end
      end;
  handle_stream_start(Else, _, _) ->
  