couch_replicator_manager.erl

update_rep_doc(RepDbName, RepDocId, KVs) when is_binary(RepDocId) ->
    {Pid, Ref} =
    spawn_monitor(fun() ->
        try
            case fabric:open_doc(mem3:dbname(RepDbName), RepDocId, []) of
                {ok, LatestRepDoc} ->
                    update_rep_doc(RepDbName, LatestRepDoc, KVs);
                _ ->
                    ok
            end
        catch throw:Error ->
            % Shouldn't happen, as by default only the role _replicator can
            % update replication documents.
            case Error of
            conflict ->
                twig:log(error, "Conflict error when updating replication document `~s`."
                         " Retrying.", [RepDocId]),
                ok = timer:sleep(5),
                update_rep_doc(RepDbName, RepDocId, KVs);
            OtherError ->
                exit(OtherError)
            end
        end
                  end),
    receive
    {'DOWN', Ref, process, Pid, normal} ->
        ok;
    {'DOWN', Ref, process, Pid, Error} ->
        throw(Error)
    end;

update_rep_doc(RepDbName, #doc{body = {RepDocBody}} = RepDoc, KVs) ->
    NewRepDocBody = lists:foldl(
        fun({<<"_replication_state">> = K, State} = KV, Body) ->
                case get_value(K, Body) of
                State ->
                    Body;
                _ ->
                    Body1 = lists:keystore(K, 1, Body, KV),
                    lists:keystore(
                        <<"_replication_state_time">>, 1, Body1,
                        {<<"_replication_state_time">>, timestamp()})
                end;
            ({K, _V} = KV, Body) ->
                lists:keystore(K, 1, Body, KV)
        end,
        RepDocBody, KVs),
    case NewRepDocBody of
    RepDocBody ->
        ok;
    _ ->
        % Might not succeed - when the replication doc is deleted right
        % before this update (not an error, ignore).
        fabric:update_doc(RepDbName, RepDoc#doc{body = {NewRepDocBody}}, [?CTX])
    end.