anchor.diff:              Else
       end.
   
  -handle_stream_start({rexi_DOWN, _, {_, NodeRef}, _}, _, State) ->
  -    case fabric_util:remove_down_workers(State, NodeRef) of
  -    {ok, NewState} ->
  -        {ok, NewState};
  +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, State) ->
  -    NewState = fabric_dict:erase(Worker, State),
  -    case fabric_view:is_progress_possible(NewState) of
  -    true ->
  -        {ok, NewState};
  -    false ->
  +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}, State) ->
  -    case fabric_dict:lookup_element(Worker, State) of
  +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, State};
  +        {ok, St};
       waiting ->
           % Don't ack the worker yet so they don't start sending us
           % rows until we're ready
  -        NewState0 = fabric_dict:store(Worker, From, State),
  -        NewState1 = fabric_view:remove_overlapping_shards(Worker, NewState0),
  -        case fabric_dict:any(waiting, NewState1) of
  -            true -> {ok, NewState1};
  -            false -> {stop, NewState1}
  +        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, _, _) ->
  
positive.code: cleanup(Workers) ->
      [rexi:kill(Node, Ref) || #shard{node=Node, ref=Ref} <- Workers].
  
  stream_start(Workers0, Keypos) ->
      Fun = fun handle_stream_start/3,
      Acc = fabric_dict:init(Workers0, waiting),
      Timeout = request_timeout(),
      case rexi_utils:recv(Workers0, Keypos, Fun, Acc, Timeout, infinity) of
          {ok, Workers} ->
              true = fabric_view:is_progress_possible(Workers),
              AckedWorkers = fabric_dict:fold(fun(Worker, From, WorkerAcc) ->
                  rexi:stream_start(From),
  
negative.code:         update_counter/3, remove_ancestors/2, create_monitors/1, kv/2,
          remove_down_workers/2]).
  -export([request_timeout/0]).
  -export([stream_start/2, stream_start/4]).
  
  -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").
  
  -record(stream_acc, {
      workers,
      start_fun,
      replacements
  }).
  
  remove_down_workers(Workers, BadNode) ->
      Filter = fun(#shard{node = Node}, _) -> Node =/= BadNode end,
  