path: root/src/couch/test/eunit/couch_bt_engine_compactor_ev_tests.erl

% 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_bt_engine_compactor_ev_tests).

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

-define(TIMEOUT_EUNIT, 60).
-define(EV_MOD, couch_bt_engine_compactor_ev).
-define(INIT_DOCS, 2500).
-define(WRITE_DOCS, 20).

% The idea behind the tests in this module are to attempt to
% cover the number of restart/recopy events during compaction
% so that we can be as sure as possible that the compactor
% is resilient to errors in the face of external conditions
% (i.e., the VM rebooted). The single linear pass is easy enough
% to prove, however restarting is important enough that we don't
% want to waste work if a VM happens to bounce a lot.
%
% To try and cover as many restart situations we have created a
% number of events in the compactor code that are present during
% a test compiled version of the module. These events can then
% be used (via meck) to introduce errors and coordinate writes
% to the database while compaction is in progress.

% This list of events is where we'll insert our errors.

events() ->
    [
        init,               % The compactor process is spawned
        files_opened,       % After compaction files have opened

        purge_init,         % Just before apply purge changes
        purge_done,         % Just after finish purge updates

        % The firs phase is when we copy all document body and attachment
        % data to the new database file in order of update sequence so
        % that we can resume on crash.

        seq_init,           % Before the first change is copied
        {seq_copy, 0},      % After change N is copied
        {seq_copy, ?INIT_DOCS div 2},
        {seq_copy, ?INIT_DOCS - 2},
        seq_done,           % After last change is copied

        % The id copy phases come in two flavors. Before a compaction
        % swap is attempted they're copied from the id_tree in the
        % database being compacted. After a swap attempt they are
        % stored in an emsort file on disk. Thus the two sets of
        % related events here.

        md_sort_init,       % Just before metadata sort starts
        md_sort_done,       % Justa after metadata sort finished
        md_copy_init,       % Just before metadata copy starts
        {md_copy_row, 0},   % After docid N is copied
        {md_copy_row, ?INIT_DOCS div 2},
        {md_copy_row, ?INIT_DOCS - 2},
        md_copy_done,       % Just after the last docid is copied

        % And then the final steps before we finish

        before_final_sync,  % Just before final sync
        after_final_sync,   % Just after the final sync
        before_notify       % Just before the final notification
    ].

% Mark which evens only happen when documents are present

requires_docs({seq_copy, _}) -> true;
requires_docs(md_sort_init) -> true;
requires_docs(md_sort_done) -> true;
requires_docs(md_copy_init) -> true;
requires_docs({md_copy_row, _}) -> true;
requires_docs(md_copy_done) -> true;
requires_docs(_) -> false.


% Mark which events only happen when there's write activity during
% a compaction.

requires_write(md_sort_init) -> true;
requires_write(md_sort_done) -> true;
requires_write(md_copy_init) -> true;
requires_write({md_copy_row, _}) -> true;
requires_write(md_copy_done) -> true;
requires_write(_) -> false.


setup() ->
    purge_module(),
    ?EV_MOD:init(),
    test_util:start_couch().


teardown(Ctx) ->
    test_util:stop_couch(Ctx),
    ?EV_MOD:terminate().


start_empty_db_test(_Event) ->
    ?EV_MOD:clear(),
    DbName = ?tempdb(),
    {ok, _} = couch_db:create(DbName, [?ADMIN_CTX]),
    DbName.


start_populated_db_test(Event) ->
    DbName = start_empty_db_test(Event),
    {ok, Db} = couch_db:open_int(DbName, []),
    try
        populate_db(Db, ?INIT_DOCS)
    after
        couch_db:close(Db)
    end,
    DbName.


stop_test(_Event, DbName) ->
    couch_server:delete(DbName, [?ADMIN_CTX]).


static_empty_db_test_() ->
    FiltFun = fun(E) ->
        not (requires_docs(E) or requires_write(E))
    end,
    Events = lists:filter(FiltFun, events()) -- [init],
    {
        "Idle empty database",
        {
            setup,
            fun setup/0,
            fun teardown/1,
            [
                {
                    foreachx,
                    fun start_empty_db_test/1,
                    fun stop_test/2,
                    [{Event, fun run_static_init/2} || Event <- Events]
                }
            ]
        }
    }.


static_populated_db_test_() ->
    FiltFun = fun(E) -> not requires_write(E) end,
    Events = lists:filter(FiltFun, events()) -- [init],
    {
        "Idle populated database",
        {
            setup,
            fun setup/0,
            fun teardown/1,
            [
                {
                    foreachx,
                    fun start_populated_db_test/1,
                    fun stop_test/2,
                    [{Event, fun run_static_init/2} || Event <- Events]
                }
            ]
        }
    }.


dynamic_empty_db_test_() ->
    FiltFun = fun(E) -> not requires_docs(E) end,
    Events = lists:filter(FiltFun, events()) -- [init],
    {
        "Writes to empty database",
        {
            setup,
            fun setup/0,
            fun teardown/1,
            [
                {
                    foreachx,
                    fun start_empty_db_test/1,
                    fun stop_test/2,
                    [{Event, fun run_dynamic_init/2} || Event <- Events]
                }
            ]
        }
    }.


