<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_compact_tests).

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

-import(couch_replicator_test_helper, [
    db_url/1,
    get_pid/1
]).

-define(ATTFILE, filename:join([?FIXTURESDIR, "logo.png"])).
-define(DELAY, 100).
-define(TIMEOUT, 30000).
-define(TIMEOUT_WRITER, 18000).
-define(TIMEOUT_EUNIT, ?TIMEOUT div 1000 + 70).

setup() ->
    DbName = ?tempdb(),
    {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
    ok = couch_db:close(Db),
    DbName.

setup(local) ->
    setup();
setup(remote) ->
    {remote, setup()};
setup({A, B}) ->
    Ctx = test_util:start_couch([couch_replicator]),
    Source = setup(A),
    Target = setup(B),
    {Ctx, {Source, Target}}.

teardown({remote, DbName}) ->
    teardown(DbName);
teardown(DbName) ->
    ok = couch_server:delete(DbName, [?ADMIN_CTX]),
    ok.

teardown(_, {Ctx, {Source, Target}}) ->
    teardown(Source),
    teardown(Target),
    ok = application:stop(couch_replicator),
    ok = test_util:stop_couch(Ctx).

compact_test_() ->
    Pairs = [{local, local}, {local, remote},
             {remote, local}, {remote, remote}],
    {
        "Compaction during replication tests",
        {
            foreachx,
            fun setup/1, fun teardown/2,
            [{Pair, fun should_populate_replicate_compact/2}
             || Pair <- Pairs]
        }
    }.


should_populate_replicate_compact({From, To}, {_Ctx, {Source, Target}}) ->
    {ok, RepPid, RepId} = replicate(Source, Target),
    {lists:flatten(io_lib:format("~p -> ~p", [From, To])),
     {inorder, [
         should_run_replication(RepPid, RepId, Source, Target),
         should_all_processes_be_alive(RepPid, Source, Target),
         should_populate_and_compact(RepPid, Source, Target, 50, 5),
         should_wait_target_in_sync(Source, Target),
         should_ensure_replication_still_running(RepPid, RepId, Source, Target),
         should_cancel_replication(RepId, RepPid),
         should_compare_databases(Source, Target)
     ]}}.

should_all_processes_be_alive(RepPid, Source, Target) ->
    ?_test(begin
        {ok, SourceDb} = reopen_db(Source),
        {ok, TargetDb} = reopen_db(Target),
        ?assert(is_process_alive(RepPid)),
        ?assert(is_process_alive(SourceDb#db.main_pid)),
        ?assert(is_process_alive(TargetDb#db.main_pid))
    end).

should_run_replication(RepPid, RepId, Source, Target) ->
    ?_test(check_active_tasks(RepPid, RepId, Source, Target)).

should_ensure_replication_still_running(RepPid, RepId, Source, Target) ->
    ?_test(check_active_tasks(RepPid, RepId, Source, Target)).

check_active_tasks(RepPid, {BaseId, Ext} = RepId, Src, Tgt) ->
    Source = case Src of
        {remote, NameSrc} ->
            <<(db_url(NameSrc))/binary, $/>>;
        _ ->
            Src
    end,
    Target = case Tgt of
        {remote, NameTgt} ->
            <<(db_url(NameTgt))/binary, $/>>;
        _ ->
            Tgt
    end,
    FullRepId = ?l2b(BaseId ++ Ext),
    Pid = ?l2b(pid_to_list(RepPid)),
    ok = wait_for_replicator(RepId),
    RepTasks = wait_for_task_status(),
    ?assertNotEqual(timeout, RepTasks),
    [RepTask] = RepTasks,
    ?assertEqual(Pid, couch_util:get_value(pid, RepTask)),
    ?assertEqual(FullRepId, couch_util:get_value(replication_id, RepTask)),
    ?assertEqual(true, couch_util:get_value(continuous, RepTask)),
    ?assertEqual(Source, couch_util:get_value(source, RepTask)),
    ?assertEqual(Target, couch_util:get_value(target, RepTask)),
    ?assert(is_integer(couch_util:get_value(docs_read, RepTask))),
    ?assert(is_integer(couch_util:get_value(docs_written, RepTask))),
    ?assert(is_integer(couch_util:get_value(doc_write_failures, RepTask))),
    ?assert(is_integer(couch_util:get_value(revisions_checked, RepTask))),
    ?assert(is_integer(couch_util:get_value(missing_revisions_found, RepTask))),
    ?assert(is_integer(couch_util:get_value(checkpointed_source_seq, RepTask))),
    ?assert(is_integer(couch_util:get_value(source_seq, RepTask))),
    Pending = couch_util:get_value(changes_pending, RepTask),
    ?assert(is_integer(Pending)).

rep_details(RepId) ->
    gen_server:call(get_pid(RepId), get_details).

replication_tasks() ->
    lists:filter(fun(P) ->
        couch_util:get_value(type, P) =:= replication
    end, couch_task_status:all()).

wait_for_replicator(RepId) ->
    %% since replicator started asynchronously
    %% we need to wait when it would be in couch_task_status
    %% we query replicator:details to ensure that do_init happen
    ?assertMatch({ok, _}, rep_details(RepId)),
    ok.

wait_for_task_status() ->
    test_util:wait(fun() ->
        case couch_task_status:all() of
            [] ->
                wait;
            Tasks ->
                Tasks
        end
    end).

should_cancel_replication(RepId, RepPid) ->
    ?_assertNot(begin
        ok = couch_replicator_scheduler:remove_job(RepId),
        is_process_alive(RepPid)
    end).

should_populate_and_compact(RepPid, Source, Target, BatchSize, Rounds) ->
    {timeout, ?TIMEOUT_EUNIT, ?_test(begin
        {ok, SourceDb0} = reopen_db(Source),
        Writer = spawn_writer(SourceDb0),
        lists:foreach(
            fun(N) -><commit_msg>Fix merge of 27f5067c and aca6a459<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_compact_tests).

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

-import(couch_replicator_test_helper, [
    db_url/1,
    get_pid/1
]).

-define(ATTFILE, filename:join([?FIXTURESDIR, "logo.png"])).
-define(DELAY, 100).
-define(TIMEOUT, 30000).
-define(TIMEOUT_WRITER, 18000).
-define(TIMEOUT_EUNIT, ?TIMEOUT div 1000 + 70).

setup() ->
    DbName = ?tempdb(),
    {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
    ok = couch_db:close(Db),
    DbName.

setup(local) ->
    setup();
setup(remote) ->
    {remote, setup()};
setup({A, B}) ->
    Ctx = test_util:start_couch([couch_replicator]),
    Source = setup(A),
    Target = setup(B),
    {Ctx, {Source, Target}}.

teardown({remote, DbName}) ->
    teardown(DbName);
teardown(DbName) ->
    ok = couch_server:delete(DbName, [?ADMIN_CTX]),
    ok.

teardown(_, {Ctx, {Source, Target}}) ->
    teardown(Source),
    teardown(Target),
    ok = application:stop(couch_replicator),
    ok = test_util:stop_couch(Ctx).

compact_test_() ->
    Pairs = [{local, local}, {local, remote},
             {remote, local}, {remote, remote}],
    {
        "Compaction during replication tests",
        {
            foreachx,
            fun setup/1, fun teardown/2,
            [{Pair, fun should_populate_replicate_compact/2}
             || Pair <- Pairs]
        }
    }.


should_populate_replicate_compact({From, To}, {_Ctx, {Source, Target}}) ->
    {ok, RepPid, RepId} = replicate(Source, Target),
    {lists:flatten(io_lib:format("~p -> ~p", [From, To])),
     {inorder, [
         should_run_replication(RepPid, RepId, Source, Target),
         should_all_processes_be_alive(RepPid, Source, Target),
         should_populate_and_compact(RepPid, Source, Target, 50, 5),
         should_wait_target_in_sync(Source, Target),
         should_ensure_replication_still_running(RepPid, RepId, Source, Target),
         should_cancel_replication(RepId, RepPid),
         should_compare_databases(Source, Target)
     ]}}.

should_all_processes_be_alive(RepPid, Source, Target) ->
    ?_test(begin
        {ok, SourceDb} = reopen_db(Source),
        {ok, TargetDb} = reopen_db(Target),
        ?assert(is_process_alive(RepPid)),
        ?assert(is_process_alive(SourceDb#db.main_pid)),
        ?assert(is_process_alive(TargetDb#db.main_pid))
    end).

should_run_replication(RepPid, RepId, Source, Target) ->
    ?_test(check_active_tasks(RepPid, RepId, Source, Target)).

should_ensure_replication_still_running(RepPid, RepId, Source, Target) ->
    ?_test(check_active_tasks(RepPid, RepId, Source, Target)).

check_active_tasks(RepPid, {BaseId, Ext} = RepId, Src, Tgt) ->
    Source = case Src of
        {remote, NameSrc} ->
            <<(db_url(NameSrc))/binary, $/>>;
        _ ->
            Src
    end,
    Target = case Tgt of
        {remote, NameTgt} ->
            <<(db_url(NameTgt))/binary, $/>>;
        _ ->
            Tgt
    end,
    FullRepId = ?l2b(BaseId ++ Ext),
    Pid = ?l2b(pid_to_list(RepPid)),
    ok = wait_for_replicator(RepId),
    RepTasks = wait_for_task_status(),
    ?assertNotEqual(timeout, RepTasks),
    [RepTask] = RepTasks,
    ?assertEqual(Pid, couch_util:get_value(pid, RepTask)),
    ?assertEqual(FullRepId, couch_util:get_value(replication_id, RepTask)),
    ?assertEqual(true, couch_util:get_value(continuous, RepTask)),
    ?assertEqual(Source, couch_util:get_value(source, RepTask)),
    ?assertEqual(Target, couch_util:get_value(target, RepTask)),
    ?assert(is_integer(couch_util:get_value(docs_read, RepTask))),
    ?assert(is_integer(couch_util:get_value(docs_written, RepTask))),
    ?assert(is_integer(couch_util:get_value(doc_write_failures, RepTask))),
    ?assert(is_integer(couch_util:get_value(revisions_checked, RepTask))),
    ?assert(is_integer(couch_util:get_value(missing_revisions_found, RepTask))),
    ?assert(is_integer(couch_util:get_value(checkpointed_source_seq, RepTask))),
    ?assert(is_integer(couch_util:get_value(source_seq, RepTask))),
    Pending = couch_util:get_value(changes_pending, RepTask),
    ?assert(is_integer(Pending)).

rep_details(RepId) ->
    gen_server:call(get_pid(RepId), get_details).

replication_tasks() ->
    lists:filter(fun(P) ->
        couch_util:get_value(type, P) =:= replication
    end, couch_task_status:all()).

wait_for_replicator(RepId) ->
    %% since replicator started asynchronously
    %% we need to wait when it would be in couch_task_status
    %% we query replicator:details to ensure that do_init happen
    ?assertMatch({ok, _}, rep_details(RepId)),
    ok.

wait_for_task_status() ->
    test_util:wait(fun() ->
        case replication_tasks() of
            [] ->
                wait;
            Tasks ->
                Tasks
        end
    end).

replication_tasks() ->
    lists:filter(fun(P) ->
        couch_util:get_value(type, P) =:= replication
    end, couch_task_status:all()).

should_cancel_replication(RepId, RepPid) ->
    ?_assertNot(begin
        ok = couch_replicator_scheduler:remove_job(RepId),
        is_process_alive(RepPid)
    end).

should_populate_and_compact(RepPid, Source, Target, BatchSize, Rounds) ->
    {timeout, ?TIMEOUT_EUNIT, ?_test(begin
        {ok, SourceDb0} = reopen_db(Source),
        Writer = spawn_writer(SourceDb0),
        lists:foreach(
            fun(N) ->
{'fraction_non_alphanumeric': 0.1341885391172251, 'fraction_numerical': 0.006751624609671702, 'mean_word_length': 3.6727129337539433, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 71, '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': '4891070', 'n_tokens_mistral': 4183, 'n_tokens_neox': 4064, 'n_words': 758}