anchor.diff:              ?xdcr_error("replication mode: ~p, unable to replicate ~p docs to target ~p",
                           [RepMode, length(DocsList), misc:sanitize_url(Target#httpdb.url)]),
               exit({failed_write, Error})
  -    end;
  -
  -flush_docs_helper(Target, DocsList, XMemLoc) ->
  +    end.
  +
  +do_flush_docs_xmem(Target, DocsList, XMemLoc) ->
       BeforeTS = case ?x_trace_enabled() of
                      true ->
                          os:timestamp();
  
positive.code:         {error, Msg} ->
              {failed_write, Msg}
      end.
  
  -spec maybe_flush_docs_xmem(any(), #batch{}, #dcp_mutation{},
                              integer(), integer()) -> {#batch{}, integer()}.
  maybe_flush_docs_xmem(XMemLoc, Batch, Mutation, BatchSize, BatchItems) ->
      #batch{docs = DocAcc, size = SizeAcc, items = ItemsAcc} = Batch,
  
      DocSize = case Mutation#dcp_mutation.deleted of
                    true ->
                        0;
                    _ ->
                        iolist_size(Mutation#dcp_mutation.body)
                end,
  
      SizeAcc2 = SizeAcc + DocSize,
      ItemsAcc2 = ItemsAcc + 1,
  
      %% if reach the limit in terms of docs, flush them
      case ItemsAcc2 >= BatchItems orelse SizeAcc2 >= BatchSize of
          true ->
              DocsList = [Mutation | DocAcc],
              ok = flush_docs_xmem(XMemLoc, DocsList),
              ?x_trace(flushed, []),
              %% data flushed, return empty batch and size of # of docs flushed
              {#batch{}, SizeAcc2};
          _ ->            %% no data flushed in this turn, return the new batch
              {#batch{docs = [Mutation | DocAcc], size = SizeAcc2, items = ItemsAcc2}, 0}
      end.
  
negative.code:                              OptRepThreshold, BatchSize, BatchItems, XMemLoc)
      end.
  
  send_xmem_batch(Docs, Target, XMemLoc) ->
      ok = do_flush_docs_xmem(Target, Docs, XMemLoc),
      {ok, compute_body_sizes_loop(Docs, 0)}.
  
  compute_body_sizes_loop([], Acc) ->
      Acc;
  compute_body_sizes_loop([Doc | Rest], Acc) ->
      case Doc#dcp_mutation.deleted of
          true ->
              compute_body_sizes_loop(Rest, Acc);
          _ ->
              NewAcc = iolist_size(Doc#dcp_mutation.body) + Acc,
              compute_body_sizes_loop(Rest, NewAcc)
      end.
  
  local_process_batch([], _Tgt, #batch{docs = []}, _BatchSize, _BatchItems, Acc) ->
      {ok, Acc};
  local_process_batch([], #httpdb{} = Target,
                      #batch{docs = Docs, size = Size}, BatchSize, BatchItems, Acc) ->
      ok = flush_docs_helper(Target, Docs),
      local_process_batch([], Target, #batch{}, BatchSize, BatchItems, Acc + Size);
  
  local_process_batch([Mutation | Rest],
                      #httpdb{} = Target, Batch, BatchSize, BatchItems, Acc) ->
      #dcp_mutation{id = Key,
                    rev = Rev,
                    body = Body,
                    deleted = Deleted} = Mutation,
      Doc0 = couch_doc:from_binary(Key, Body, true),
      Doc = Doc0#doc{rev = Rev,
                     deleted = Deleted},
      {Batch2, DataFlushed} = maybe_flush_docs_capi(Target, Batch, Doc, BatchSize, BatchItems),
      local_process_batch(Rest, Target, Batch2, BatchSize, BatchItems, Acc + DataFlushed).
  
  
  -spec flush_docs_helper(any(), list()) -> ok.
  flush_docs_helper(Target, DocsList) ->
      BeforeTS = case ?x_trace_enabled() of
                     true ->
                         os:timestamp();
  