dynamic_populated_db_test_() ->
    Events = events() -- [init],
    {
        "Writes to populated database",
        {
            setup,
            fun setup/0,
            fun teardown/1,
            [
                {
                    foreachx,
                    fun start_populated_db_test/1,
                    fun stop_test/2,
                    [{Event, fun run_dynamic_init/2} || Event <- Events]
                }
            ]
        }
    }.


run_static_init(Event, DbName) ->
    Name = lists:flatten(io_lib:format("~p", [Event])),
    Test = {timeout, ?TIMEOUT_EUNIT, ?_test(run_static(Event, DbName))},
    {Name, Test}.


run_static(Event, DbName) ->
    {ok, ContinueFun} = ?EV_MOD:set_wait(init),
    {ok, Reason} = ?EV_MOD:set_crash(Event),
    {ok, Db} = couch_db:open_int(DbName, []),
    Ref = couch_db:monitor(Db),
    {ok, CPid} = couch_db:start_compact(Db),
    ContinueFun(CPid),
    receive
        {'DOWN', Ref, _, _, Reason} ->
            wait_db_cleared(Db)
    end,
    run_successful_compaction(DbName),
    couch_db:close(Db).


run_dynamic_init(Event, DbName) ->
    Name = lists:flatten(io_lib:format("~p", [Event])),
    Test = {timeout, ?TIMEOUT_EUNIT, ?_test(run_dynamic(Event, DbName))},
    {Name, Test}.


run_dynamic(Event, DbName) ->
    {ok, ContinueFun} = ?EV_MOD:set_wait(init),
    {ok, Reason} = ?EV_MOD:set_crash(Event),
    {ok, Db} = couch_db:open_int(DbName, []),
    Ref = couch_db:monitor(Db),
    {ok, CPid} = couch_db:start_compact(Db),
    ok = populate_db(Db, 10),
    ContinueFun(CPid),
    receive
        {'DOWN', Ref, _, _, Reason} ->
            wait_db_cleared(Db)
    end,
    run_successful_compaction(DbName),
    couch_db:close(Db).


run_successful_compaction(DbName) ->
    ?EV_MOD:clear(),
    {ok, ContinueFun} = ?EV_MOD:set_wait(init),
    {ok, Db} = couch_db:open_int(DbName, []),
    {ok, CPid} = couch_db:start_compact(Db),
    Ref = erlang:monitor(process, CPid),
    ContinueFun(CPid),
    receive
        {'DOWN', Ref, _, _, normal} -> ok
    end,
    Pid = couch_db:get_pid(Db),
    {ok, NewDb} = gen_server:call(Pid, get_db),
    validate_compaction(NewDb),
    couch_db:close(Db).


wait_db_cleared(Db) ->
    wait_db_cleared(Db, 5).


wait_db_cleared(Db, N) when N < 0 ->
    erlang:error({db_clear_timeout, couch_db:name(Db)});

wait_db_cleared(Db, N) ->
    Tab = couch_server:couch_dbs(couch_db:name(Db)),
    case ets:lookup(Tab, couch_db:name(Db)) of
        [] ->
            ok;
        [#entry{db = NewDb}] ->
            OldPid = couch_db:get_pid(Db),
            NewPid = couch_db:get_pid(NewDb),
            if NewPid /= OldPid -> ok; true ->
                timer:sleep(100),
                wait_db_cleared(Db, N - 1)
            end
    end.


populate_db(_Db, NumDocs) when NumDocs =< 0 ->
    ok;
populate_db(Db, NumDocs) ->
    String = [$a || _ <- lists:seq(1, erlang:min(NumDocs, 500))],
    Docs = lists:map(
        fun(_) ->
            couch_doc:from_json_obj({[
                {<<"_id">>, couch_uuids:random()},
                {<<"string">>, list_to_binary(String)}
            ]})
        end,
        lists:seq(1, 500)),
    {ok, _} = couch_db:update_docs(Db, Docs, []),
    populate_db(Db, NumDocs - 500).


validate_compaction(Db) ->
    {ok, DocCount} = couch_db:get_doc_count(Db),
    {ok, DelDocCount} = couch_db:get_del_doc_count(Db),
    NumChanges = couch_db:count_changes_since(Db, 0),
    FoldFun = fun(FDI, {PrevId, CountAcc}) ->
        ?assert(FDI#full_doc_info.id > PrevId),
        {ok, {FDI#full_doc_info.id, CountAcc + 1}}
    end,
    {ok, {_, LastCount}} = couch_db:fold_docs(Db, FoldFun, {<<>>, 0}),
    ?assertEqual(DocCount + DelDocCount, LastCount),
    ?assertEqual(NumChanges, LastCount).


purge_module() ->
    case code:which(couch_db_updater) of
        cover_compiled ->
            ok;
        _ ->
            code:delete(couch_db_updater),
            code:purge(couch_db_updater)
    end.