{"action":"created","comment":{"url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls/comments/64261042","id":64261042,"diff_hunk":"@@ -103,935 +68,272 @@ replicate(PostBody, Ctx) ->\n     end.\n \n \n-do_replication_loop(#rep{id = {BaseId, Ext} = Id, options = Options} = Rep) ->\n-    case async_replicate(Rep) of\n-    {ok, _Pid} ->\n-        case get_value(continuous, Options, false) of\n-        true ->\n-            {ok, {continuous, ?l2b(BaseId ++ Ext)}};\n-        false ->\n-            wait_for_result(Id)\n-        end;\n-    Error ->\n-        Error\n-    end.\n-\n-\n-async_replicate(#rep{id = {BaseId, Ext}, source = Src, target = Tgt} = Rep) ->\n-    RepChildId = BaseId ++ Ext,\n-    Source = couch_replicator_api_wrap:db_uri(Src),\n-    Target = couch_replicator_api_wrap:db_uri(Tgt),\n-    Timeout = get_value(connection_timeout, Rep#rep.options),\n-    ChildSpec = {\n-        RepChildId,\n-        {gen_server, start_link, [?MODULE, Rep, [{timeout, Timeout}]]},\n-        temporary,\n-        250,\n-        worker,\n-        [?MODULE]\n-    },\n-    % All these nested cases to attempt starting/restarting a replication child\n-    % are ugly and not 100% race condition free. The following patch submission\n-    % is a solution:\n-    %\n-    % http://erlang.2086793.n4.nabble.com/PATCH-supervisor-atomically-delete-child-spec-when-child-terminates-td3226098.html\n-    %\n-    case supervisor:start_child(couch_replicator_job_sup, ChildSpec) of\n-    {ok, Pid} ->\n-        couch_log:notice(\"starting new replication `~s` at ~p (`~s` -> `~s`)\",\n-            [RepChildId, Pid, Source, Target]),\n-        {ok, Pid};\n-    {error, already_present} ->\n-        case supervisor:restart_child(couch_replicator_job_sup, RepChildId) of\n-        {ok, Pid} ->\n-            couch_log:notice(\"restarting replication `~s` at ~p (`~s` -> `~s`)\",\n-                [RepChildId, Pid, Source, Target]),\n-            {ok, Pid};\n-        {error, running} ->\n-            %% this error occurs if multiple replicators are racing\n-            %% each other to start and somebody else won. Just grab\n-            %% the Pid by calling start_child again.\n-            timer:sleep(50 + random:uniform(100)),\n-            async_replicate(Rep);\n-        {error, {'EXIT', {badarg,\n-            [{erlang, apply, [gen_server, start_link, undefined]} | _]}}} ->\n-            % Clause to deal with a change in the supervisor module introduced\n-            % in R14B02. For more details consult the thread at:\n-            %     http://erlang.org/pipermail/erlang-bugs/2011-March/002273.html\n-            _ = supervisor:delete_child(couch_replicator_job_sup, RepChildId),\n-            async_replicate(Rep);\n-        {error, _} = Error ->\n-            Error\n-        end;\n-    {error, {already_started, Pid}} ->\n-        couch_log:notice(\"replication `~s` already running at ~p (`~s` -> `~s`)\",\n-            [RepChildId, Pid, Source, Target]),\n-        {ok, Pid};\n-    {error, {Error, _}} ->\n-        {error, Error}\n-    end.\n-\n-\n-rep_result_listener(RepId) ->\n-    ReplyTo = self(),\n-    {ok, _Listener} = couch_replicator_notifier:start_link(\n-        fun({_, RepId2, _} = Ev) when RepId2 =:= RepId ->\n-                ReplyTo ! Ev;\n-            (_) ->\n-                ok\n-        end).\n-\n-\n-wait_for_result(RepId) ->\n-    receive\n-    {finished, RepId, RepResult} ->\n-        {ok, RepResult};\n-    {error, RepId, Reason} ->\n-        {error, Reason}\n-    end.\n-\n-\n-cancel_replication({BaseId, Extension}) ->\n-    FullRepId = BaseId ++ Extension,\n-    couch_log:notice(\"Canceling replication `~s`...\", [FullRepId]),\n-    case supervisor:terminate_child(couch_replicator_job_sup, FullRepId) of\n-    ok ->\n-        couch_log:notice(\"Replication `~s` canceled.\", [FullRepId]),\n-        case supervisor:delete_child(couch_replicator_job_sup, FullRepId) of\n-            ok ->\n-                {ok, {cancelled, ?l2b(FullRepId)}};\n-            {error, not_found} ->\n-                {ok, {cancelled, ?l2b(FullRepId)}};\n-            Error ->\n-                Error\n-        end;\n-    Error ->\n-        couch_log:error(\"Error canceling replication `~s`: ~p\", [FullRepId, Error]),\n-        Error\n-    end.\n-\n-cancel_replication(RepId, #user_ctx{name = Name, roles = Roles}) ->\n-    case lists:member(<<\"_admin\">>, Roles) of\n-    true ->\n-        cancel_replication(RepId);\n-    false ->\n-        case find_replicator(RepId) of\n-        {ok, Pid} ->\n-            case details(Pid) of\n-            {ok, #rep{user_ctx = #user_ctx{name = Name}}} ->\n-                cancel_replication(RepId);\n-            {ok, _} ->\n-                throw({unauthorized,\n-                    <<\"Can't cancel a replication triggered by another user\">>});\n-            Error ->\n-                Error\n-            end;\n-        Error ->\n-            Error\n-        end\n-    end.\n-\n-find_replicator({BaseId, Ext} = _RepId) ->\n-    case lists:keysearch(\n-        BaseId ++ Ext, 1, supervisor:which_children(couch_replicator_job_sup)) of\n-    {value, {_, Pid, _, _}} when is_pid(Pid) ->\n-            {ok, Pid};\n-    _ ->\n-            {error, not_found}\n-    end.\n-\n-details(Pid) ->\n-    case (catch gen_server:call(Pid, get_details)) of\n-    {ok, Rep} ->\n-        {ok, Rep};\n-    {'EXIT', {noproc, {gen_server, call, _}}} ->\n-        {error, not_found};\n-    Error ->\n-        throw(Error)\n-    end.\n-\n-init(InitArgs) ->\n-    {ok, InitArgs, 0}.\n-\n-do_init(#rep{options = Options, id = {BaseId, Ext}, user_ctx=UserCtx} = Rep) ->\n-    process_flag(trap_exit, true),\n-\n-    random:seed(os:timestamp()),\n-\n-    #rep_state{\n-        source = Source,\n-        target = Target,\n-        source_name = SourceName,\n-        target_name = TargetName,\n-        start_seq = {_Ts, StartSeq},\n-        committed_seq = {_, CommittedSeq},\n-        highest_seq_done = {_, HighestSeq},\n-        checkpoint_interval = CheckpointInterval\n-    } = State = init_state(Rep),\n-\n-    NumWorkers = get_value(worker_processes, Options),\n-    BatchSize = get_value(worker_batch_size, Options),\n-    {ok, ChangesQueue} = couch_work_queue:new([\n-        {max_items, BatchSize * NumWorkers * 2},\n-        {max_size, 100 * 1024 * NumWorkers}\n-    ]),\n-    % This starts the _changes reader process. It adds the changes from\n-    % the source db to the ChangesQueue.\n-    {ok, ChangesReader} = couch_replicator_changes_reader:start_link(\n-        StartSeq, Source, ChangesQueue, Options\n-    ),\n-    % Changes manager - responsible for dequeing batches from the changes queue\n-    % and deliver them to the worker processes.\n-    ChangesManager = spawn_changes_manager(self(), ChangesQueue, BatchSize),\n-    % This starts the worker processes. They ask the changes queue manager for a\n-    % a batch of _changes rows to process -> check which revs are missing in the\n-    % target, and for the missing ones, it copies them from the source to the target.\n-    MaxConns = get_value(http_connections, Options),\n-    Workers = lists:map(\n-        fun(_) ->\n-            couch_stats:increment_counter([couch_replicator, workers_started]),\n-            {ok, Pid} = couch_replicator_worker:start_link(\n-                self(), Source, Target, ChangesManager, MaxConns),\n-            Pid\n-        end,\n-        lists:seq(1, NumWorkers)),\n-\n-    couch_task_status:add_task([\n-        {type, replication},\n-        {user, UserCtx#user_ctx.name},\n-        {replication_id, ?l2b(BaseId ++ Ext)},\n-        {database, Rep#rep.db_name},\n-        {doc_id, Rep#rep.doc_id},\n-        {source, ?l2b(SourceName)},\n-        {target, ?l2b(TargetName)},\n-        {continuous, get_value(continuous, Options, false)},\n-        {revisions_checked, 0},\n-        {missing_revisions_found, 0},\n-        {docs_read, 0},\n-        {docs_written, 0},\n-        {changes_pending, get_pending_count(State)},\n-        {doc_write_failures, 0},\n-        {source_seq, HighestSeq},\n-        {checkpointed_source_seq, CommittedSeq},\n-        {checkpoint_interval, CheckpointInterval}\n-    ]),\n-    couch_task_status:set_update_frequency(1000),\n-\n-    % Until OTP R14B03:\n-    %\n-    % Restarting a temporary supervised child implies that the original arguments\n-    % (#rep{} record) specified in the MFA component of the supervisor\n-    % child spec will always be used whenever the child is restarted.\n-    % This implies the same replication performance tunning parameters will\n-    % always be used. The solution is to delete the child spec (see\n-    % cancel_replication/1) and then start the replication again, but this is\n-    % unfortunately not immune to race conditions.\n-\n-    couch_log:notice(\"Replication `~p` is using:~n\"\n-        \"~c~p worker processes~n\"\n-        \"~ca worker batch size of ~p~n\"\n-        \"~c~p HTTP connections~n\"\n-        \"~ca connection timeout of ~p milliseconds~n\"\n-        \"~c~p retries per request~n\"\n-        \"~csocket options are: ~s~s\",\n-        [BaseId ++ Ext, $\\t, NumWorkers, $\\t, BatchSize, $\\t,\n-            MaxConns, $\\t, get_value(connection_timeout, Options),\n-            $\\t, get_value(retries, Options),\n-            $\\t, io_lib:format(\"~p\", [get_value(socket_options, Options)]),\n-            case StartSeq of\n-            ?LOWEST_SEQ ->\n-                \"\";\n-            _ ->\n-                io_lib:format(\"~n~csource start sequence ~p\", [$\\t, StartSeq])\n-            end]),\n-\n-    couch_log:debug(\"Worker pids are: ~p\", [Workers]),\n-\n-    couch_replicator_manager:replication_started(Rep),\n-\n-    {ok, State#rep_state{\n-            changes_queue = ChangesQueue,\n-            changes_manager = ChangesManager,\n-            changes_reader = ChangesReader,\n-            workers = Workers\n-        }\n-    }.\n-\n-adjust_maxconn(Src = #httpdb{http_connections = 1}, RepId) ->\n-    Msg = \"Adjusting minimum number of HTTP source connections to 2 for ~p\",\n-    couch_log:notice(Msg, [RepId]),\n-    Src#httpdb{http_connections = 2};\n-\n-adjust_maxconn(Src, _RepId) ->\n-    Src.\n+% This gen_server is only used to create local _replicator db then exits by\n+% returning `ignore` from init.\n \n-handle_info(shutdown, St) ->\n-    {stop, shutdown, St};\n+start_link() ->\n+    gen_server:start_link(?MODULE, [], []).\n \n-handle_info({'DOWN', Ref, _, _, Why}, #rep_state{source_monitor = Ref} = St) ->\n-    couch_log:error(\"Source database is down. Reason: ~p\", [Why]),\n-    {stop, source_db_down, St};\n+init(_) ->\n+    couch_replicator_docs:ensure_rep_db_exists(),\n+    couch_log:notice(\"~p : created local _replicator database\", [?MODULE]),\n+    ignore.\n \n-handle_info({'DOWN', Ref, _, _, Why}, #rep_state{target_monitor = Ref} = St) ->\n-    couch_log:error(\"Target database is down. Reason: ~p\", [Why]),\n-    {stop, target_db_down, St};\n+handle_call(_, _From, State) ->\n+    {noreply, State}.\n \n-handle_info({'EXIT', Pid, normal}, #rep_state{changes_reader=Pid} = State) ->\n-    {noreply, State};\n-\n-handle_info({'EXIT', Pid, Reason}, #rep_state{changes_reader=Pid} = State) ->\n-    couch_stats:increment_counter([couch_replicator, changes_reader_deaths]),\n-    couch_log:error(\"ChangesReader process died with reason: ~p\", [Reason]),\n-    {stop, changes_reader_died, cancel_timer(State)};\n-\n-handle_info({'EXIT', Pid, normal}, #rep_state{changes_manager = Pid} = State) ->\n-    {noreply, State};\n-\n-handle_info({'EXIT', Pid, Reason}, #rep_state{changes_manager = Pid} = State) ->\n-    couch_stats:increment_counter([couch_replicator, changes_manager_deaths]),\n-    couch_log:error(\"ChangesManager process died with reason: ~p\", [Reason]),\n-    {stop, changes_manager_died, cancel_timer(State)};\n-\n-handle_info({'EXIT', Pid, normal}, #rep_state{changes_queue=Pid} = State) ->\n-    {noreply, State};\n-\n-handle_info({'EXIT', Pid, Reason}, #rep_state{changes_queue=Pid} = State) ->\n-    couch_stats:increment_counter([couch_replicator, changes_queue_deaths]),\n-    couch_log:error(\"ChangesQueue process died with reason: ~p\", [Reason]),\n-    {stop, changes_queue_died, cancel_timer(State)};\n-\n-handle_info({'EXIT', Pid, normal}, #rep_state{workers = Workers} = State) ->\n-    case Workers -- [Pid] of\n-    Workers ->\n-        couch_log:error(\"unknown pid bit the dust ~p ~n\",[Pid]),\n-        {noreply, State#rep_state{workers = Workers}};\n-        %% not clear why a stop was here before\n-        %%{stop, {unknown_process_died, Pid, normal}, State};\n-    [] ->\n-        catch unlink(State#rep_state.changes_manager),\n-        catch exit(State#rep_state.changes_manager, kill),\n-        do_last_checkpoint(State);\n-    Workers2 ->\n-        {noreply, State#rep_state{workers = Workers2}}\n-    end;\n-\n-handle_info({'EXIT', Pid, Reason}, #rep_state{workers = Workers} = State) ->\n-    State2 = cancel_timer(State),\n-    case lists:member(Pid, Workers) of\n-    false ->\n-        {stop, {unknown_process_died, Pid, Reason}, State2};\n-    true ->\n-        couch_stats:increment_counter([couch_replicator, worker_deaths]),\n-        couch_log:error(\"Worker ~p died with reason: ~p\", [Pid, Reason]),\n-        {stop, {worker_died, Pid, Reason}, State2}\n-    end;\n-\n-handle_info(timeout, InitArgs) ->\n-    try do_init(InitArgs) of {ok, State} ->\n-        {noreply, State}\n-    catch Class:Error ->\n-        Stack = erlang:get_stacktrace(),\n-        {stop, shutdown, {error, Class, Error, Stack, InitArgs}}\n-    end.\n-\n-handle_call(get_details, _From, #rep_state{rep_details = Rep} = State) ->\n-    {reply, {ok, Rep}, State};\n-\n-handle_call({add_stats, Stats}, From, State) ->\n-    gen_server:reply(From, ok),\n-    NewStats = couch_replicator_utils:sum_stats(State#rep_state.stats, Stats),\n-    {noreply, State#rep_state{stats = NewStats}};\n-\n-handle_call({report_seq_done, Seq, StatsInc}, From,\n-    #rep_state{seqs_in_progress = SeqsInProgress, highest_seq_done = HighestDone,\n-        current_through_seq = ThroughSeq, stats = Stats} = State) ->\n-    gen_server:reply(From, ok),\n-    {NewThroughSeq0, NewSeqsInProgress} = case SeqsInProgress of\n-    [] ->\n-        {Seq, []};\n-    [Seq | Rest] ->\n-        {Seq, Rest};\n-    [_ | _] ->\n-        {ThroughSeq, ordsets:del_element(Seq, SeqsInProgress)}\n-    end,\n-    NewHighestDone = lists:max([HighestDone, Seq]),\n-    NewThroughSeq = case NewSeqsInProgress of\n-    [] ->\n-        lists:max([NewThroughSeq0, NewHighestDone]);\n-    _ ->\n-        NewThroughSeq0\n-    end,\n-    couch_log:debug(\"Worker reported seq ~p, through seq was ~p, \"\n-        \"new through seq is ~p, highest seq done was ~p, \"\n-        \"new highest seq done is ~p~n\"\n-        \"Seqs in progress were: ~p~nSeqs in progress are now: ~p\",\n-        [Seq, ThroughSeq, NewThroughSeq, HighestDone,\n-            NewHighestDone, SeqsInProgress, NewSeqsInProgress]),\n-    NewState = State#rep_state{\n-        stats = couch_replicator_utils:sum_stats(Stats, StatsInc),\n-        current_through_seq = NewThroughSeq,\n-        seqs_in_progress = NewSeqsInProgress,\n-        highest_seq_done = NewHighestDone\n-    },\n-    update_task(NewState),\n-    {noreply, NewState}.\n-\n-\n-handle_cast({db_compacted, DbName},\n-    #rep_state{source = #db{name = DbName} = Source} = State) ->\n-    {ok, NewSource} = couch_db:reopen(Source),\n-    {noreply, State#rep_state{source = NewSource}};\n-\n-handle_cast({db_compacted, DbName},\n-    #rep_state{target = #db{name = DbName} = Target} = State) ->\n-    {ok, NewTarget} = couch_db:reopen(Target),\n-    {noreply, State#rep_state{target = NewTarget}};\n-\n-handle_cast(checkpoint, State) ->\n-    #rep_state{rep_details = #rep{} = Rep} = State,\n-    case couch_replicator_manager:continue(Rep) of\n-    {true, _} ->\n-        case do_checkpoint(State) of\n-        {ok, NewState} ->\n-            couch_stats:increment_counter([couch_replicator, checkpoints, success]),\n-            {noreply, NewState#rep_state{timer = start_timer(State)}};\n-        Error ->\n-            couch_stats:increment_counter([couch_replicator, checkpoints, failure]),\n-            {stop, Error, State}\n-        end;\n-    {false, Owner} ->\n-        couch_replicator_manager:replication_usurped(Rep, Owner),\n-        {stop, shutdown, State}\n-    end;\n+handle_cast(_, State) ->\n+    {noreply, State}.\n \n-handle_cast({report_seq, Seq},\n-    #rep_state{seqs_in_progress = SeqsInProgress} = State) ->\n-    NewSeqsInProgress = ordsets:add_element(Seq, SeqsInProgress),\n-    {noreply, State#rep_state{seqs_in_progress = NewSeqsInProgress}}.\n+handle_info(_, State) ->\n+    {noreply, State}.\n \n+terminate(_Reason, _State) ->\n+    ok.\n \n-code_change(_OldVsn, #rep_state{}=State, _Extra) ->\n+code_change(_OldVsn, State, _Extra) ->\n     {ok, State}.\n \n \n-headers_strip_creds([], Acc) ->\n-    lists:reverse(Acc);\n-headers_strip_creds([{Key, Value0} | Rest], Acc) ->\n-    Value = case string:to_lower(Key) of\n-    \"authorization\" ->\n-        \"****\";\n-    _ ->\n-        Value0\n-    end,\n-    headers_strip_creds(Rest, [{Key, Value} | Acc]).\n-\n-\n-httpdb_strip_creds(#httpdb{url = Url, headers = Headers} = HttpDb) ->\n-    HttpDb#httpdb{\n-        url = couch_util:url_strip_password(Url),\n-        headers = headers_strip_creds(Headers, [])\n-    };\n-httpdb_strip_creds(LocalDb) ->\n-    LocalDb.\n-\n-\n-rep_strip_creds(#rep{source = Source, target = Target} = Rep) ->\n-    Rep#rep{\n-        source = httpdb_strip_creds(Source),\n-        target = httpdb_strip_creds(Target)\n-    }.\n \n+%%%%%% Multidb changes callbacks\n \n-state_strip_creds(#rep_state{rep_details = Rep, source = Source, target = Target} = State) ->\n-    % #rep_state contains the source and target at the top level and also\n-    % in the nested #rep_details record\n-    State#rep_state{\n-        rep_details = rep_strip_creds(Rep),\n-        source = httpdb_strip_creds(Source),\n-        target = httpdb_strip_creds(Target)\n-    }.\n+db_created(DbName, Server) ->\n+    couch_replicator_docs:ensure_rep_ddoc_exists(DbName),\n+    Server.\n \n+db_deleted(DbName, Server) ->\n+    clean_up_replications(DbName),\n+    Server.\n \n-terminate(normal, #rep_state{rep_details = #rep{id = RepId} = Rep,\n-    checkpoint_history = CheckpointHistory} = State) ->\n-    terminate_cleanup(State),\n-    couch_replicator_notifier:notify({finished, RepId, CheckpointHistory}),\n-    couch_replicator_manager:replication_completed(Rep, rep_stats(State));\n+db_found(DbName, Server) ->\n+    couch_replicator_docs:ensure_rep_ddoc_exists(DbName),\n+    Server.\n \n-terminate(shutdown, #rep_state{rep_details = #rep{id = RepId}} = State) ->\n-    % cancelled replication throught ?MODULE:cancel_replication/1\n-    couch_replicator_notifier:notify({error, RepId, <<\"cancelled\">>}),\n-    terminate_cleanup(State);\n-\n-terminate(shutdown, {error, Class, Error, Stack, InitArgs}) ->\n-    #rep{id=RepId} = InitArgs,\n-    couch_stats:increment_counter([couch_replicator, failed_starts]),\n-    CleanInitArgs = rep_strip_creds(InitArgs),\n-    couch_log:error(\"~p:~p: Replication failed to start for args ~p: ~p\",\n-             [Class, Error, CleanInitArgs, Stack]),\n-    case Error of\n-    {unauthorized, DbUri} ->\n-        NotifyError = {unauthorized, <<\"unauthorized to access or create database \", DbUri/binary>>};\n-    {db_not_found, DbUri} ->\n-        NotifyError = {db_not_found, <<\"could not open \", DbUri/binary>>};\n-    _ ->\n-        NotifyError = Error\n+db_change(DbName, {ChangeProps} = Change, Server) ->\n+    try\n+        ok = process_update(DbName, Change)\n+    catch\n+    _Tag:Error ->\n+        {RepProps} = get_json_value(doc, ChangeProps),\n+        DocId = get_json_value(<<\"_id\">>, RepProps),\n+        couch_replicator_docs:update_doc_process_error(DbName, DocId, Error)\n     end,\n-    couch_replicator_notifier:notify({error, RepId, NotifyError}),\n-    couch_replicator_manager:replication_error(InitArgs, NotifyError);\n-terminate(Reason, State) ->\n-    #rep_state{\n-        source_name = Source,\n-        target_name = Target,\n-        rep_details = #rep{id = {BaseId, Ext} = RepId} = Rep\n-    } = State,\n-    couch_log:error(\"Replication `~s` (`~s` -> `~s`) failed: ~s\",\n-        [BaseId ++ Ext, Source, Target, to_binary(Reason)]),\n-    terminate_cleanup(State),\n-    couch_replicator_notifier:notify({error, RepId, Reason}),\n-    couch_replicator_manager:replication_error(Rep, Reason).\n-\n-terminate_cleanup(State) ->\n-    update_task(State),\n-    stop_db_compaction_notifier(State#rep_state.source_db_compaction_notifier),\n-    stop_db_compaction_notifier(State#rep_state.target_db_compaction_notifier),\n-    couch_replicator_api_wrap:db_close(State#rep_state.source),\n-    couch_replicator_api_wrap:db_close(State#rep_state.target).\n-\n-\n-format_status(_Opt, [_PDict, State]) ->\n-    [{data, [{\"State\", state_strip_creds(State)}]}].\n-\n-\n-do_last_checkpoint(#rep_state{seqs_in_progress = [],\n-    highest_seq_done = {_Ts, ?LOWEST_SEQ}} = State) ->\n-    {stop, normal, cancel_timer(State)};\n-do_last_checkpoint(#rep_state{seqs_in_progress = [],\n-    highest_seq_done = Seq} = State) ->\n-    case do_checkpoint(State#rep_state{current_through_seq = Seq}) of\n-    {ok, NewState} ->\n-        couch_stats:increment_counter([couch_replicator, checkpoints, success]),\n-        {stop, normal, cancel_timer(NewState)};\n-    Error ->\n-        couch_stats:increment_counter([couch_replicator, checkpoints, failure]),\n-        {stop, Error, State}\n-    end.\n-\n-\n-start_timer(State) ->\n-    After = State#rep_state.checkpoint_interval,\n-    case timer:apply_after(After, gen_server, cast, [self(), checkpoint]) of\n-    {ok, Ref} ->\n-        Ref;\n-    Error ->\n-        couch_log:error(\"Replicator, error scheduling checkpoint:  ~p\", [Error]),\n-        nil\n-    end.\n-\n-\n-cancel_timer(#rep_state{timer = nil} = State) ->\n-    State;\n-cancel_timer(#rep_state{timer = Timer} = State) ->\n-    {ok, cancel} = timer:cancel(Timer),\n-    State#rep_state{timer = nil}.\n-\n-\n-init_state(Rep) ->\n-    #rep{\n-        id = {BaseId, _Ext},\n-        source = Src0, target = Tgt,\n-        options = Options, user_ctx = UserCtx,\n-        type = Type, view = View\n-    } = Rep,\n-    % Adjust minimum number of http source connections to 2 to avoid deadlock\n-    Src = adjust_maxconn(Src0, BaseId),\n-    {ok, Source} = couch_replicator_api_wrap:db_open(Src, [{user_ctx, UserCtx}]),\n-    {ok, Target} = couch_replicator_api_wrap:db_open(Tgt, [{user_ctx, UserCtx}],\n-        get_value(create_target, Options, false)),\n-\n-    {ok, SourceInfo} = couch_replicator_api_wrap:get_db_info(Source),\n-    {ok, TargetInfo} = couch_replicator_api_wrap:get_db_info(Target),\n+    Server.\n \n-    [SourceLog, TargetLog] = find_replication_logs([Source, Target], Rep),\n \n-    {StartSeq0, History} = compare_replication_logs(SourceLog, TargetLog),\n-    StartSeq1 = get_value(since_seq, Options, StartSeq0),\n-    StartSeq = {0, StartSeq1},\n+%%%%%%% Scheduler job callbacks\n \n-    SourceSeq = get_value(<<\"update_seq\">>, SourceInfo, ?LOWEST_SEQ),\n+-spec replication_started(#rep{}) -> ok.\n+replication_started(#rep{db_name = null}) ->\n+    ok;\n+replication_started(#rep{id = RepId, db_name = DbName, doc_id = DocId}) ->\n+    couch_replicator_docs:update_doc_triggered(DbName, DocId, RepId),\n+    couch_log:notice(\"Document `~s` triggered replication `~s`\",\n+        [DocId, pp_rep_id(RepId)]),\n+    ok.\n \n-    #doc{body={CheckpointHistory}} = SourceLog,\n-    State = #rep_state{\n-        rep_details = Rep,\n-        source_name = couch_replicator_api_wrap:db_uri(Source),\n-        target_name = couch_replicator_api_wrap:db_uri(Target),\n-        source = Source,\n-        target = Target,\n-        history = History,\n-        checkpoint_history = {[{<<\"no_changes\">>, true}| CheckpointHistory]},\n-        start_seq = StartSeq,\n-        current_through_seq = StartSeq,\n-        committed_seq = StartSeq,\n-        source_log = SourceLog,\n-        target_log = TargetLog,\n-        rep_starttime = httpd_util:rfc1123_date(),\n-        src_starttime = get_value(<<\"instance_start_time\">>, SourceInfo),\n-        tgt_starttime = get_value(<<\"instance_start_time\">>, TargetInfo),\n-        session_id = couch_uuids:random(),\n-        source_db_compaction_notifier =\n-            start_db_compaction_notifier(Source, self()),\n-        target_db_compaction_notifier =\n-            start_db_compaction_notifier(Target, self()),\n-        source_monitor = db_monitor(Source),\n-        target_monitor = db_monitor(Target),\n-        source_seq = SourceSeq,\n-        use_checkpoints = get_value(use_checkpoints, Options, true),\n-        checkpoint_interval = get_value(checkpoint_interval, Options,\n-                                        ?DEFAULT_CHECKPOINT_INTERVAL),\n-        type = Type,\n-        view = View\n-    },\n-    State#rep_state{timer = start_timer(State)}.\n \n+-spec replication_completed(#rep{}, list()) -> ok.\n+replication_completed(#rep{db_name = null}, _Stats) ->\n+    ok;\n+replication_completed(#rep{id = RepId, doc_id = DocId, db_name = DbName}, Stats) ->\n+    couch_replicator_docs:update_doc_completed(DbName, DocId, Stats),\n+    couch_log:notice(\"Replication `~s` completed (triggered by `~s`)\",\n+        [pp_rep_id(RepId), DocId]),\n+    ok.\n \n-find_replication_logs(DbList, #rep{id = {BaseId, _}} = Rep) ->\n-    LogId = ?l2b(?LOCAL_DOC_PREFIX ++ BaseId),\n-    fold_replication_logs(DbList, ?REP_ID_VERSION, LogId, LogId, Rep, []).\n \n \n-fold_replication_logs([], _Vsn, _LogId, _NewId, _Rep, Acc) ->\n-    lists:reverse(Acc);\n+%%%%%%%%% Private helper functions\n \n-fold_replication_logs([Db | Rest] = Dbs, Vsn, LogId, NewId, Rep, Acc) ->\n-    case couch_replicator_api_wrap:open_doc(Db, LogId, [ejson_body]) of\n-    {error, <<\"not_found\">>} when Vsn > 1 ->\n-        OldRepId = couch_replicator_utils:replication_id(Rep, Vsn - 1),\n-        fold_replication_logs(Dbs, Vsn - 1,\n-            ?l2b(?LOCAL_DOC_PREFIX ++ OldRepId), NewId, Rep, Acc);\n-    {error, <<\"not_found\">>} ->\n-        fold_replication_logs(\n-            Rest, ?REP_ID_VERSION, NewId, NewId, Rep, [#doc{id = NewId} | Acc]);\n-    {ok, Doc} when LogId =:= NewId ->\n-        fold_replication_logs(\n-            Rest, ?REP_ID_VERSION, NewId, NewId, Rep, [Doc | Acc]);\n-    {ok, Doc} ->\n-        MigratedLog = #doc{id = NewId, body = Doc#doc.body},\n-        fold_replication_logs(\n-            Rest, ?REP_ID_VERSION, NewId, NewId, Rep, [MigratedLog | Acc])\n+-spec do_replication_loop(#rep{}) ->\n+    {ok, {continuous, binary()}} | {ok, tuple()} | {error, any()}.\n+do_replication_loop(#rep{id = {BaseId, Ext} = Id, options = Options} = Rep) ->\n+    case couch_replicator_scheduler:add_job(Rep) of\n+    ok ->\n+        ok;\n+    {error, already_added} ->\n+        couch_log:notice(\"Replication '~s' already running\", [BaseId ++ Ext]),\n+        ok\n+    end,\n+    case get_value(continuous, Options, false) of\n+    true ->\n+        {ok, {continuous, ?l2b(BaseId ++ Ext)}};\n+    false ->\n+        wait_for_result(Id)\n     end.\n \n+-spec rep_result_listener(rep_id()) -> {ok, pid()}.\n+rep_result_listener(RepId) ->\n+    ReplyTo = self(),\n+    {ok, _Listener} = couch_replicator_notifier:start_link(\n+        fun({_, RepId2, _} = Ev) when RepId2 =:= RepId ->\n+                ReplyTo ! Ev;\n+            (_) ->\n+                ok\n+        end).\n \n-spawn_changes_manager(Parent, ChangesQueue, BatchSize) ->\n-    spawn_link(fun() ->\n-        changes_manager_loop_open(Parent, ChangesQueue, BatchSize, 1)\n-    end).\n-\n-changes_manager_loop_open(Parent, ChangesQueue, BatchSize, Ts) ->\n+-spec wait_for_result(rep_id()) ->\n+    {ok, any()} | {error, any()}.\n+wait_for_result(RepId) ->\n     receive\n-    {get_changes, From} ->\n-        case couch_work_queue:dequeue(ChangesQueue, BatchSize) of\n-        closed ->\n-            From ! {closed, self()};\n-        {ok, Changes} ->\n-            #doc_info{high_seq = Seq} = lists:last(Changes),\n-            ReportSeq = {Ts, Seq},\n-            ok = gen_server:cast(Parent, {report_seq, ReportSeq}),\n-            From ! {changes, self(), Changes, ReportSeq}\n-        end,\n-        changes_manager_loop_open(Parent, ChangesQueue, BatchSize, Ts + 1)\n-    end.\n-\n-\n-do_checkpoint(#rep_state{use_checkpoints=false} = State) ->\n-    NewState = State#rep_state{checkpoint_history = {[{<<\"use_checkpoints\">>, false}]} },\n-    {ok, NewState};\n-do_checkpoint(#rep_state{current_through_seq=Seq, committed_seq=Seq} = State) ->\n-    update_task(State),\n-    {ok, State};\n-do_checkpoint(State) ->\n-    #rep_state{\n-        source_name=SourceName,\n-        target_name=TargetName,\n-        source = Source,\n-        target = Target,\n-        history = OldHistory,\n-        start_seq = {_, StartSeq},\n-        current_through_seq = {_Ts, NewSeq} = NewTsSeq,\n-        source_log = SourceLog,\n-        target_log = TargetLog,\n-        rep_starttime = ReplicationStartTime,\n-        src_starttime = SrcInstanceStartTime,\n-        tgt_starttime = TgtInstanceStartTime,\n-        stats = Stats,\n-        rep_details = #rep{options = Options},\n-        session_id = SessionId\n-    } = State,\n-    case commit_to_both(Source, Target) of\n-    {source_error, Reason} ->\n-         {checkpoint_commit_failure,\n-             <<\"Failure on source commit: \", (to_binary(Reason))/binary>>};\n-    {target_error, Reason} ->\n-         {checkpoint_commit_failure,\n-             <<\"Failure on target commit: \", (to_binary(Reason))/binary>>};\n-    {SrcInstanceStartTime, TgtInstanceStartTime} ->\n-        couch_log:notice(\"recording a checkpoint for `~s` -> `~s` at source update_seq ~p\",\n-            [SourceName, TargetName, NewSeq]),\n-        StartTime = ?l2b(ReplicationStartTime),\n-        EndTime = ?l2b(httpd_util:rfc1123_date()),\n-        NewHistoryEntry = {[\n-            {<<\"session_id\">>, SessionId},\n-            {<<\"start_time\">>, StartTime},\n-            {<<\"end_time\">>, EndTime},\n-            {<<\"start_last_seq\">>, StartSeq},\n-            {<<\"end_last_seq\">>, NewSeq},\n-            {<<\"recorded_seq\">>, NewSeq},\n-            {<<\"missing_checked\">>, couch_replicator_stats:missing_checked(Stats)},\n-            {<<\"missing_found\">>, couch_replicator_stats:missing_found(Stats)},\n-            {<<\"docs_read\">>, couch_replicator_stats:docs_read(Stats)},\n-            {<<\"docs_written\">>, couch_replicator_stats:docs_written(Stats)},\n-            {<<\"doc_write_failures\">>, couch_replicator_stats:doc_write_failures(Stats)}\n-        ]},\n-        BaseHistory = [\n-            {<<\"session_id\">>, SessionId},\n-            {<<\"source_last_seq\">>, NewSeq},\n-            {<<\"replication_id_version\">>, ?REP_ID_VERSION}\n-        ] ++ case get_value(doc_ids, Options) of\n-        undefined ->\n-            [];\n-        _DocIds ->\n-            % backwards compatibility with the result of a replication by\n-            % doc IDs in versions 0.11.x and 1.0.x\n-            % TODO: deprecate (use same history format, simplify code)\n-            [\n-                {<<\"start_time\">>, StartTime},\n-                {<<\"end_time\">>, EndTime},\n-                {<<\"docs_read\">>, couch_replicator_stats:docs_read(Stats)},\n-                {<<\"docs_written\">>, couch_replicator_stats:docs_written(Stats)},\n-                {<<\"doc_write_failures\">>, couch_replicator_stats:doc_write_failures(Stats)}\n-            ]\n-        end,\n-        % limit history to 50 entries\n-        NewRepHistory = {\n-            BaseHistory ++\n-            [{<<\"history\">>, lists:sublist([NewHistoryEntry | OldHistory], 50)}]\n-        },\n-\n-        try\n-            {SrcRevPos, SrcRevId} = update_checkpoint(\n-                Source, SourceLog#doc{body = NewRepHistory}, source),\n-            {TgtRevPos, TgtRevId} = update_checkpoint(\n-                Target, TargetLog#doc{body = NewRepHistory}, target),\n-            NewState = State#rep_state{\n-                checkpoint_history = NewRepHistory,\n-                committed_seq = NewTsSeq,\n-                source_log = SourceLog#doc{revs={SrcRevPos, [SrcRevId]}},\n-                target_log = TargetLog#doc{revs={TgtRevPos, [TgtRevId]}}\n-            },\n-            update_task(NewState),\n-            {ok, NewState}\n-        catch throw:{checkpoint_commit_failure, _} = Failure ->\n-            Failure\n-        end;\n-    {SrcInstanceStartTime, _NewTgtInstanceStartTime} ->\n-        {checkpoint_commit_failure, <<\"Target database out of sync. \"\n-            \"Try to increase max_dbs_open at the target's server.\">>};\n-    {_NewSrcInstanceStartTime, TgtInstanceStartTime} ->\n-        {checkpoint_commit_failure, <<\"Source database out of sync. \"\n-            \"Try to increase max_dbs_open at the source's server.\">>};\n-    {_NewSrcInstanceStartTime, _NewTgtInstanceStartTime} ->\n-        {checkpoint_commit_failure, <<\"Source and target databases out of \"\n-            \"sync. Try to increase max_dbs_open at both servers.\">>}\n-    end.\n-\n-\n-update_checkpoint(Db, Doc, DbType) ->\n-    try\n-        update_checkpoint(Db, Doc)\n-    catch throw:{checkpoint_commit_failure, Reason} ->\n-        throw({checkpoint_commit_failure,\n-            <<\"Error updating the \", (to_binary(DbType))/binary,\n-                \" checkpoint document: \", (to_binary(Reason))/binary>>})\n-    end.\n-\n-update_checkpoint(Db, #doc{id = LogId, body = LogBody} = Doc) ->\n-    try\n-        case couch_replicator_api_wrap:update_doc(Db, Doc, [delay_commit]) of\n-        {ok, PosRevId} ->\n-            PosRevId;\n-        {error, Reason} ->\n-            throw({checkpoint_commit_failure, Reason})\n-        end\n-    catch throw:conflict ->\n-        case (catch couch_replicator_api_wrap:open_doc(Db, LogId, [ejson_body])) of\n-        {ok, #doc{body = LogBody, revs = {Pos, [RevId | _]}}} ->\n-            % This means that we were able to update successfully the\n-            % checkpoint doc in a previous attempt but we got a connection\n-            % error (timeout for e.g.) before receiving the success response.\n-            % Therefore the request was retried and we got a conflict, as the\n-            % revision we sent is not the current one.\n-            % We confirm this by verifying the doc body we just got is the same\n-            % that we have just sent.\n-            {Pos, RevId};\n-        _ ->\n-            throw({checkpoint_commit_failure, conflict})\n-        end\n-    end.\n-\n-\n-commit_to_both(Source, Target) ->\n-    % commit the src async\n-    ParentPid = self(),\n-    SrcCommitPid = spawn_link(\n-        fun() ->\n-            Result = (catch couch_replicator_api_wrap:ensure_full_commit(Source)),\n-            ParentPid ! {self(), Result}\n-        end),\n-\n-    % commit tgt sync\n-    TargetResult = (catch couch_replicator_api_wrap:ensure_full_commit(Target)),\n-\n-    SourceResult = receive\n-    {SrcCommitPid, Result} ->\n-        unlink(SrcCommitPid),\n-        receive {'EXIT', SrcCommitPid, _} -> ok after 0 -> ok end,\n-        Result;\n-    {'EXIT', SrcCommitPid, Reason} ->\n+    {finished, RepId, RepResult} ->\n+        {ok, RepResult};\n+    {error, RepId, Reason} ->\n         {error, Reason}\n-    end,\n-    case TargetResult of\n-    {ok, TargetStartTime} ->\n-        case SourceResult of\n-        {ok, SourceStartTime} ->\n-            {SourceStartTime, TargetStartTime};\n-        SourceError ->\n-            {source_error, SourceError}\n-        end;\n-    TargetError ->\n-        {target_error, TargetError}\n     end.\n \n \n-compare_replication_logs(SrcDoc, TgtDoc) ->\n-    #doc{body={RepRecProps}} = SrcDoc,\n-    #doc{body={RepRecPropsTgt}} = TgtDoc,\n-    case get_value(<<\"session_id\">>, RepRecProps) ==\n-            get_value(<<\"session_id\">>, RepRecPropsTgt) of\n-    true ->\n-        % if the records have the same session id,\n-        % then we have a valid replication history\n-        OldSeqNum = get_value(<<\"source_last_seq\">>, RepRecProps, ?LOWEST_SEQ),\n-        OldHistory = get_value(<<\"history\">>, RepRecProps, []),\n-        {OldSeqNum, OldHistory};\n-    false ->\n-        SourceHistory = get_value(<<\"history\">>, RepRecProps, []),\n-        TargetHistory = get_value(<<\"history\">>, RepRecPropsTgt, []),\n-        couch_log:notice(\"Replication records differ. \"\n-                \"Scanning histories to find a common ancestor.\", []),\n-        couch_log:debug(\"Record on source:~p~nRecord on target:~p~n\",\n-                [RepRecProps, RepRecPropsTgt]),\n-        compare_rep_history(SourceHistory, TargetHistory)\n+-spec cancel_replication(rep_id()) ->\n+    {ok, {cancelled, binary()}} | {error, not_found}.\n+cancel_replication({BasedId, Extension} = RepId) ->\n+    FullRepId = BasedId ++ Extension,\n+    couch_log:notice(\"Canceling replication '~s' ...\", [FullRepId]),\n+    case rep_state(RepId) of\n+    #rep{} ->\n+        ok = couch_replicator_scheduler:remove_job(RepId),\n+        couch_log:notice(\"Replication '~s' cancelled\", [FullRepId]),\n+        {ok, {cancelled, ?l2b(FullRepId)}};\n+    nil ->\n+        couch_log:notice(\"Replication '~s' not found\", [FullRepId]),\n+        {error, not_found}\n     end.\n \n-compare_rep_history(S, T) when S =:= [] orelse T =:= [] ->\n-    couch_log:notice(\"no common ancestry -- performing full replication\", []),\n-    {?LOWEST_SEQ, []};\n-compare_rep_history([{S} | SourceRest], [{T} | TargetRest] = Target) ->\n-    SourceId = get_value(<<\"session_id\">>, S),\n-    case has_session_id(SourceId, Target) of\n+\n+-spec cancel_replication(rep_id(), #user_ctx{}) ->\n+    {ok, {cancelled, binary()}} | {error, not_found}.\n+cancel_replication(RepId, #user_ctx{name = Name, roles = Roles}) ->\n+    case lists:member(<<\"_admin\">>, Roles) of\n     true ->\n-        RecordSeqNum = get_value(<<\"recorded_seq\">>, S, ?LOWEST_SEQ),\n-        couch_log:notice(\"found a common replication record with source_seq ~p\",\n-            [RecordSeqNum]),\n-        {RecordSeqNum, SourceRest};\n+        cancel_replication(RepId);\n     false ->\n-        TargetId = get_value(<<\"session_id\">>, T),\n-        case has_session_id(TargetId, SourceRest) of\n-        true ->\n-            RecordSeqNum = get_value(<<\"recorded_seq\">>, T, ?LOWEST_SEQ),\n-            couch_log:notice(\"found a common replication record with source_seq ~p\",\n-                [RecordSeqNum]),\n-            {RecordSeqNum, TargetRest};\n-        false ->\n-            compare_rep_history(SourceRest, TargetRest)\n+        case rep_state(RepId) of\n+        #rep{user_ctx = #user_ctx{name = Name}} ->\n+            cancel_replication(RepId);\n+        #rep{user_ctx = #user_ctx{name = _Other}} ->\n+            throw({unauthorized,\n+                <<\"Can't cancel a replication triggered by another user\">>});\n+        nil ->\n+            {error, not_found}\n         end\n-    end.\n-\n-\n-has_session_id(_SessionId, []) ->\n-    false;\n-has_session_id(SessionId, [{Props} | Rest]) ->\n-    case get_value(<<\"session_id\">>, Props, nil) of\n-    SessionId ->\n-        true;\n-    _Else ->\n-        has_session_id(SessionId, Rest)\n-    end.\n-\n-\n-db_monitor(#db{} = Db) ->\n-    couch_db:monitor(Db);\n-db_monitor(_HttpDb) ->\n-    nil.\n-\n-get_pending_count(St) ->\n-    Rep = St#rep_state.rep_details,\n-    Timeout = get_value(connection_timeout, Rep#rep.options),\n-    TimeoutMicro = Timeout * 1000,\n-    case get(pending_count_state) of\n-        {LastUpdate, PendingCount} ->\n-            case timer:now_diff(os:timestamp(), LastUpdate) > TimeoutMicro of\n-                true ->\n-                    NewPendingCount = get_pending_count_int(St),\n-                    put(pending_count_state, {os:timestamp(), NewPendingCount}),\n-                    NewPendingCount;\n-                false ->\n-                    PendingCount\n-            end;\n+     end.\n+\n+\n+\n+-spec process_update(binary(), {[_]}) -> ok.\n+process_update(DbName, {Change}) ->\n+    {RepProps} = JsonRepDoc = get_json_value(doc, Change),\n+    DocId = get_json_value(<<\"_id\">>, RepProps),\n+    Owner = couch_replicator_clustering:owner(DbName, DocId),\n+    case {Owner, get_json_value(deleted, Change, false)} of\n+    {_, true} ->\n+        remove_jobs(DbName, DocId);\n+    {unstable, false} ->\n+\tcouch_log:notice(\"Not starting '~s' as cluster is unstable\", [DocId]);\n+    {ThisNode, false} when ThisNode =:= node() ->\n+        couch_log:notice(\"Maybe starting '~s' as I'm the owner\", [DocId]),\n+        case get_json_value(<<\"_replication_state\">>, RepProps) of\n         undefined ->\n-            NewPendingCount = get_pending_count_int(St),\n-            put(pending_count_state, {os:timestamp(), NewPendingCount}),\n-            NewPendingCount\n-    end.\n+            maybe_start_replication(DbName, DocId, JsonRepDoc);\n+        <<\"triggered\">> ->\n+            maybe_start_replication(DbName, DocId, JsonRepDoc);\n+        <<\"completed\">> ->\n+            couch_log:notice(\"Replication '~s' marked as completed\", [DocId])\n+        end;\n+     {Owner, false} ->\n+         couch_log:notice(\"Not starting '~s' as owner is ~s.\", [DocId, Owner])\n+    end,\n+    ok.\n+\n+\n+-spec maybe_start_replication(binary(), binary(), {[_]}) -> ok.\n+maybe_start_replication(DbName, DocId, RepDoc) ->\n+    Rep0 = couch_replicator_docs:parse_rep_doc(RepDoc),\n+    #rep{id = {BaseId, _} = RepId} = Rep0,\n+    Rep = Rep0#rep{db_name = DbName},\n+    case rep_state(RepId) of\n+    nil ->\n+        couch_log:notice(\"Attempting to start replication `~s` (document `~s`).\",\n+            [pp_rep_id(RepId), DocId]),\n+        case couch_replicator_scheduler:add_job(Rep) of\n+        ok ->\n+            ok;\n+        {error, already_added} ->\n+            couch_log:warning(\"replicator scheduler: ~p was already added\", [Rep])\n+        end,\n+        ok;\n+    #rep{doc_id = DocId} ->\n+        ok;\n+    #rep{db_name = DbName, doc_id = OtherDocId} ->\n+        couch_log:notice(\"The replication specified by the document `~s` already started\"\n+            \" triggered by the document `~s`\", [DocId, OtherDocId]),\n+        maybe_tag_rep_doc(DbName, DocId, RepDoc, ?l2b(BaseId))\n+    end,\n+    ok.\n \n \n-get_pending_count_int(#rep_state{source = #httpdb{} = Db0}=St) ->\n-    {_, Seq} = St#rep_state.highest_seq_done,\n-    Db = Db0#httpdb{retries = 3},\n-    case (catch couch_replicator_api_wrap:get_pending_count(Db, Seq)) of\n-    {ok, Pending} ->\n-        Pending;\n+-spec maybe_tag_rep_doc(binary(), binary(), {[_]}, binary()) -> ok.\n+maybe_tag_rep_doc(DbName, DocId, {RepProps}, RepId) ->\n+    case get_json_value(<<\"_replication_id\">>, RepProps) of\n+    RepId ->\n+        ok;\n     _ ->\n-        null\n-    end;\n-get_pending_count_int(#rep_state{source = Db}=St) ->\n-    {_, Seq} = St#rep_state.highest_seq_done,\n-    {ok, Pending} = couch_replicator_api_wrap:get_pending_count(Db, Seq),\n-    Pending.\n-\n-\n-update_task(State) ->\n-    #rep_state{\n-        current_through_seq = {_, ThroughSeq},\n-        highest_seq_done = {_, HighestSeq}\n-    } = State,\n-    couch_task_status:update(\n-        rep_stats(State) ++ [\n-        {source_seq, HighestSeq},\n-        {through_seq, ThroughSeq}\n-    ]).\n+        couch_replicator_docs:update_doc_replication_id(DbName, DocId, RepId)\n+    end.\n \n \n-rep_stats(State) ->\n-    #rep_state{\n-        committed_seq = {_, CommittedSeq},\n-        stats = Stats\n-    } = State,\n+-spec remove_jobs(binary(), binary()) -> ok.\n+remove_jobs(DbName, DocId) ->\n+    LogMsg = \"Stopped replication `~s` , replication document `~s`\",\n     [\n-        {revisions_checked, couch_replicator_stats:missing_checked(Stats)},\n-        {missing_revisions_found, couch_replicator_stats:missing_found(Stats)},\n-        {docs_read, couch_replicator_stats:docs_read(Stats)},\n-        {docs_written, couch_replicator_stats:docs_written(Stats)},\n-        {changes_pending, get_pending_count(State)},\n-        {doc_write_failures, couch_replicator_stats:doc_write_failures(Stats)},\n-        {checkpointed_source_seq, CommittedSeq}\n-    ].\n+        begin\n+            couch_replicator_scheduler:remove_job(RepId),\n+            couch_log:notice(LogMsg, [pp_rep_id(RepId), DocId])\n+        end || RepId <- find_jobs_by_doc(DbName, DocId)\n+    ],\n+    ok.\n+\n+\n+% pretty-print replication id\n+-spec pp_rep_id(#rep{} | rep_id()) -> string().\n+pp_rep_id(#rep{id = RepId}) ->\n+    pp_rep_id(RepId);\n+pp_rep_id({Base, Extension}) ->\n+    Base ++ Extension.\n+\n+\n+\n+% TODO: make this a function in couch_replicator_scheduler API\n+-spec clean_up_replications(binary()) -> ok.\n+clean_up_replications(DbName) ->\n+    RepIds = find_jobs_by_dbname(DbName),\n+    [couch_replicator_scheduler:remove_job(RepId) || RepId <- RepIds],\n+    ok.","path":"src/couch_replicator.erl","position":1232,"original_position":1232,"commit_id":"f54a2ea6f54c2ce646b6255e92ae2df3f46c3102","original_commit_id":"f54a2ea6f54c2ce646b6255e92ae2df3f46c3102","user":{"login":"sagelywizard","id":167249,"avatar_url":"https://avatars.githubusercontent.com/u/167249?v=3","gravatar_id":"","url":"https://api.github.com/users/sagelywizard","html_url":"https://github.com/sagelywizard","followers_url":"https://api.github.com/users/sagelywizard/followers","following_url":"https://api.github.com/users/sagelywizard/following{/other_user}","gists_url":"https://api.github.com/users/sagelywizard/gists{/gist_id}","starred_url":"https://api.github.com/users/sagelywizard/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sagelywizard/subscriptions","organizations_url":"https://api.github.com/users/sagelywizard/orgs","repos_url":"https://api.github.com/users/sagelywizard/repos","events_url":"https://api.github.com/users/sagelywizard/events{/privacy}","received_events_url":"https://api.github.com/users/sagelywizard/received_events","type":"User","site_admin":false},"body":"Another nit: You can replace the above two lines with `lists:foreach(fun couch_replicator_scheduler:remove_job/1, RepIds).`","created_at":"2016-05-23T18:05:11Z","updated_at":"2016-05-23T18:05:11Z","html_url":"https://github.com/cloudant/couchdb-couch-replicator/pull/7#discussion_r64261042","pull_request_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls/7","_links":{"self":{"href":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls/comments/64261042"},"html":{"href":"https://github.com/cloudant/couchdb-couch-replicator/pull/7#discussion_r64261042"},"pull_request":{"href":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls/7"}}},"pull_request":{"url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls/7","id":71010043,"html_url":"https://github.com/cloudant/couchdb-couch-replicator/pull/7","diff_url":"https://github.com/cloudant/couchdb-couch-replicator/pull/7.diff","patch_url":"https://github.com/cloudant/couchdb-couch-replicator/pull/7.patch","issue_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/issues/7","number":7,"state":"open","locked":false,"title":"Remove `couch_replicator_manager`","user":{"login":"nickva","id":211822,"avatar_url":"https://avatars.githubusercontent.com/u/211822?v=3","gravatar_id":"","url":"https://api.github.com/users/nickva","html_url":"https://github.com/nickva","followers_url":"https://api.github.com/users/nickva/followers","following_url":"https://api.github.com/users/nickva/following{/other_user}","gists_url":"https://api.github.com/users/nickva/gists{/gist_id}","starred_url":"https://api.github.com/users/nickva/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/nickva/subscriptions","organizations_url":"https://api.github.com/users/nickva/orgs","repos_url":"https://api.github.com/users/nickva/repos","events_url":"https://api.github.com/users/nickva/events{/privacy}","received_events_url":"https://api.github.com/users/nickva/received_events","type":"User","site_admin":false},"body":"Its logic is now spread amongst a few modules:\r\n\r\n - `couch_multidb_changes`:\r\n    Monitor `*/_replicator` dbs, run callbacks for db / doc changes.\r\n\r\n - `couch_replicator_clustering`:\r\n    Keep track of cluster membership. Trigger rescans, by restarting\r\n    `couch_multidb_changes`. Resolve replication ownership.\r\n\r\n - `couch_replicator`:\r\n    Create local `_replicator` db on start-up. Add / remove replication jobs.\r\n    Call `couch_replicator_docs` to write info back to replicator docs. Handle\r\n    `_replicate` endpoint replications but use job scheduler instead.\r\n\r\nAlso:\r\n\r\n* Dialyze more code, especially in couch_replicatorl and couch_replicator_docs.\r\n\r\n* Keep `couch_replicator_manager` module around for a while in order to\r\nproxy calls made from other applications. Otherwise would have to clone other\r\nrepos. Remove it and update code accordingly on final cleanup.\r\n\r\n* Update `owner` function from couch_replicator_clustering to have result which\r\nmakes more sense for replication jobs.\r\n\r\n* Avoided touching scheduler as there are prs pending for it. Created a few\r\njob manipulation functions in couch_replicator.erl. Moving those to scheduler\r\nAPI eventually, as they access internal job scheduler data in a hacky way.\r\n\r\n* A major change to behavior is errors are not written back to docs anymore\r\n(except errors in parsing the doc). Triggered updates are still written because\r\nthey also tag the doc with the replication_id.\r\n\r\nUpdated eunit tests, all pass (still):\r\n\r\n```\r\n$ make eunit apps=couch_replicator\r\n...\r\n [done in 51.177 s]\r\n=======================================================\r\n  All 171 tests passed.\r\n```\r\n\r\nReplications running from dev cluster with `rep` tool still trigger documents\r\nand replicate data.\r\n\r\n```\r\n>>> rep.replicate_1_to_n_then_check_replication(2)\r\ncreating rdyno_src_0001\r\n > created  1 dbs with prefix rdyno_src_\r\ncreating rdyno_tgt_0001\r\ncreating rdyno_tgt_0002\r\n > created  2 dbs with prefix rdyno_tgt_\r\nupdating documents\r\n > _replicator rdyno_0001_0001 : 1-34f09a6a62181f4d4631bf8544b582fc\r\n > _replicator rdyno_0001_0002 : 1-0f52a360f959fea67e6cb20aa1bb0606\r\nwaiting for replication documents to trigger\r\n > function wait_to_trigger succeded after 0.011 +/- 10.0  sec.\r\nall replication documents triggered\r\n>>> update cycle 0  <<<\r\n > waiting for target  1\r\n > retrying function wait_till_dbs_equal\r\n...\r\n > function wait_till_dbs_equal succeded after 49.574 +/- 1.0  sec.\r\n > waiting for target  2\r\n > function wait_till_dbs_equal succeded after 0.019 +/- 1.0  sec.\r\n > waiting to propagate changes from  1 to 2  : 49.623 sec.\r\n```","created_at":"2016-05-23T06:23:00Z","updated_at":"2016-05-23T18:05:11Z","closed_at":null,"merged_at":null,"merge_commit_sha":"61b5239206f73b926604903b73ba26f2d296c14d","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls/7/commits","review_comments_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls/7/comments","review_comment_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls/comments{/number}","comments_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/issues/7/comments","statuses_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/statuses/f54a2ea6f54c2ce646b6255e92ae2df3f46c3102","head":{"label":"cloudant:63012-removed-replicator-manager","ref":"63012-removed-replicator-manager","sha":"f54a2ea6f54c2ce646b6255e92ae2df3f46c3102","user":{"login":"cloudant","id":44642,"avatar_url":"https://avatars.githubusercontent.com/u/44642?v=3","gravatar_id":"","url":"https://api.github.com/users/cloudant","html_url":"https://github.com/cloudant","followers_url":"https://api.github.com/users/cloudant/followers","following_url":"https://api.github.com/users/cloudant/following{/other_user}","gists_url":"https://api.github.com/users/cloudant/gists{/gist_id}","starred_url":"https://api.github.com/users/cloudant/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cloudant/subscriptions","organizations_url":"https://api.github.com/users/cloudant/orgs","repos_url":"https://api.github.com/users/cloudant/repos","events_url":"https://api.github.com/users/cloudant/events{/privacy}","received_events_url":"https://api.github.com/users/cloudant/received_events","type":"Organization","site_admin":false},"repo":{"id":38246096,"name":"couchdb-couch-replicator","full_name":"cloudant/couchdb-couch-replicator","owner":{"login":"cloudant","id":44642,"avatar_url":"https://avatars.githubusercontent.com/u/44642?v=3","gravatar_id":"","url":"https://api.github.com/users/cloudant","html_url":"https://github.com/cloudant","followers_url":"https://api.github.com/users/cloudant/followers","following_url":"https://api.github.com/users/cloudant/following{/other_user}","gists_url":"https://api.github.com/users/cloudant/gists{/gist_id}","starred_url":"https://api.github.com/users/cloudant/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cloudant/subscriptions","organizations_url":"https://api.github.com/users/cloudant/orgs","repos_url":"https://api.github.com/users/cloudant/repos","events_url":"https://api.github.com/users/cloudant/events{/privacy}","received_events_url":"https://api.github.com/users/cloudant/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/cloudant/couchdb-couch-replicator","description":"Mirror of Apache CouchDB","fork":true,"url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator","forks_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/forks","keys_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/keys{/key_id}","collaborators_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/teams","hooks_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/hooks","issue_events_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/issues/events{/number}","events_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/events","assignees_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/assignees{/user}","branches_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/branches{/branch}","tags_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/tags","blobs_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/git/refs{/sha}","trees_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/git/trees{/sha}","statuses_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/statuses/{sha}","languages_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/languages","stargazers_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/stargazers","contributors_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/contributors","subscribers_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/subscribers","subscription_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/subscription","commits_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/commits{/sha}","git_commits_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/git/commits{/sha}","comments_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/comments{/number}","issue_comment_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/issues/comments{/number}","contents_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/contents/{+path}","compare_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/compare/{base}...{head}","merges_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/merges","archive_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/downloads","issues_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/issues{/number}","pulls_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls{/number}","milestones_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/milestones{/number}","notifications_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/labels{/name}","releases_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/releases{/id}","deployments_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/deployments","created_at":"2015-06-29T12:19:20Z","updated_at":"2016-05-13T14:06:45Z","pushed_at":"2016-05-23T06:23:01Z","git_url":"git://github.com/cloudant/couchdb-couch-replicator.git","ssh_url":"git@github.com:cloudant/couchdb-couch-replicator.git","clone_url":"https://github.com/cloudant/couchdb-couch-replicator.git","svn_url":"https://github.com/cloudant/couchdb-couch-replicator","homepage":null,"size":576,"stargazers_count":0,"watchers_count":0,"language":"Erlang","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":1,"forks":0,"open_issues":1,"watchers":0,"default_branch":"master"}},"base":{"label":"cloudant:63012-scheduler","ref":"63012-scheduler","sha":"e69aaab6b4a98afada3992908dddd611d61b4f0e","user":{"login":"cloudant","id":44642,"avatar_url":"https://avatars.githubusercontent.com/u/44642?v=3","gravatar_id":"","url":"https://api.github.com/users/cloudant","html_url":"https://github.com/cloudant","followers_url":"https://api.github.com/users/cloudant/followers","following_url":"https://api.github.com/users/cloudant/following{/other_user}","gists_url":"https://api.github.com/users/cloudant/gists{/gist_id}","starred_url":"https://api.github.com/users/cloudant/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cloudant/subscriptions","organizations_url":"https://api.github.com/users/cloudant/orgs","repos_url":"https://api.github.com/users/cloudant/repos","events_url":"https://api.github.com/users/cloudant/events{/privacy}","received_events_url":"https://api.github.com/users/cloudant/received_events","type":"Organization","site_admin":false},"repo":{"id":38246096,"name":"couchdb-couch-replicator","full_name":"cloudant/couchdb-couch-replicator","owner":{"login":"cloudant","id":44642,"avatar_url":"https://avatars.githubusercontent.com/u/44642?v=3","gravatar_id":"","url":"https://api.github.com/users/cloudant","html_url":"https://github.com/cloudant","followers_url":"https://api.github.com/users/cloudant/followers","following_url":"https://api.github.com/users/cloudant/following{/other_user}","gists_url":"https://api.github.com/users/cloudant/gists{/gist_id}","starred_url":"https://api.github.com/users/cloudant/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/cloudant/subscriptions","organizations_url":"https://api.github.com/users/cloudant/orgs","repos_url":"https://api.github.com/users/cloudant/repos","events_url":"https://api.github.com/users/cloudant/events{/privacy}","received_events_url":"https://api.github.com/users/cloudant/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/cloudant/couchdb-couch-replicator","description":"Mirror of Apache CouchDB","fork":true,"url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator","forks_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/forks","keys_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/keys{/key_id}","collaborators_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/teams","hooks_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/hooks","issue_events_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/issues/events{/number}","events_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/events","assignees_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/assignees{/user}","branches_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/branches{/branch}","tags_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/tags","blobs_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/git/refs{/sha}","trees_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/git/trees{/sha}","statuses_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/statuses/{sha}","languages_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/languages","stargazers_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/stargazers","contributors_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/contributors","subscribers_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/subscribers","subscription_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/subscription","commits_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/commits{/sha}","git_commits_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/git/commits{/sha}","comments_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/comments{/number}","issue_comment_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/issues/comments{/number}","contents_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/contents/{+path}","compare_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/compare/{base}...{head}","merges_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/merges","archive_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/downloads","issues_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/issues{/number}","pulls_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls{/number}","milestones_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/milestones{/number}","notifications_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/labels{/name}","releases_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/releases{/id}","deployments_url":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/deployments","created_at":"2015-06-29T12:19:20Z","updated_at":"2016-05-13T14:06:45Z","pushed_at":"2016-05-23T06:23:01Z","git_url":"git://github.com/cloudant/couchdb-couch-replicator.git","ssh_url":"git@github.com:cloudant/couchdb-couch-replicator.git","clone_url":"https://github.com/cloudant/couchdb-couch-replicator.git","svn_url":"https://github.com/cloudant/couchdb-couch-replicator","homepage":null,"size":576,"stargazers_count":0,"watchers_count":0,"language":"Erlang","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":1,"forks":0,"open_issues":1,"watchers":0,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls/7"},"html":{"href":"https://github.com/cloudant/couchdb-couch-replicator/pull/7"},"issue":{"href":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/issues/7"},"comments":{"href":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/issues/7/comments"},"review_comments":{"href":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls/7/comments"},"review_comment":{"href":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/pulls/7/commits"},"statuses":{"href":"https://api.github.com/repos/cloudant/couchdb-couch-replicator/statuses/f54a2ea6f54c2ce646b6255e92ae2df3f46c3102"}}}}
{"actor":{"display_login":"sagelywizard"}}