anchor.diff:                {<<"worker_processes">>, 1},
                 {<<"http_connections">>, 10},
                 {<<"connection_timeout">>, 300000},
  -              {<<"continuous">>, true}
  +              {<<"continuous">>, false}
                ]},
               #user_ctx{roles = [<<"_admin">>]}),
   
  -    ok = timer:sleep(Wait * 1000),
       case couch_replicator:async_replicate(CRep) of
  -    {ok, CRepPid} = Result ->
  +    {ok, CRepPid} ->
           erlang:monitor(process, CRepPid),
           CRepState = triggered,
  -        true = ets:insert(?CSTORE, {CRepPid, CRep, Vb, CRepState, Wait}),
  +        true = ets:insert(?CSTORE, {CRepPid, CRep, Vb, CRepState}),
           true = ets:insert(?X2CSTORE, {XDocId, CRepPid}),
  +        couch_replication_manager:update_rep_doc(
  +            xdc_rep_utils:info_doc_id(XDocId),
  +            [{?l2b("replication_state_vb_" ++ ?i2l(Vb)), <<"triggered">>}]),
           ?log_info("~s: triggered replication for vbucket ~p", [XDocId, Vb]),
  -        Result;
  +        {ok, CRepPid};
       Error ->
           ?log_info("~s: triggering of replication for vbucket ~p failed due to: "
                     "~p", [XDocId, Vb, Error]),
  -        Error
  +        {error, Error}
       end.
   
   
   cancel_couch_replication(XDocId, CRepPid) ->
  -    [CRep, Vb] =
  -        lists:flatten(ets:match(?CSTORE, {CRepPid, '$1', '$2', '_', '_'})),
  -
  -    case couch_replicator:cancel_replication(CRep#rep.id) of
  -    {ok, _Res} ->
  -        ?log_info("~s: cancelled replication for vbucket ~p", [XDocId, Vb]);
  -    Error ->
  -        ?log_info("~s: cancellation of replication for vbucket ~p failed due "
  -                  "to: ~p", [XDocId, Vb, Error])
  +    [CRep, CRepState] =
  +        lists:flatten(ets:match(?CSTORE, {CRepPid, '$1', '_', '$2'})),
  +    case CRepState of
  +    triggered ->
  +        couch_replicator:cancel_replication(CRep#rep.id);
  +    _ ->
  +        ok
       end,
  -
       true = ets:delete(?CSTORE, CRepPid),
       true = ets:delete_object(?X2CSTORE, {XDocId, CRepPid}),
       ok.
   
   
  -maybe_retry_all_couch_replications() ->
  +manage_vbucket_replications() ->
       lists:foreach(
           fun([XDocId,
                #rep{source = SrcBucket,
                     target = {_, TgtBucket, _, _, _, _, _, _, _, _}},
  -             TriggeredVbs, UntriggeredVbs]) ->
  -
  -            FailedCouchReps = failed_couch_replications(XDocId),
  -            case {FailedCouchReps, UntriggeredVbs} of
  -            {[], []} ->
  -                ?log_info("~s: no failed vbucket replications found", [XDocId]);
  -            {_, _} ->
  +             Vbs]) ->
  +
  +            MaxConcurrentReps = misc:getenv_int("MAX_CONCURRENT_REPS_PER_DOC",
  +                                                ?MAX_CONCURRENT_REPS_PER_DOC),
  +            NumActiveReps = length(active_replications_for_doc(XDocId)),
  +            case (MaxConcurrentReps - NumActiveReps) of
  +            0 ->
  +                ok;
  +            NumFreeSlots ->
  +                {Vbs1, Vbs2} = lists:split(NumFreeSlots, Vbs),
  +
                   % Reread the target vbucket map once before retrying all reps
  -                {ok, {TgtVbMap, TgtNodes} = TgtVbInfo} =
  +                {ok, TgtVbInfo} =
                       xdc_rep_utils:remote_vbucketmap_nodelist(TgtBucket),
   
  -                % Retry couch replications that failed after they were triggered
  -                Success = lists:foldl(
  -                    fun(FailedCouchRep, Success0) ->
  -                        retry_couch_replication(XDocId, SrcBucket, TgtVbInfo,
  -                                                FailedCouchRep) and Success0
  -                    end,
  -                    true, FailedCouchReps),
  -
  -                % Retry couch replications that did not trigger at all
  -                {TriggeredVbs1, UntriggeredVbs1} = start_vbucket_replications(
  -                    XDocId, SrcBucket, TgtVbMap, TgtNodes, UntriggeredVbs, 0),
  -                true = ets:update_element(
  -                    ?XSTORE, XDocId, [{3, (TriggeredVbs ++ TriggeredVbs1)},
  -                                      {4, UntriggeredVbs1}]),
  -                couch_replication_manager:update_rep_doc(
  -                    xdc_rep_utils:info_doc_id(XDocId),
  -                    xdc_rep_utils:vb_rep_state_list(
  -                        TriggeredVbs1, <<"triggered">>)),
  -
  -                case {Success, UntriggeredVbs1} of
  -                {true, []} ->
  -                    couch_replication_manager:update_rep_doc(
  -                        xdc_rep_utils:info_doc_id(XDocId),
  -                        [{<<"_replication_state">>, <<"triggered">>}]);
  -                _ ->
  -                    ok
  -                end
  +                lists:foreach(
  +                    fun(Vb) ->
  +                        case lists:flatten(ets:match(
  +                            ?CSTORE, {'$1', '_', Vb, '$2'})) of
  +                        [] ->
  +                            % New vbucket
  +                            start_vbucket_replication(XDocId, SrcBucket,
  +                                TgtVbInfo, Vb);
  +                        [CRepPid, error] ->
  +                            % Failed vbucket
  +                            restart_vbucket_replication(XDocId, SrcBucket,
  +                                TgtVbInfo, Vb, CRepPid);
  +                        [CRepPid, completed] ->
  +                            % Completed vbucket
  +                            restart_vbucket_replication(XDocId, SrcBucket,
  +                                TgtVbInfo, Vb, CRepPid);
  +                        [_, triggered] ->
  +                            % In progress vbucket
  +                            ok
  +                        end
  +                    end, Vbs1),
  +
  +                % Update XSTORE with the new vbuckets list
  +                true = ets:update_element(?XSTORE, XDocId, [{3, (Vbs2 ++ Vbs1)}])
               end
           end,
  -        ets:match(?XSTORE, {'$1', '$2', '$3', '$4'})).
  -
  -
  -retry_couch_replication(XDocId,
  -                        SrcBucket,
  -                        {TgtVbMap, TgtNodes},
  -                        {CRepPid, Vb, Wait}) ->
  -    cancel_couch_replication(XDocId, CRepPid),
  -
  -    NewWait = erlang:max(?INITIAL_WAIT, trunc(Wait * 2)),
  -    SrcURI = xdc_rep_utils:local_couch_uri_for_vbucket(SrcBucket, Vb),
  -    TgtURI = xdc_rep_utils:remote_couch_uri_for_vbucket(TgtVbMap, TgtNodes, Vb),
  -    ?log_info("~s: will retry replication for vbucket ~p after ~p seconds",
  -              [XDocId, Vb, NewWait]),
  -    case start_couch_replication(SrcURI, TgtURI, Vb, XDocId, NewWait) of
  -    {ok, _Res} ->
  -        couch_replication_manager:update_rep_doc(
  -            xdc_rep_utils:info_doc_id(XDocId),
  -            [{?l2b("replication_state_vb_" ++ ?i2l(Vb)), <<"triggered">>}]),
  -        true;
  -    _Error ->
  -        false
  -    end.
  -
  -
  -couch_replication_completed(XDocId, CRepPid, Vb) ->
  -    true = ets:delete(?CSTORE, CRepPid),
  -    true = ets:delete_object(?X2CSTORE, {XDocId, CRepPid}),
  -    ?log_info("~s: replication of vbucket ~p completed", [XDocId, Vb]),
  -
  -    case ets:lookup(?X2CSTORE, XDocId) of
  -    [] ->
  -        % All couch replications for this XDocId have completed.
  -        true = ets:delete(?XSTORE, XDocId),
  -        couch_replication_manager:update_rep_doc(
  -            xdc_rep_utils:info_doc_id(XDocId),
  -            [{?l2b("replication_state_vb_" ++ ?i2l(Vb)), <<"completed">>},
  -             {<<"_replication_state">>, <<"completed">>}]),
  -        ?log_info("~s: replication of all vbuckets completed", [XDocId]);
  -    _ ->
  -        couch_replication_manager:update_rep_doc(
  -            xdc_rep_utils:info_doc_id(XDocId),
  -            [{?l2b("replication_state_vb_" ++ ?i2l(Vb)), <<"completed">>}])
  -    end.
  -
  -
  -failed_couch_replications(XDocId) ->
  +        ets:match(?XSTORE, {'$1', '$2', '$3'})).
  +
  +
  +active_replications_for_doc(XDocId) ->
       % For the given XDocId:
       % 1. Lookup all CRepPids in X2CSTORE.
       % 2. For each CRepPid, lookup its record in CSTORE.
  -    % 3. Only select records whose State attribute equals 'error'.
  -    % 4. Filter out CRep and State in the final output and just return the
  -    %    CRepPid, Vb and Wait attributes.
  +    % 3. Only select records whose State attribute equals 'triggered'.
  +    % 4. Return the list of records found.
       Pids =
           try
               ets:lookup_element(?X2CSTORE, XDocId, 2)
  
