reorganize couch .erl and driver code into rebar layout

1 files changed, 0 insertions, 748 deletions

diff --git a/src/couchdb/couch_rep.erl b/src/couchdb/couch_rep.erl
deleted file mode 100644
index 65573e8c..00000000
--- a/src/couchdb/couch_rep.erl
+++ /dev/null
@@ -1,748 +0,0 @@
-% Licensed under the Apache License, Version 2.0 (the "License"); you may not
-% use this file except in compliance with the License. You may obtain a copy of
-% the License at
-%
-%   http://www.apache.org/licenses/LICENSE-2.0
-%
-% Unless required by applicable law or agreed to in writing, software
-% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-% License for the specific language governing permissions and limitations under
-% the License.
-
--module(couch_rep).
--behaviour(gen_server).
--export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
-    code_change/3]).
-
--export([replicate/2, checkpoint/1]).
-
--include("couch_db.hrl").
-
--record(state, {
-    changes_feed,
-    missing_revs,
-    reader,
-    writer,
-
-    source,
-    target,
-    continuous,
-    create_target,
-    init_args,
-    checkpoint_scheduled = nil,
-
-    start_seq,
-    history,
-    source_log,
-    target_log,
-    rep_starttime,
-    src_starttime,
-    tgt_starttime,
-    checkpoint_history = nil,
-
-    listeners = [],
-    complete = false,
-    committed_seq = 0,
-
-    stats = nil,
-    doc_ids = nil
-}).
-
-%% convenience function to do a simple replication from the shell
-replicate(Source, Target) when is_list(Source) ->
-    replicate(?l2b(Source), Target);
-replicate(Source, Target) when is_binary(Source), is_list(Target) ->
-    replicate(Source, ?l2b(Target));
-replicate(Source, Target) when is_binary(Source), is_binary(Target) ->
-    replicate({[{<<"source">>, Source}, {<<"target">>, Target}]}, #user_ctx{});
-
-%% function handling POST to _replicate
-replicate({Props}=PostBody, UserCtx) ->
-    {BaseId, Extension} = make_replication_id(PostBody, UserCtx),
-    Replicator = {BaseId ++ Extension,
-        {gen_server, start_link, [?MODULE, [BaseId, PostBody, UserCtx], []]},
-        temporary,
-        1,
-        worker,
-        [?MODULE]
-    },
-
-    case couch_util:get_value(<<"cancel">>, Props, false) of
-    true ->
- case supervisor:terminate_child(couch_rep_sup, BaseId ++ Extension) of
-        {error, not_found} ->
-     {error, not_found};
-        ok ->
-     ok = supervisor:delete_child(couch_rep_sup, BaseId ++ Extension),
-            {ok, {cancelled, ?l2b(BaseId)}}
- end;
-    false ->
-        Server = start_replication_server(Replicator),
-
-        case couch_util:get_value(<<"continuous">>, Props, false) of
-        true ->
-            {ok, {continuous, ?l2b(BaseId)}};
-        false ->
-            get_result(Server, PostBody, UserCtx)
-        end
-    end.
-
-checkpoint(Server) ->
-    gen_server:cast(Server, do_checkpoint).
-
-get_result(Server, PostBody, UserCtx) ->
-    try gen_server:call(Server, get_result, infinity) of
-    retry -> replicate(PostBody, UserCtx);
-    Else -> Else
-    catch
-    exit:{noproc, {gen_server, call, [Server, get_result , infinity]}} ->
-        %% oops, this replication just finished -- restart it.
-        replicate(PostBody, UserCtx);
-    exit:{normal, {gen_server, call, [Server, get_result , infinity]}} ->
-        %% we made the call during terminate
-        replicate(PostBody, UserCtx)
-    end.
-
-init(InitArgs) ->
-    try do_init(InitArgs)
-    catch throw:{db_not_found, DbUrl} -> {stop, {db_not_found, DbUrl}} end.
-
-do_init([RepId, {PostProps}, UserCtx] = InitArgs) ->
-    process_flag(trap_exit, true),
-
-    SourceProps = couch_util:get_value(<<"source">>, PostProps),
-    TargetProps = couch_util:get_value(<<"target">>, PostProps),
-
-    DocIds = couch_util:get_value(<<"doc_ids">>, PostProps, nil),
-    Continuous = couch_util:get_value(<<"continuous">>, PostProps, false),
-    CreateTarget = couch_util:get_value(<<"create_target">>, PostProps, false),
-
-    ProxyParams = parse_proxy_params(
-        couch_util:get_value(<<"proxy">>, PostProps, [])),
-    Source = open_db(SourceProps, UserCtx, ProxyParams),
-    Target = open_db(TargetProps, UserCtx, ProxyParams, CreateTarget),
-
-    SourceInfo = dbinfo(Source),
-    TargetInfo = dbinfo(Target),
-
-    case DocIds of
-    List when is_list(List) ->
-        % Fast replication using only a list of doc IDs to replicate.
-        % Replication sessions, checkpoints and logs are not created
-        % since the update sequence number of the source DB is not used
-        % for determining which documents are copied into the target DB.
-        SourceLog = nil,
-        TargetLog = nil,
-
-        StartSeq = nil,
-        History = nil,
-
-        ChangesFeed = nil,
-        MissingRevs = nil,
-
-        {ok, Reader} =
-        couch_rep_reader:start_link(self(), Source, DocIds, PostProps);
-
-    _ ->
-        % Replication using the _changes API (DB sequence update numbers).
-        SourceLog = open_replication_log(Source, RepId),
-        TargetLog = open_replication_log(Target, RepId),
-    
-        {StartSeq, History} = compare_replication_logs(SourceLog, TargetLog),
-
-        {ok, ChangesFeed} =
-        couch_rep_changes_feed:start_link(self(), Source, StartSeq, PostProps),
-        {ok, MissingRevs} =
-        couch_rep_missing_revs:start_link(self(), Target, ChangesFeed, PostProps),
-        {ok, Reader} =
-        couch_rep_reader:start_link(self(), Source, MissingRevs, PostProps)
-    end,
-
-    {ok, Writer} =
-    couch_rep_writer:start_link(self(), Target, Reader, PostProps),
-
-    Stats = ets:new(replication_stats, [set, private]),
-    ets:insert(Stats, {total_revs,0}),
-    ets:insert(Stats, {missing_revs, 0}),
-    ets:insert(Stats, {docs_read, 0}),
-    ets:insert(Stats, {docs_written, 0}),
-    ets:insert(Stats, {doc_write_failures, 0}),
-
-    {ShortId, _} = lists:split(6, RepId),
-    couch_task_status:add_task("Replication", io_lib:format("~s: ~s -> ~s",
-        [ShortId, dbname(Source), dbname(Target)]), "Starting"),
-
-    State = #state{
-        changes_feed = ChangesFeed,
-        missing_revs = MissingRevs,
-        reader = Reader,
-        writer = Writer,
-
-        source = Source,
-        target = Target,
-        continuous = Continuous,
-        create_target = CreateTarget,
-        init_args = InitArgs,
-        stats = Stats,
-        checkpoint_scheduled = nil,
-
-        start_seq = StartSeq,
-        history = History,
-        source_log = SourceLog,
-        target_log = TargetLog,
-        rep_starttime = httpd_util:rfc1123_date(),
-        src_starttime = couch_util:get_value(instance_start_time, SourceInfo),
-        tgt_starttime = couch_util:get_value(instance_start_time, TargetInfo),
-        doc_ids = DocIds
-    },
-    {ok, State}.
-
-handle_call(get_result, From, #state{complete=true, listeners=[]} = State) ->
-    {stop, normal, State#state{listeners=[From]}};
-handle_call(get_result, From, State) ->
-    Listeners = State#state.listeners,
-    {noreply, State#state{listeners=[From|Listeners]}}.
-
-handle_cast(do_checkpoint, State) ->
-    {noreply, do_checkpoint(State)};
-
-handle_cast(_Msg, State) ->
-    {noreply, State}.
-
-handle_info({missing_revs_checkpoint, SourceSeq}, State) ->
-    couch_task_status:update("MR Processed source update #~p", [SourceSeq]),
-    {noreply, schedule_checkpoint(State#state{committed_seq = SourceSeq})};
-
-handle_info({writer_checkpoint, SourceSeq}, #state{committed_seq=N} = State)
-        when SourceSeq > N ->
-    MissingRevs = State#state.missing_revs,
-    ok = gen_server:cast(MissingRevs, {update_committed_seq, SourceSeq}),
-    couch_task_status:update("W Processed source update #~p", [SourceSeq]),
-    {noreply, schedule_checkpoint(State#state{committed_seq = SourceSeq})};
-handle_info({writer_checkpoint, _}, State) ->
-    {noreply, State};
-
-handle_info({update_stats, Key, N}, State) ->
-    ets:update_counter(State#state.stats, Key, N),
-    {noreply, State};
-
-handle_info({'DOWN', _, _, _, _}, State) ->
-    ?LOG_INFO("replication terminating because local DB is shutting down", []),
-    timer:cancel(State#state.checkpoint_scheduled),
-    {stop, shutdown, State};
-
-handle_info({'EXIT', Writer, normal}, #state{writer=Writer} = State) ->
-    case State#state.listeners of
-    [] ->
-        {noreply, State#state{complete = true}};
-    _Else ->
-        {stop, normal, State}
-    end;
-
-handle_info({'EXIT', _, normal}, State) ->
-    {noreply, State};
-handle_info({'EXIT', _Pid, {Err, Reason}}, State) when Err == source_error;
-        Err == target_error ->
-    ?LOG_INFO("replication terminating due to ~p: ~p", [Err, Reason]),
-    timer:cancel(State#state.checkpoint_scheduled),
-    {stop, shutdown, State};
-handle_info({'EXIT', _Pid, Reason}, State) ->
-    {stop, Reason, State}.
-
-terminate(normal, #state{checkpoint_scheduled=nil} = State) ->
-    do_terminate(State);
-    
-terminate(normal, State) ->
-    timer:cancel(State#state.checkpoint_scheduled),
-    do_terminate(do_checkpoint(State));
-
-terminate(Reason, State) ->
-    #state{
-        listeners = Listeners,
-        source = Source,
-        target = Target,
-        stats = Stats
-    } = State,
-    [gen_server:reply(L, {error, Reason}) || L <- Listeners],
-    ets:delete(Stats),
-    close_db(Target),
-    close_db(Source).
-
-code_change(_OldVsn, State, _Extra) ->
-    {ok, State}.
-
-% internal funs
-
-start_replication_server(Replicator) ->
-    RepId = element(1, Replicator),
-    case supervisor:start_child(couch_rep_sup, Replicator) of
-    {ok, Pid} ->
-        ?LOG_INFO("starting new replication ~p at ~p", [RepId, Pid]),
-        Pid;
-    {error, already_present} ->
-        case supervisor:restart_child(couch_rep_sup, RepId) of
-        {ok, Pid} ->
-            ?LOG_INFO("starting replication ~p at ~p", [RepId, Pid]),
-            Pid;
-        {error, running} ->
-            %% this error occurs if multiple replicators are racing
-            %% each other to start and somebody else won.  Just grab
-            %% the Pid by calling start_child again.
-            {error, {already_started, Pid}} =
-                supervisor:start_child(couch_rep_sup, Replicator),
-            ?LOG_DEBUG("replication ~p already running at ~p", [RepId, Pid]),
-            Pid;
-        {error, {db_not_found, DbUrl}} ->
-            throw({db_not_found, <<"could not open ", DbUrl/binary>>})
-        end;
-    {error, {already_started, Pid}} ->
-        ?LOG_DEBUG("replication ~p already running at ~p", [RepId, Pid]),
-        Pid;
-    {error, {{db_not_found, DbUrl}, _}} ->
-        throw({db_not_found, <<"could not open ", DbUrl/binary>>})
-    end.
-
-compare_replication_logs(SrcDoc, TgtDoc) ->
-    #doc{body={RepRecProps}} = SrcDoc,
-    #doc{body={RepRecPropsTgt}} = TgtDoc,
-    case couch_util:get_value(<<"session_id">>, RepRecProps) ==
-            couch_util:get_value(<<"session_id">>, RepRecPropsTgt) of
-    true ->
-        % if the records have the same session id,
-        % then we have a valid replication history
-        OldSeqNum = couch_util:get_value(<<"source_last_seq">>, RepRecProps, 0),
-        OldHistory = couch_util:get_value(<<"history">>, RepRecProps, []),
-        {OldSeqNum, OldHistory};
-    false ->
-        SourceHistory = couch_util:get_value(<<"history">>, RepRecProps, []),
-        TargetHistory = couch_util:get_value(<<"history">>, RepRecPropsTgt, []),
-        ?LOG_INFO("Replication records differ. "
-                "Scanning histories to find a common ancestor.", []),
-        ?LOG_DEBUG("Record on source:~p~nRecord on target:~p~n",
-                [RepRecProps, RepRecPropsTgt]),
-        compare_rep_history(SourceHistory, TargetHistory)
-    end.
-
-compare_rep_history(S, T) when S =:= [] orelse T =:= [] ->
-    ?LOG_INFO("no common ancestry -- performing full replication", []),
-    {0, []};
-compare_rep_history([{S}|SourceRest], [{T}|TargetRest]=Target) ->
-    SourceId = couch_util:get_value(<<"session_id">>, S),
-    case has_session_id(SourceId, Target) of
-    true ->
-        RecordSeqNum = couch_util:get_value(<<"recorded_seq">>, S, 0),
-        ?LOG_INFO("found a common replication record with source_seq ~p",
-            [RecordSeqNum]),
-        {RecordSeqNum, SourceRest};
-    false ->
-        TargetId = couch_util:get_value(<<"session_id">>, T),
-        case has_session_id(TargetId, SourceRest) of
-        true ->
-            RecordSeqNum = couch_util:get_value(<<"recorded_seq">>, T, 0),
-            ?LOG_INFO("found a common replication record with source_seq ~p",
-                [RecordSeqNum]),
-            {RecordSeqNum, TargetRest};
-        false ->
-            compare_rep_history(SourceRest, TargetRest)
-        end
-    end.
-
-close_db(#http_db{}) ->
-    ok;
-close_db(Db) ->
-    couch_db:close(Db).
-
-dbname(#http_db{url = Url}) ->
-    strip_password(Url);
-dbname(#db{name = Name}) ->
-    Name.
-
-strip_password(Url) ->
-    re:replace(Url,
-        "http(s)?://([^:]+):[^@]+@(.*)$",
-        "http\\1://\\2:*****@\\3",
-        [{return, list}]).
-
-dbinfo(#http_db{} = Db) ->
-    {DbProps} = couch_rep_httpc:request(Db),
-    [{list_to_existing_atom(?b2l(K)), V} || {K,V} <- DbProps];
-dbinfo(Db) ->
-    {ok, Info} = couch_db:get_db_info(Db),
-    Info.
-
-do_terminate(#state{doc_ids=DocIds} = State) when is_list(DocIds) ->
-    #state{
-        listeners = Listeners,
-        rep_starttime = ReplicationStartTime,
-        stats = Stats
-    } = State,
-
-    RepByDocsJson = {[
-        {<<"start_time">>, ?l2b(ReplicationStartTime)},
-        {<<"end_time">>, ?l2b(httpd_util:rfc1123_date())},
-        {<<"docs_read">>, ets:lookup_element(Stats, docs_read, 2)},
-        {<<"docs_written">>, ets:lookup_element(Stats, docs_written, 2)},
-        {<<"doc_write_failures">>,
-            ets:lookup_element(Stats, doc_write_failures, 2)}
-    ]},
-
-    terminate_cleanup(State),
-    [gen_server:reply(L, {ok, RepByDocsJson}) || L <- lists:reverse(Listeners)];
-
-do_terminate(State) ->
-    #state{
-        checkpoint_history = CheckpointHistory,
-        committed_seq = NewSeq,
-        listeners = Listeners,
-        source = Source,
-        continuous = Continuous,
-        source_log = #doc{body={OldHistory}}
-    } = State,
-    
-    NewRepHistory = case CheckpointHistory of
-    nil ->
-        {[{<<"no_changes">>, true} | OldHistory]};
-    _Else ->
-        CheckpointHistory
-    end,
-
-    %% reply to original requester
-    OtherListeners = case Continuous of
-    true ->
-        []; % continuous replications have no listeners
-    _ ->
-        [Original|Rest] = lists:reverse(Listeners),
-        gen_server:reply(Original, {ok, NewRepHistory}),
-        Rest
-    end,
-
-    %% maybe trigger another replication. If this replicator uses a local
-    %% source Db, changes to that Db since we started will not be included in
-    %% this pass.
-    case up_to_date(Source, NewSeq) of
-        true ->
-            [gen_server:reply(R, {ok, NewRepHistory}) || R <- OtherListeners];
-        false ->
-            [gen_server:reply(R, retry) || R <- OtherListeners]
-    end,
-    terminate_cleanup(State).
-
-terminate_cleanup(#state{source=Source, target=Target, stats=Stats}) ->
-    couch_task_status:update("Finishing"),
-    close_db(Target),
-    close_db(Source),
-    ets:delete(Stats).
-
-has_session_id(_SessionId, []) ->
-    false;
-has_session_id(SessionId, [{Props} | Rest]) ->
-    case couch_util:get_value(<<"session_id">>, Props, nil) of
-    SessionId ->
-        true;
-    _Else ->
-        has_session_id(SessionId, Rest)
-    end.
-
-maybe_append_options(Options, Props) ->
-    lists:foldl(fun(Option, Acc) ->
-        Acc ++
-        case couch_util:get_value(Option, Props, false) of
-        true ->
-            "+" ++ ?b2l(Option);
-        false ->
-            ""
-        end
-    end, [], Options).
-
-make_replication_id({Props}, UserCtx) ->
-    %% funky algorithm to preserve backwards compatibility
-    {ok, HostName} = inet:gethostname(),
-    % Port = mochiweb_socket_server:get(couch_httpd, port),
-    Src = get_rep_endpoint(UserCtx, couch_util:get_value(<<"source">>, Props)),
-    Tgt = get_rep_endpoint(UserCtx, couch_util:get_value(<<"target">>, Props)),
-    Base = [HostName, Src, Tgt] ++
-        case couch_util:get_value(<<"filter">>, Props) of
-        undefined ->
-            case couch_util:get_value(<<"doc_ids">>, Props) of
-            undefined ->
-                [];
-            DocIds ->
-                [DocIds]
-            end;
-        Filter ->
-            [Filter, couch_util:get_value(<<"query_params">>, Props, {[]})]
-        end,
-    Extension = maybe_append_options(
-        [<<"continuous">>, <<"create_target">>], Props),
-    {couch_util:to_hex(couch_util:md5(term_to_binary(Base))), Extension}.
-
-maybe_add_trailing_slash(Url) ->
-    re:replace(Url, "[^/]$", "&/", [{return, list}]).
-
-get_rep_endpoint(_UserCtx, {Props}) ->
-    Url = maybe_add_trailing_slash(couch_util:get_value(<<"url">>, Props)),
-    {BinHeaders} = couch_util:get_value(<<"headers">>, Props, {[]}),
-    {Auth} = couch_util:get_value(<<"auth">>, Props, {[]}),
-    case couch_util:get_value(<<"oauth">>, Auth) of
-    undefined ->
-        {remote, Url, [{?b2l(K),?b2l(V)} || {K,V} <- BinHeaders]};
-    {OAuth} ->
-        {remote, Url, [{?b2l(K),?b2l(V)} || {K,V} <- BinHeaders], OAuth}
-    end;
-get_rep_endpoint(_UserCtx, <<"http://",_/binary>>=Url) ->
-    {remote, maybe_add_trailing_slash(Url), []};
-get_rep_endpoint(_UserCtx, <<"https://",_/binary>>=Url) ->
-    {remote, maybe_add_trailing_slash(Url), []};
-get_rep_endpoint(UserCtx, <<DbName/binary>>) ->
-    {local, DbName, UserCtx}.
-
-open_replication_log(#http_db{}=Db, RepId) ->
-    DocId = ?LOCAL_DOC_PREFIX ++ RepId,
-    Req = Db#http_db{resource=couch_util:url_encode(DocId)},
-    case couch_rep_httpc:request(Req) of
-    {[{<<"error">>, _}, {<<"reason">>, _}]} ->
-        ?LOG_DEBUG("didn't find a replication log for ~s", [Db#http_db.url]),
-        #doc{id=?l2b(DocId)};
-    Doc ->
-        ?LOG_DEBUG("found a replication log for ~s", [Db#http_db.url]),
-        couch_doc:from_json_obj(Doc)
-    end;
-open_replication_log(Db, RepId) ->
-    DocId = ?l2b(?LOCAL_DOC_PREFIX ++ RepId),
-    case couch_db:open_doc(Db, DocId, []) of
-    {ok, Doc} ->
-        ?LOG_DEBUG("found a replication log for ~s", [Db#db.name]),
-        Doc;
-    _ ->
-        ?LOG_DEBUG("didn't find a replication log for ~s", [Db#db.name]),
-        #doc{id=DocId}
-    end.
-
-open_db(Props, UserCtx, ProxyParams) ->
-    open_db(Props, UserCtx, ProxyParams, false).
-
-open_db({Props}, _UserCtx, ProxyParams, CreateTarget) ->
-    Url = maybe_add_trailing_slash(couch_util:get_value(<<"url">>, Props)),
-    {AuthProps} = couch_util:get_value(<<"auth">>, Props, {[]}),
-    {BinHeaders} = couch_util:get_value(<<"headers">>, Props, {[]}),
-    Headers = [{?b2l(K),?b2l(V)} || {K,V} <- BinHeaders],
-    DefaultHeaders = (#http_db{})#http_db.headers,
-    Db1 = #http_db{
-        url = Url,
-        auth = AuthProps,
-        headers = lists:ukeymerge(1, Headers, DefaultHeaders)
-    },
-    Db = Db1#http_db{options = Db1#http_db.options ++ ProxyParams},
-    couch_rep_httpc:db_exists(Db, CreateTarget);
-open_db(<<"http://",_/binary>>=Url, _, ProxyParams, CreateTarget) ->
-    open_db({[{<<"url">>,Url}]}, [], ProxyParams, CreateTarget);
-open_db(<<"https://",_/binary>>=Url, _, ProxyParams, CreateTarget) ->
-    open_db({[{<<"url">>,Url}]}, [], ProxyParams, CreateTarget);
-open_db(<<DbName/binary>>, UserCtx, _ProxyParams, CreateTarget) ->
-    case CreateTarget of
-    true ->
-        ok = couch_httpd:verify_is_server_admin(UserCtx),
-        couch_server:create(DbName, [{user_ctx, UserCtx}]);
-    false -> ok
-    end,
-
-    case couch_db:open(DbName, [{user_ctx, UserCtx}]) of
-    {ok, Db} ->
-        couch_db:monitor(Db),
-        Db;
-    {not_found, no_db_file} -> throw({db_not_found, DbName})
-    end.
-
-schedule_checkpoint(#state{checkpoint_scheduled = nil} = State) ->
-    Server = self(),
-    case timer:apply_after(5000, couch_rep, checkpoint, [Server]) of
-    {ok, TRef} ->
-        State#state{checkpoint_scheduled = TRef};
-    Error ->
-        ?LOG_ERROR("tried to schedule a checkpoint but got ~p", [Error]),
-        State
-    end;
-schedule_checkpoint(State) ->
-    State.
-
-do_checkpoint(State) ->
-    #state{
-        source = Source,
-        target = Target,
-        committed_seq = NewSeqNum,
-        start_seq = StartSeqNum,
-        history = OldHistory,
-        source_log = SourceLog,
-        target_log = TargetLog,
-        rep_starttime = ReplicationStartTime,
-        src_starttime = SrcInstanceStartTime,
-        tgt_starttime = TgtInstanceStartTime,
-        stats = Stats
-    } = State,
-    case commit_to_both(Source, Target, NewSeqNum) of
-    {SrcInstanceStartTime, TgtInstanceStartTime} ->
-        ?LOG_INFO("recording a checkpoint for ~s -> ~s at source update_seq ~p",
-            [dbname(Source), dbname(Target), NewSeqNum]),
-        SessionId = couch_uuids:random(),
-        NewHistoryEntry = {[
-            {<<"session_id">>, SessionId},
-            {<<"start_time">>, list_to_binary(ReplicationStartTime)},
-            {<<"end_time">>, list_to_binary(httpd_util:rfc1123_date())},
-            {<<"start_last_seq">>, StartSeqNum},
-            {<<"end_last_seq">>, NewSeqNum},
-            {<<"recorded_seq">>, NewSeqNum},
-            {<<"missing_checked">>, ets:lookup_element(Stats, total_revs, 2)},
-            {<<"missing_found">>, ets:lookup_element(Stats, missing_revs, 2)},
-            {<<"docs_read">>, ets:lookup_element(Stats, docs_read, 2)},
-            {<<"docs_written">>, ets:lookup_element(Stats, docs_written, 2)},
-            {<<"doc_write_failures">>,
-                ets:lookup_element(Stats, doc_write_failures, 2)}
-        ]},
-        % limit history to 50 entries
-        NewRepHistory = {[
-            {<<"session_id">>, SessionId},
-            {<<"source_last_seq">>, NewSeqNum},
-            {<<"history">>, lists:sublist([NewHistoryEntry | OldHistory], 50)}
-        ]},
-
-        try
-        {SrcRevPos,SrcRevId} =
-            update_local_doc(Source, SourceLog#doc{body=NewRepHistory}),
-        {TgtRevPos,TgtRevId} =
-            update_local_doc(Target, TargetLog#doc{body=NewRepHistory}),
-        State#state{
-            checkpoint_scheduled = nil,
-            checkpoint_history = NewRepHistory,
-            source_log = SourceLog#doc{revs={SrcRevPos, [SrcRevId]}},
-            target_log = TargetLog#doc{revs={TgtRevPos, [TgtRevId]}}
-        }
-        catch throw:conflict ->
-        ?LOG_ERROR("checkpoint failure: conflict (are you replicating to "
-            "yourself?)", []),
-        State
-        end;
-    _Else ->
-        ?LOG_INFO("rebooting ~s -> ~s from last known replication checkpoint",
-            [dbname(Source), dbname(Target)]),
-        #state{
-            changes_feed = CF,
-            missing_revs = MR,
-            reader = Reader,
-            writer = Writer
-        } = State,
-        Pids = [Writer, Reader, MR, CF],
-        [unlink(Pid) || Pid <- Pids],
-        [exit(Pid, shutdown) || Pid <- Pids],
-        close_db(Target),
-        close_db(Source),
-        {ok, NewState} = init(State#state.init_args),
-        NewState#state{listeners=State#state.listeners}
-    end.
-
-commit_to_both(Source, Target, RequiredSeq) ->
-    % commit the src async
-    ParentPid = self(),
-    SrcCommitPid = spawn_link(fun() ->
-            ParentPid ! {self(), ensure_full_commit(Source, RequiredSeq)} end),
-
-    % commit tgt sync
-    TargetStartTime = ensure_full_commit(Target),
-
-    SourceStartTime =
-    receive
-    {SrcCommitPid, Timestamp} ->
-        Timestamp;
-    {'EXIT', SrcCommitPid, {http_request_failed, _}} ->
-        exit(replication_link_failure)
-    end,
-    {SourceStartTime, TargetStartTime}.
-    
-ensure_full_commit(#http_db{headers = Headers} = Target) ->
-    Req = Target#http_db{
-        resource = "_ensure_full_commit",
-        method = post,
-        headers = couch_util:proplist_apply_field({"Content-Type", "application/json"}, Headers)
-    },
-    {ResultProps} = couch_rep_httpc:request(Req),
-    true = couch_util:get_value(<<"ok">>, ResultProps),
-    couch_util:get_value(<<"instance_start_time">>, ResultProps);
-ensure_full_commit(Target) ->
-    {ok, NewDb} = couch_db:open_int(Target#db.name, []),
-    UpdateSeq = couch_db:get_update_seq(Target),
-    CommitSeq = couch_db:get_committed_update_seq(NewDb),
-    InstanceStartTime = NewDb#db.instance_start_time,
-    couch_db:close(NewDb),
-    if UpdateSeq > CommitSeq ->
-        ?LOG_DEBUG("target needs a full commit: update ~p commit ~p",
-            [UpdateSeq, CommitSeq]),
-        {ok, DbStartTime} = couch_db:ensure_full_commit(Target),
-        DbStartTime;
-    true ->
-        ?LOG_DEBUG("target doesn't need a full commit", []),
-        InstanceStartTime
-    end.
-
-ensure_full_commit(#http_db{headers = Headers} = Source, RequiredSeq) ->
-    Req = Source#http_db{
-        resource = "_ensure_full_commit",
-        method = post,
-        qs = [{seq, RequiredSeq}],
-        headers = couch_util:proplist_apply_field({"Content-Type", "application/json"}, Headers)
-    },
-    {ResultProps} = couch_rep_httpc:request(Req),
-    case couch_util:get_value(<<"ok">>, ResultProps) of
-    true ->
-        couch_util:get_value(<<"instance_start_time">>, ResultProps);
-    undefined -> nil end;
-ensure_full_commit(Source, RequiredSeq) ->
-    {ok, NewDb} = couch_db:open_int(Source#db.name, []),
-    CommitSeq = couch_db:get_committed_update_seq(NewDb),
-    InstanceStartTime = NewDb#db.instance_start_time,
-    couch_db:close(NewDb),
-    if RequiredSeq > CommitSeq ->
-        ?LOG_DEBUG("source needs a full commit: required ~p committed ~p",
-            [RequiredSeq, CommitSeq]),
-        {ok, DbStartTime} = couch_db:ensure_full_commit(Source),
-        DbStartTime;
-    true ->
-        ?LOG_DEBUG("source doesn't need a full commit", []),
-        InstanceStartTime
-    end.
-
-update_local_doc(#http_db{} = Db, #doc{id=DocId} = Doc) ->
-    Req = Db#http_db{
-        resource = couch_util:url_encode(DocId),
-        method = put,
-        body = couch_doc:to_json_obj(Doc, [attachments]),
-        headers = [{"x-couch-full-commit", "false"} | Db#http_db.headers]
-    },
-    {ResponseMembers} = couch_rep_httpc:request(Req),
-    Rev = couch_util:get_value(<<"rev">>, ResponseMembers),
-    couch_doc:parse_rev(Rev);
-update_local_doc(Db, Doc) ->
-    {ok, Result} = couch_db:update_doc(Db, Doc, [delay_commit]),
-    Result.
-
-up_to_date(#http_db{}, _Seq) ->
-    true;
-up_to_date(Source, Seq) ->
-    {ok, NewDb} = couch_db:open_int(Source#db.name, []),
-    T = NewDb#db.update_seq == Seq,
-    couch_db:close(NewDb),
-    T.
-
-parse_proxy_params(ProxyUrl) when is_binary(ProxyUrl) ->
-    parse_proxy_params(?b2l(ProxyUrl));
-parse_proxy_params([]) ->
-    [];
-parse_proxy_params(ProxyUrl) ->
-    {url, _, Base, Port, User, Passwd, _Path, _Proto} =
-        ibrowse_lib:parse_url(ProxyUrl),
-    [{proxy_host, Base}, {proxy_port, Port}] ++
-        case is_list(User) andalso is_list(Passwd) of
-        false ->
-            [];
-        true ->
-            [{proxy_user, User}, {proxy_password, Passwd}]
-        end.