<commit_before>% 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_replicator).
-behaviour(gen_server).
-vsn(1).

% public API
-export([replicate/2]).

% meant to be used only by the replicator database listener
-export([async_replicate/1]).
-export([cancel_replication/1]).

% gen_server callbacks
-export([init/1, terminate/2, code_change/3]).
-export([handle_call/3, handle_cast/2, handle_info/2]).
-export([format_status/2]).

-export([details/1]).

-include_lib("couch/include/couch_db.hrl").
-include("couch_replicator_api_wrap.hrl").
-include("couch_replicator.hrl").

-define(LOWEST_SEQ, 0).

-import(couch_util, [
    get_value/2,
    get_value/3,
    to_binary/1
]).

-import(couch_replicator_utils, [
    start_db_compaction_notifier/2,
    stop_db_compaction_notifier/1
]).

-record(rep_state, {
    rep_details,
    source_name,
    target_name,
    source,
    target,
    history,
    checkpoint_history,
    start_seq,
    committed_seq,
    current_through_seq,
    seqs_in_progress = [],
    highest_seq_done = {0, ?LOWEST_SEQ},
    source_log,
    target_log,
    rep_starttime,
    src_starttime,
    tgt_starttime,
    timer, % checkpoint timer
    changes_queue,
    changes_manager,
    changes_reader,
    workers,
    stats = couch_replicator_stats:new(),
    session_id,
    source_db_compaction_notifier = nil,
    target_db_compaction_notifier = nil,
    source_monitor = nil,
    target_monitor = nil,
    source_seq = nil,
    use_checkpoints = true,
    checkpoint_interval = 5000,
    type = db,
    view = nil
}).