positive.code:                             end,
      ns_pubsub:subscribe_link(ns_config_events, NsConfigEventsHandler, []),
  
      % Periodically check whether any Couch replications have failed due to
      % errors and, if so, restart them. Among other reasons, Couch replications
      % may fail due to rebalance and failover operations at the destination.
      % Checking periodically in this manner allows us to batch several failed
      % replications together and only read the destination vbucket map once
      % before retrying all of them.
      {ok, _Tref} = timer:send_interval(?RETRY_INTERVAL * 1000,
                                        retry_failed_couch_replications),
  
      <<"_replicator">> = ?l2b(couch_config:get("replicator", "db",
                                                "_replicator")),
  
negative.code:         xdc_rep_utils:vb_rep_state_list(LostVbs, <<"cancelled">>)),
  
      % Update XSTORE with the current active vbuckets list
      true = ets:update_element(?XSTORE, XDocId, [{3, CurrVbs}]).
  
  
  start_vbucket_replication(XDocId, SrcBucket, {TgtVbMap, TgtNodes}, Vb) ->
      SrcURI = xdc_rep_utils:local_couch_uri_for_vbucket(SrcBucket, Vb),
      TgtURI = xdc_rep_utils:remote_couch_uri_for_vbucket(TgtVbMap, TgtNodes, Vb),
      start_couch_replication(SrcURI, TgtURI, Vb, XDocId).
  
  
  restart_vbucket_replication(XDocId, SrcBucket, TgtVbInfo, Vb, CRepPid) ->
      cancel_couch_replication(XDocId, CRepPid),
      start_vbucket_replication(XDocId, SrcBucket, TgtVbInfo, Vb).
  
  %
  % Couch specific functions
  %
  
  start_couch_replication(SrcCouchURI, TgtCouchURI, Vb, XDocId) ->
      % Until scheduled XDC replication support is added, this function will
      % trigger continuous replication by default at the Couch level.
      {ok, CRep} =
  