Tests for replicator use_checkpoints option

Author: vmx

src/couch_replicator/Makefile.am

@@ -41,7 +41,8 @@ test_files = \
 	test/03-replication-compact.t \
 	test/04-replication-large-atts.t \
 	test/05-replication-many-leaves.t \
-	test/06-doc-missing-stubs.t
+	test/06-doc-missing-stubs.t \
+	test/07-use-checkpoints.t
 
 compiled_files = \
 	ebin/couch_replicator_api_wrap.beam \

src/couch_replicator/test/07-use-checkpoints.t

@@ -0,0 +1,256 @@
+#!/usr/bin/env escript
+%% -*- erlang -*-
+% 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.
+
+% Verify that compacting databases that are being used as the source or
+% target of a replication doesn't affect the replication and that the
+% replication doesn't hold their reference counters forever.
+
+-define(b2l(B), binary_to_list(B)).
+
+-record(user_ctx, {
+    name = null,
+    roles = [],
+    handler
+}).
+
+-record(doc, {
+    id = <<"">>,
+    revs = {0, []},
+    body = {[]},
+    atts = [],
+    deleted = false,
+    meta = []
+}).
+
+-record(db, {
+    main_pid = nil,
+    update_pid = nil,
+    compactor_pid = nil,
+    instance_start_time, % number of microsecs since jan 1 1970 as a binary string
+    fd,
+    updater_fd,
+    fd_ref_counter,
+    header = nil,
+    committed_update_seq,
+    fulldocinfo_by_id_btree,
+    docinfo_by_seq_btree,
+    local_docs_btree,
+    update_seq,
+    name,
+    filepath,
+    validate_doc_funs = [],
+    security = [],
+    security_ptr = nil,
+    user_ctx = #user_ctx{},
+    waiting_delayed_commit = nil,
+    revs_limit = 1000,
+    fsync_options = [],
+    options = [],
+    compression,
+    before_doc_update,
+    after_doc_read
+}).
+
+-record(rep, {
+    id,
+    source,
+    target,
+    options,
+    user_ctx,
+    doc_id
+}).
+
+
+source_db_name() -> <<"couch_test_rep_db_a">>.
+target_db_name() -> <<"couch_test_rep_db_b">>.
+
+
+main(_) ->
+    test_util:init_code_path(),
+
+    etap:plan(16),
+    case (catch test()) of
+        ok ->
+            etap:end_tests();
+        Other ->
+            etap:diag(io_lib:format("Test died abnormally: ~p", [Other])),
+            etap:bail(Other)
+    end,
+    ok.
+
+
+test() ->
+    couch_server_sup:start_link(test_util:config_files()),
+    ibrowse:start(),
+
+    test_use_checkpoints(false),
+    test_use_checkpoints(true),
+
+    couch_server_sup:stop(),
+    ok.
+
+
+test_use_checkpoints(UseCheckpoints) ->
+    Pairs = [
+        {source_db_name(), target_db_name()},
+        {{remote, source_db_name()}, target_db_name()},
+        {source_db_name(), {remote, target_db_name()}},
+        {{remote, source_db_name()}, {remote, (target_db_name())}}
+    ],
+
+    ListenerFun = case UseCheckpoints of
+    false ->
+        fun({finished, _, {CheckpointHistory}}) ->
+            etap:is(CheckpointHistory,
+            [{<<"use_checkpoints">>,false}],
+            "No checkpoints found");
+        (_) ->
+            ok
+        end;
+    true ->
+        fun({finished, _, {CheckpointHistory}}) ->
+            SessionId = lists:keyfind(
+                <<"session_id">>, 1, CheckpointHistory),
+            etap:isnt(SessionId, false, "There's a checkpoint");
+        (_) ->
+            ok
+        end
+    end,
+    {ok, Listener} = couch_replicator_notifier:start_link(ListenerFun),
+
+    lists:foreach(
+        fun({Source, Target}) ->
+            {ok, SourceDb} = create_db(source_db_name()),
+            etap:diag("Populating source database"),
+            populate_db(SourceDb, 100),
+            ok = couch_db:close(SourceDb),
+
+            etap:diag("Creating target database"),
+            {ok, TargetDb} = create_db(target_db_name()),
+            ok = couch_db:close(TargetDb),
+
+            etap:diag("Setup replicator notifier listener"),
+
+            etap:diag("Triggering replication"),
+            replicate(Source, Target, UseCheckpoints),
+
+            etap:diag("Replication finished, comparing source and target databases"),
+            compare_dbs(SourceDb, TargetDb),
+
+            etap:diag("Deleting databases"),
+            delete_db(TargetDb),
+            delete_db(SourceDb),
+
+            ok = timer:sleep(1000)
+        end,
+        Pairs),
+
+    couch_replicator_notifier:stop(Listener).
+
+
+populate_db(Db, DocCount) ->
+    Docs = lists:foldl(
+        fun(DocIdCounter, Acc) ->
+            Id = iolist_to_binary(["doc", integer_to_list(DocIdCounter)]),
+            Value = iolist_to_binary(["val", integer_to_list(DocIdCounter)]),
+            Doc = #doc{
+                id = Id,
+                body = {[ {<<"value">>, Value} ]}
+            },
+            [Doc | Acc]
+        end,
+        [], lists:seq(1, DocCount)),
+    {ok, _} = couch_db:update_docs(Db, Docs, []).
+
+
+compare_dbs(#db{name = SourceName}, #db{name = TargetName}) ->
+    {ok, SourceDb} = couch_db:open_int(SourceName, []),
+    {ok, TargetDb} = couch_db:open_int(TargetName, []),
+    Fun = fun(FullDocInfo, _, Acc) ->
+        {ok, Doc} = couch_db:open_doc(SourceDb, FullDocInfo),
+        {Props} = DocJson = couch_doc:to_json_obj(Doc, [attachments]),
+        DocId = couch_util:get_value(<<"_id">>, Props),
+        DocTarget = case couch_db:open_doc(TargetDb, DocId) of
+        {ok, DocT} ->
+            DocT;
+        Error ->
+            etap:bail("Error opening document '" ++ ?b2l(DocId) ++
+                "' from target: " ++ couch_util:to_list(Error))
+        end,
+        DocTargetJson = couch_doc:to_json_obj(DocTarget, [attachments]),
+        case DocTargetJson of
+        DocJson ->
+            ok;
+        _ ->
+            etap:bail("Content from document '" ++ ?b2l(DocId) ++
+                "' differs in target database")
+        end,
+        {ok, Acc}
+    end,
+    {ok, _, _} = couch_db:enum_docs(SourceDb, Fun, [], []),
+    etap:diag("Target database has the same documents as the source database"),
+    ok = couch_db:close(SourceDb),
+    ok = couch_db:close(TargetDb).
+
+
+db_url(DbName) ->
+    iolist_to_binary([
+        "http://", couch_config:get("httpd", "bind_address", "127.0.0.1"),
+        ":", integer_to_list(mochiweb_socket_server:get(couch_httpd, port)),
+        "/", DbName
+    ]).
+
+
+create_db(DbName) ->
+    {ok, Db} = couch_db:create(
+        DbName,
+        [{user_ctx, #user_ctx{roles = [<<"_admin">>]}}, overwrite]),
+    couch_db:close(Db),
+    {ok, Db}.
+
+
+delete_db(#db{name = DbName, main_pid = Pid}) ->
+    ok = couch_server:delete(
+        DbName, [{user_ctx, #user_ctx{roles = [<<"_admin">>]}}]),
+    MonRef = erlang:monitor(process, Pid),
+    receive
+    {'DOWN', MonRef, process, Pid, _Reason} ->
+        ok
+    after 30000 ->
+        etap:bail("Timeout deleting database")
+    end.
+
+
+replicate({remote, Db}, Target, UseCheckpoints) ->
+    replicate(db_url(Db), Target, UseCheckpoints);
+
+replicate(Source, {remote, Db}, UseCheckpoints) ->
+    replicate(Source, db_url(Db), UseCheckpoints);
+
+replicate(Source, Target, UseCheckpoints) ->
+    RepObject = {[
+        {<<"source">>, Source},
+        {<<"target">>, Target},
+        {<<"use_checkpoints">>, UseCheckpoints}
+    ]},
+    {ok, Rep} = couch_replicator_utils:parse_rep_doc(
+        RepObject, #user_ctx{roles = [<<"_admin">>]}),
+    {ok, Pid} = couch_replicator:async_replicate(Rep),
+    MonRef = erlang:monitor(process, Pid),
+    receive
+    {'DOWN', MonRef, process, Pid, Reason} ->
+        etap:is(Reason, normal, "Replication finished successfully")
+    after 300000 ->
+        etap:bail("Timeout waiting for replication to finish")
+    end.