replicate(PostBody, Ctx) ->
    {ok, #rep{id = RepId, options = Options, user_ctx = UserCtx} = Rep} =
        couch_replicator_utils:parse_rep_doc(PostBody, Ctx),
    case get_value(cancel, Options, false) of
    true ->
        case get_value(id, Options, nil) of
        nil ->
            cancel_replication(RepId);
        RepId2 ->
            cancel_replication(RepId2, UserCtx)
        end;
    false ->
        {ok, Listener} = rep_result_listener(RepId),
        Result = do_replication_loop(Rep),
        couch_replicator_notifier:stop(Listener),
        Result
    end.


do_replication_loop(#rep{id = {BaseId, Ext} = Id, options = Options} = Rep) ->
    case async_replicate(Rep) of
    {ok, _Pid} ->
        case get_value(continuous, Options, false) of
        true ->
            {ok, {continuous, ?l2b(BaseId ++ Ext)}};
        false ->
            wait_for_result(Id)
        end;
    Error ->
        Error
    end.


async_replicate(#rep{id = {BaseId, Ext}, source = Src, target = Tgt} = Rep) ->
    RepChildId = BaseId ++ Ext,
    Source = couch_replicator_api_wrap:db_uri(Src),
    Target = couch_replicator_api_wrap:db_uri(Tgt),
    Timeout = get_value(connection_timeout, Rep#rep.options),
    ChildSpec = {
        RepChildId,
        {gen_server, start_link, [?MODULE, Rep, [{timeout, Timeout}]]},
        temporary,
        250,
        worker,
        [?MODULE]
    },
    % All these nested cases to attempt starting/restarting a replication child
    % are ugly and not 100% race condition free. The following patch submission
    % is a solution:
    %
    % http://erlang.2086793.n4.nabble.com/PATCH-supervisor-atomically-delete-child-spec-when-child-terminates-td3226098.html
    %
    case supervisor:start_child(couch_replicator_job_sup, ChildSpec) of
    {ok, Pid} ->
        couch_log:notice("starting new replication `~s` at ~p (`~s` -> `~s`)",
            [RepChildId, Pid, Source, Target]),
        {ok, Pid};
    {error, already_present} ->
        case supervisor:restart_child(couch_replicator_job_sup, RepChildId) of
        {ok, Pid} ->
            couch_log:notice("restarting replication `~s` at ~p (`~s` -> `~s`)",
                [RepChildId, Pid, Source, Target]),
            {ok, 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.
            timer:sleep(50 + random:uniform(100)),
            async_replicate(Rep);
        {error, _} = Error ->
            Error
        end;
    {error, {already_started, Pid}} ->
        couch_log:notice("replication `~s` already running at ~p (`~s` -> `~s`)",
            [RepChildId, Pid, Source, Target]),
        {ok, Pid};
    {error, {Error, _}} ->
        {error, Error}
    end.


rep_result_listener(RepId) ->
    ReplyTo = self(),
    {ok, _Listener} = couch_replicator_notifier:start_link(
        fun({_, RepId2, _} = Ev) when RepId2 =:= RepId ->
                ReplyTo ! Ev;
            (_) ->
                ok
        end).


wait_for_result(RepId) ->
    receive
    {finished, RepId, RepResult} ->
        {ok, RepResult};
    {error, RepId, Reason} ->
        {error, Reason}
    end.


cancel_replication({BaseId, Extension}) ->
    FullRepId = BaseId ++ Extension,
    couch_log:notice("Canceling replication `~s`...", [FullRepId]),
    case supervisor:terminate_child(couch_replicator_job_sup, FullRepId) of
    ok ->
        couch_log:notice("Replication `~s` canceled.", [FullRepId]),
        case supervisor:delete_child(couch_replicator_job_sup, FullRepId) of
            ok ->
                {ok, {cancelled, ?l2b(FullRepId)}};
            {error, not_found} ->
                {ok, {cancelled, ?l2b(FullRepId)}};
            Error ->
                Error
        end;
    Error ->
        couch_log:error("Error canceling replication `~s`: ~p", [FullRepId, Error]),
        Error
    end.

cancel_replication(RepId, #user_ctx{name = Name, roles = Roles}) ->
    case lists:member(<<"_admin">>, Roles) of
    true ->
        cancel_replication(RepId);
    false ->
        case find_replicator(RepId) of
        {ok, Pid} ->
            case details(Pid) of
            {ok, #rep{user_ctx = #user_ctx{name = Name}}} ->
                cancel_replication(RepId);
            {ok, _} ->
                throw({unauthorized,
                    <<"Can't cancel a replication triggered by another user">>});
            Error ->
                Error
            end;
        Error ->
            Error
        end
    end.

find_replicator({BaseId, Ext} = _RepId) ->
    case lists:keysearch(
        BaseId ++ Ext, 1, supervisor:which_children(couch_replicator_job_sup)) of
    {value, {_, Pid, _, _}} when is_pid(Pid) ->
            {ok, Pid};
    _ ->
            {error, not_found}
    end.

details(Pid) ->
    case (catch gen_server:call(Pid, get_details)) of
    {ok, Rep} ->
        {ok, Rep};
    {'EXIT', {noproc, {gen_server, call, _}}} ->
        {error, not_found};
    Error ->
        throw(Error)
    end.

init(InitArgs) ->
    {ok, InitArgs, 0}.

do_init(#rep{options = Options, id = {BaseId, Ext}, user_ctx=UserCtx} = Rep) ->
    process_flag(trap_exit, true),

    #rep_state{
        source = Source,
        target = Target,
        source_name = SourceName,
        target_name = TargetName,
        start_seq = {_Ts, StartSeq},
        committed_seq = {_, CommittedSeq},
        highest_seq_done = {_, HighestSeq},
        checkpoint_interval = CheckpointInterval
    } = State = init_state(Rep),

    NumWorkers = get_value(worker_processes, Options),
    BatchSize = get_value(worker_batch_size, Options),
    {ok, ChangesQueue} = couch_work_queue:new([
        {max_items, BatchSize * NumWorkers * 2},
        {max_size, 100 * 1024 * NumWorkers}
    ]),
    % This starts the _changes reader process. It adds the changes from
    % the source db to the ChangesQueue.
    {ok, ChangesReader} = couch_replicator_changes_reader:start_link(
        StartSeq, Source, ChangesQueue, Options
    ),
    % Changes manager - responsible for dequeing batches from the changes queue
    % and deliver them to the worker processes.
    ChangesManager = spawn_changes_manager(self(), ChangesQueue, BatchSize),
    % This starts the worker processes. They ask the changes queue manager for a
    % a batch of _changes rows to process -> check which revs are missing in the
    % target, and for the missing ones, it copies them from the source to the target.
    MaxConns = get_value(http_connections, Options),
    Workers = lists:map(
        fun(_) ->
            couch_stats:increment_counter([couch_replicator, workers_started]),
            {ok, Pid} = couch_replicator_worker:start_link(
                self(), Source, Target, ChangesManager, MaxConns),
            Pid
        end,
        lists:seq(1, NumWorkers)),

    couch_task_status:add_task([
        {type, replication},
        {user, UserCtx#user_ctx.name},
        {replication_id, ?l2b(BaseId ++ Ext)},
        {database, Rep#rep.db_name},
        {doc_id, Rep#rep.doc_id},
        {source, ?l2b(SourceName)},
        {target, ?l2b(TargetName)},
        {continuous, get_value(continuous, Options, false)},
        {revisions_checked, 0},
        {missing_revisions_found, 0},
        {docs_read, 0},
        {docs_written, 0},
        {changes_pending, get_pending_count(State)},
        {doc_write_failures, 0},
        {source_seq, HighestSeq},
        {checkpointed_source_seq, CommittedSeq},
        {checkpoint_interval, CheckpointInterval}
    ]),
    couch_task_status:set_update_frequency(1000),

    % Until OTP R14B03:
    %
    % Restarting a temporary supervised child implies that the original arguments
    % (#rep{} record) specified in the MFA component of the supervisor
    % child spec will always be used whenever the child is restarted.
    % This implies the same replication performance tunning parameters will
    % always be used. The solution is to delete the child spec (see
    % cancel_replication/1) and then start the replication again, but this is
    % unfortunately not immune to race conditions.

    couch_log:notice("Replication `~p` is using:~n"
        "~c~p worker processes~n"
        "~ca worker batch size of ~p~n"
        "~c~p HTTP connections~n"
        "~ca connection timeout of ~p milliseconds~n"
        "~c~p retries per request~n"
        "~csocket options are: ~s~s",
        [BaseId ++ Ext, $\t, NumWorkers, $\t, BatchSize, $\t,
            MaxConns, $\t, get_value(connection_timeout, Options),
            $\t, get_value(retries, Options),
            $\t, io_lib:format("~p", [get_value(socket_options, Options)]),
            case StartSeq of
            ?LOWEST_SEQ ->
                "";
            _ ->
                io_lib:format("~n~csource start sequence ~p", [$\t, StartSeq])
            end]),

    couch_log:debug("Worker pids are: ~p", [Workers]),
<commit_msg>Use transient restart type for all replications<commit_after>% 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_replicator).
-behaviour(gen_server).
-vsn(1).

% public API
-export([replicate/2]).

% meant to be used only by the replicator database listener
-export([async_replicate/1]).
-export([cancel_replication/1]).

% gen_server callbacks
-export([init/1, terminate/2, code_change/3]).
-export([handle_call/3, handle_cast/2, handle_info/2]).
-export([format_status/2]).

-export([details/1]).

-include_lib("couch/include/couch_db.hrl").
-include("couch_replicator_api_wrap.hrl").
-include("couch_replicator.hrl").

-define(LOWEST_SEQ, 0).

-import(couch_util, [
    get_value/2,
    get_value/3,
    to_binary/1
]).

-import(couch_replicator_utils, [
    start_db_compaction_notifier/2,
    stop_db_compaction_notifier/1
]).

-record(rep_state, {
    rep_details,
    source_name,
    target_name,
    source,
    target,
    history,
    checkpoint_history,
    start_seq,
    committed_seq,
    current_through_seq,
    seqs_in_progress = [],
    highest_seq_done = {0, ?LOWEST_SEQ},
    source_log,
    target_log,
    rep_starttime,
    src_starttime,
    tgt_starttime,
    timer, % checkpoint timer
    changes_queue,
    changes_manager,
    changes_reader,
    workers,
    stats = couch_replicator_stats:new(),
    session_id,
    source_db_compaction_notifier = nil,
    target_db_compaction_notifier = nil,
    source_monitor = nil,
    target_monitor = nil,
    source_seq = nil,
    use_checkpoints = true,
    checkpoint_interval = 5000,
    type = db,
    view = nil
}).


replicate(PostBody, Ctx) ->
    {ok, #rep{id = RepId, options = Options, user_ctx = UserCtx} = Rep} =
        couch_replicator_utils:parse_rep_doc(PostBody, Ctx),
    case get_value(cancel, Options, false) of
    true ->
        case get_value(id, Options, nil) of
        nil ->
            cancel_replication(RepId);
        RepId2 ->
            cancel_replication(RepId2, UserCtx)
        end;
    false ->
        {ok, Listener} = rep_result_listener(RepId),
        Result = do_replication_loop(Rep),
        couch_replicator_notifier:stop(Listener),
        Result
    end.


do_replication_loop(#rep{id = {BaseId, Ext} = Id, options = Options} = Rep) ->
    case async_replicate(Rep) of
    {ok, _Pid} ->
        case get_value(continuous, Options, false) of
        true ->
            {ok, {continuous, ?l2b(BaseId ++ Ext)}};
        false ->
            wait_for_result(Id)
        end;
    Error ->
        Error
    end.


async_replicate(#rep{id = {BaseId, Ext}, source = Src, target = Tgt} = Rep) ->
    RepChildId = BaseId ++ Ext,
    Source = couch_replicator_api_wrap:db_uri(Src),
    Target = couch_replicator_api_wrap:db_uri(Tgt),
    Timeout = get_value(connection_timeout, Rep#rep.options),
    ChildSpec = {
        RepChildId,
        {gen_server, start_link, [?MODULE, Rep, [{timeout, Timeout}]]},
        transient,
        250,
        worker,
        [?MODULE]
    },
    % All these nested cases to attempt starting/restarting a replication child
    % are ugly and not 100% race condition free. The following patch submission
    % is a solution:
    %
    % http://erlang.2086793.n4.nabble.com/PATCH-supervisor-atomically-delete-child-spec-when-child-terminates-td3226098.html
    %
    case supervisor:start_child(couch_replicator_job_sup, ChildSpec) of
    {ok, Pid} ->
        couch_log:notice("starting new replication `~s` at ~p (`~s` -> `~s`)",
            [RepChildId, Pid, Source, Target]),
        {ok, Pid};
    {error, already_present} ->
        case supervisor:restart_child(couch_replicator_job_sup, RepChildId) of
        {ok, Pid} ->
            couch_log:notice("restarting replication `~s` at ~p (`~s` -> `~s`)",
                [RepChildId, Pid, Source, Target]),
            {ok, 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.
            timer:sleep(50 + random:uniform(100)),
            async_replicate(Rep);
        {error, _} = Error ->
            Error
        end;
    {error, {already_started, Pid}} ->
        couch_log:notice("replication `~s` already running at ~p (`~s` -> `~s`)",
            [RepChildId, Pid, Source, Target]),
        {ok, Pid};
    {error, {Error, _}} ->
        {error, Error}
    end.


rep_result_listener(RepId) ->
    ReplyTo = self(),
    {ok, _Listener} = couch_replicator_notifier:start_link(
        fun({_, RepId2, _} = Ev) when RepId2 =:= RepId ->
                ReplyTo ! Ev;
            (_) ->
                ok
        end).


wait_for_result(RepId) ->
    receive
    {finished, RepId, RepResult} ->
        {ok, RepResult};
    {error, RepId, Reason} ->
        {error, Reason}
    end.


cancel_replication({BaseId, Extension}) ->
    FullRepId = BaseId ++ Extension,
    couch_log:notice("Canceling replication `~s`...", [FullRepId]),
    case supervisor:terminate_child(couch_replicator_job_sup, FullRepId) of
    ok ->
        couch_log:notice("Replication `~s` canceled.", [FullRepId]),
        case supervisor:delete_child(couch_replicator_job_sup, FullRepId) of
            ok ->
                {ok, {cancelled, ?l2b(FullRepId)}};
            {error, not_found} ->
                {ok, {cancelled, ?l2b(FullRepId)}};
            Error ->
                Error
        end;
    Error ->
        couch_log:error("Error canceling replication `~s`: ~p", [FullRepId, Error]),
        Error
    end.

cancel_replication(RepId, #user_ctx{name = Name, roles = Roles}) ->
    case lists:member(<<"_admin">>, Roles) of
    true ->
        cancel_replication(RepId);
    false ->
        case find_replicator(RepId) of
        {ok, Pid} ->
            case details(Pid) of
            {ok, #rep{user_ctx = #user_ctx{name = Name}}} ->
                cancel_replication(RepId);
            {ok, _} ->
                throw({unauthorized,
                    <<"Can't cancel a replication triggered by another user">>});
            Error ->
                Error
            end;
        Error ->
            Error
        end
    end.

find_replicator({BaseId, Ext} = _RepId) ->
    case lists:keysearch(
        BaseId ++ Ext, 1, supervisor:which_children(couch_replicator_job_sup)) of
    {value, {_, Pid, _, _}} when is_pid(Pid) ->
            {ok, Pid};
    _ ->
            {error, not_found}
    end.

details(Pid) ->
    case (catch gen_server:call(Pid, get_details)) of
    {ok, Rep} ->
        {ok, Rep};
    {'EXIT', {noproc, {gen_server, call, _}}} ->
        {error, not_found};
    Error ->
        throw(Error)
    end.

init(InitArgs) ->
    {ok, InitArgs, 0}.

do_init(#rep{options = Options, id = {BaseId, Ext}, user_ctx=UserCtx} = Rep) ->
    process_flag(trap_exit, true),

    #rep_state{
        source = Source,
        target = Target,
        source_name = SourceName,
        target_name = TargetName,
        start_seq = {_Ts, StartSeq},
        committed_seq = {_, CommittedSeq},
        highest_seq_done = {_, HighestSeq},
        checkpoint_interval = CheckpointInterval
    } = State = init_state(Rep),

    NumWorkers = get_value(worker_processes, Options),
    BatchSize = get_value(worker_batch_size, Options),
    {ok, ChangesQueue} = couch_work_queue:new([
        {max_items, BatchSize * NumWorkers * 2},
        {max_size, 100 * 1024 * NumWorkers}
    ]),
    % This starts the _changes reader process. It adds the changes from
    % the source db to the ChangesQueue.
    {ok, ChangesReader} = couch_replicator_changes_reader:start_link(
        StartSeq, Source, ChangesQueue, Options
    ),
    % Changes manager - responsible for dequeing batches from the changes queue
    % and deliver them to the worker processes.
    ChangesManager = spawn_changes_manager(self(), ChangesQueue, BatchSize),
    % This starts the worker processes. They ask the changes queue manager for a
    % a batch of _changes rows to process -> check which revs are missing in the
    % target, and for the missing ones, it copies them from the source to the target.
    MaxConns = get_value(http_connections, Options),
    Workers = lists:map(
        fun(_) ->
            couch_stats:increment_counter([couch_replicator, workers_started]),
            {ok, Pid} = couch_replicator_worker:start_link(
                self(), Source, Target, ChangesManager, MaxConns),
            Pid
        end,
        lists:seq(1, NumWorkers)),

    couch_task_status:add_task([
        {type, replication},
        {user, UserCtx#user_ctx.name},
        {replication_id, ?l2b(BaseId ++ Ext)},
        {database, Rep#rep.db_name},
        {doc_id, Rep#rep.doc_id},
        {source, ?l2b(SourceName)},
        {target, ?l2b(TargetName)},
        {continuous, get_value(continuous, Options, false)},
        {revisions_checked, 0},
        {missing_revisions_found, 0},
        {docs_read, 0},
        {docs_written, 0},
        {changes_pending, get_pending_count(State)},
        {doc_write_failures, 0},
        {source_seq, HighestSeq},
        {checkpointed_source_seq, CommittedSeq},
        {checkpoint_interval, CheckpointInterval}
    ]),
    couch_task_status:set_update_frequency(1000),

    couch_log:notice("Replication `~p` is using:~n"
        "~c~p worker processes~n"
        "~ca worker batch size of ~p~n"
        "~c~p HTTP connections~n"
        "~ca connection timeout of ~p milliseconds~n"
        "~c~p retries per request~n"
        "~csocket options are: ~s~s",
        [BaseId ++ Ext, $\t, NumWorkers, $\t, BatchSize, $\t,
            MaxConns, $\t, get_value(connection_timeout, Options),
            $\t, get_value(retries, Options),
            $\t, io_lib:format("~p", [get_value(socket_options, Options)]),
            case StartSeq of
            ?LOWEST_SEQ ->
                "";
            _ ->
                io_lib:format("~n~csource start sequence ~p", [$\t, StartSeq])
            end]),

    couch_log:debug("Worker pids are: ~p", [Workers]),

{'fraction_non_alphanumeric': 0.12042732275817417, 'fraction_numerical': 0.007908245849327105, 'mean_word_length': 2.5460806821908823, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 119, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3361771', 'n_tokens_mistral': 7082, 'n_tokens_neox': 6871, 'n_words': 1673}