anchor.diff:  
   maybe_adjust_all_replications(BucketConfigs) ->
       lists:foreach(
  -        fun([XDocId,
  -             #rep{source = SrcBucket,
  -                  target = {_, TgtBucket, _, _, _, _, _, _, _, _}},
  -             PrevTriggeredVbs,
  -             PrevUntriggeredVbs]) ->
  +        fun([XDocId, #rep{source = SrcBucket}, PrevVbs]) ->
               SrcBucketConfig =
                   proplists:get_value(?b2l(SrcBucket), BucketConfigs),
               CurrVbs = xdc_rep_utils:my_active_vbuckets(SrcBucketConfig),
  -            maybe_adjust_xdc_replication(
  -                XDocId, SrcBucket, TgtBucket, PrevTriggeredVbs,
  -                PrevUntriggeredVbs, CurrVbs)
  +            maybe_adjust_xdc_replication(XDocId, PrevVbs, CurrVbs)
           end,
  -        ets:match(?XSTORE, {'$1', '$2', '$3', '$4'})).
  -
  -
  -maybe_adjust_xdc_replication(XDocId, SrcBucket, TgtBucket, PrevTriggeredVbs,
  -                             PrevUntriggeredVbs, CurrVbs) ->
  -    PrevVbs = PrevTriggeredVbs ++ PrevUntriggeredVbs,
  -    {AcquiredVbs, LostVbs} = xdc_rep_utils:lists_difference(CurrVbs, PrevVbs),
  -
  -    % Start Couch replications for the newly acquired vbuckets
  -    {ok, {TgtVbMap, TgtNodes}} =
  -        xdc_rep_utils:remote_vbucketmap_nodelist(TgtBucket),
  -    {TriggeredVbs1, UntriggeredVbs1} = start_vbucket_replications(
  -        XDocId, SrcBucket, TgtVbMap, TgtNodes, AcquiredVbs, 0),
  -
  -    % Add entries for the new Couch replications to the replication info doc
  -    couch_replication_manager:update_rep_doc(
  -        xdc_rep_utils:info_doc_id(XDocId),
  -        xdc_rep_utils:vb_rep_state_list(TriggeredVbs1, <<"triggered">>)),
  +        ets:match(?XSTORE, {'$1', '$2', '$3'})).
  +
  +
  +maybe_adjust_xdc_replication(XDocId, PrevVbs, CurrVbs) ->
  +    {_AcquiredVbs, LostVbs} = xdc_rep_utils:lists_difference(CurrVbs, PrevVbs),
   
       % Cancel Couch replications for the lost vbuckets
       CRepPidsToCancel =
  
positive.code:               {<<"worker_processes">>, 1},
                {<<"http_connections">>, 10},
                {<<"connection_timeout">>, 300000},
                {<<"continuous">>, true}
               ]},
              #user_ctx{roles = [<<"_admin">>]}),
  
      ok = timer:sleep(Wait * 1000),
      case couch_replicator:async_replicate(CRep) of
      {ok, CRepPid} = Result ->
          erlang:monitor(process, CRepPid),
          CRepState = triggered,
          true = ets:insert(?CSTORE, {CRepPid, CRep, Vb, CRepState, Wait}),
          true = ets:insert(?X2CSTORE, {XDocId, CRepPid}),
          ?log_info("~s: triggered replication for vbucket ~p", [XDocId, Vb]),
          Result;
      Error ->
          ?log_info("~s: triggering of replication for vbucket ~p failed due to: "
                    "~p", [XDocId, Vb, 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])
      end,
  
      true = ets:delete(?CSTORE, CRepPid),
      true = ets:delete_object(?X2CSTORE, {XDocId, CRepPid}),
      ok.
  
  
  maybe_retry_all_couch_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]);
              {_, _} ->
                  % Reread the target vbucket map once before retrying all reps
                  {ok, {TgtVbMap, TgtNodes} = 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
              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) ->
      % 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.
      Pids =
          try
              ets:lookup_element(?X2CSTORE, XDocId, 2)
  
negative.code:         error:badarg ->
              []
          end,
      [{CRepPid} ||
          {CRepPid, _CRep, _, State} <-
              lists:flatten([(ets:lookup(?CSTORE, Pid)) || Pid <- Pids]),
          State == triggered].
  
  
  %
  % XDC replication and replication info document related functions
  %
  
  create_xdc_rep_info_doc(XDocId, {Base, Ext}, Vbs, RepDbName, XDocBody) ->
      IDocId = xdc_rep_utils:info_doc_id(XDocId),
      UserCtx = #user_ctx{roles = [<<"_admin">>, <<"_replicator">>]},
      {ok, RepDb} = couch_db:open(RepDbName, [sys_db, {user_ctx, UserCtx}]),
  
      VbStates =  xdc_rep_utils:vb_rep_state_list(Vbs, <<"undefined">>),
      Body = {[
               {<<"node">>, xdc_rep_utils:node_uuid()},
               {<<"replication_doc_id">>, XDocId},
  