COUCHDB 3326 clustered purge

src/couch_mrview_cleanup.erl

@@ -41,7 +41,23 @@ run(Db) ->
 
     lists:foreach(fun(FN) ->
         couch_log:debug("Deleting stale view file: ~s", [FN]),
-        couch_file:delete(RootDir, FN, [sync])
+        couch_file:delete(RootDir, FN, [sync]),
+        Sig = couch_mrview_util:get_signature_from_filename(FN),
+        if length(Sig) < 16 -> ok; true ->
+            case re:run(Sig,"^[a-fA-F0-9]+$",[{capture, none}]) of
+                match ->
+                    DocId = couch_mrview_util:get_local_purge_doc_id(Sig),
+                    case couch_db:open_doc(Db, DocId, []) of
+                        {ok, LocalPurgeDoc} ->
+                            couch_db:update_doc(Db,
+                                LocalPurgeDoc#doc{deleted=true}, [?ADMIN_CTX]);
+                        {not_found, _} ->
+                            ok
+                    end;
+                _ ->
+                    ok
+            end
+        end
     end, ToDelete),
 
     ok.

src/couch_mrview_index.erl

@@ -18,6 +18,7 @@
 -export([start_update/3, purge/4, process_doc/3, finish_update/1, commit/1]).
 -export([compact/3, swap_compacted/2]).
 -export([index_file_exists/1]).
+-export([update_local_purge_doc/2, verify_index_exists/1]).
 
 -include_lib("couch/include/couch_db.hrl").
 -include_lib("couch_mrview/include/couch_mrview.hrl").
@@ -134,14 +135,17 @@ open(Db, State) ->
                 {ok, {OldSig, Header}} ->
                     % Matching view signatures.
                     NewSt = couch_mrview_util:init_state(Db, Fd, State, Header),
+                    maybe_create_local_purge_doc(Db, NewSt),
                     {ok, NewSt};
                 % end of upgrade code for <= 1.2.x
                 {ok, {Sig, Header}} ->
                     % Matching view signatures.
                     NewSt = couch_mrview_util:init_state(Db, Fd, State, Header),
+                    maybe_create_local_purge_doc(Db, NewSt),
                     {ok, NewSt};
                 _ ->
                     NewSt = couch_mrview_util:reset_index(Db, Fd, State),
+                    maybe_create_local_purge_doc(Db, NewSt),
                     {ok, NewSt}
             end;
         {error, Reason} = Error ->
@@ -204,3 +208,75 @@ index_file_exists(State) ->
     } = State,
     IndexFName = couch_mrview_util:index_file(DbName, Sig),
     filelib:is_file(IndexFName).
+
+
+update_local_purge_doc(Db, State) ->
+    Sig = couch_index_util:hexsig(get(signature, State)),
+    Doc = couch_doc:from_json_obj({[
+        {<<"_id">>, couch_mrview_util:get_local_purge_doc_id(Sig)},
+        {<<"purge_seq">>, get(purge_seq, State)},
+        {<<"timestamp_utc">>, couch_util:utc_string()},
+        {<<"verify_module">>, <<"couch_mrview_index">>},
+        {<<"verify_function">>, <<"verify_index_exists">>},
+        {<<"verify_options">>, {[
+            {<<"dbname">>, get(db_name, State)},
+            {<<"ddoc_id">>, get(idx_name, State)},
+            {<<"signature">>, Sig}
+        ]}},
+        {<<"type">>, <<"mrview">>}
+    ]}),
+    couch_db:update_doc(Db, Doc, []).
+
+
+verify_index_exists(Options) ->
+    ShardDbName = couch_mrview_util:get_value_from_options(
+        <<"dbname">>,
+        Options
+    ),
+    DDocId = couch_mrview_util:get_value_from_options(
+        <<"ddoc_id">>,
+        Options
+    ),
+    SigInLocal = couch_mrview_util:get_value_from_options(
+        <<"signature">>,
+        Options
+    ),
+    case couch_db:open_int(ShardDbName, []) of
+        {ok, Db} ->
+            try
+                DbName = mem3:dbname(couch_db:name(Db)),
+                case ddoc_cache:open(DbName, DDocId) of
+                    {ok, DDoc} ->
+                        {ok, IdxState} = couch_mrview_util:ddoc_to_mrst(
+                            ShardDbName,
+                            DDoc
+                        ),
+                        IdxSig = IdxState#mrst.sig,
+                        couch_index_util:hexsig(IdxSig) == SigInLocal;
+                    _Else ->
+                        false
+                end
+            catch E:T ->
+                Stack = erlang:get_stacktrace(),
+                couch_log:error(
+                    "Error occurs when verifying existence of ~s/~s :: ~p ~p",
+                    [ShardDbName, DDocId, {E, T}, Stack]
+                ),
+                false
+            after
+                catch couch_db:close(Db)
+            end;
+        _ ->
+            false
+    end.
+
+
+maybe_create_local_purge_doc(Db, State) ->
+    Sig = couch_index_util:hexsig(get(signature, State)),
+    LocalPurgeDocId = couch_mrview_util:get_local_purge_doc_id(Sig),
+    case couch_db:open_doc(Db, LocalPurgeDocId, []) of
+        {not_found, _Reason} ->
+            update_local_purge_doc(Db, State);
+        {ok, _LocalPurgeDoc} ->
+            ok
+    end.

src/couch_mrview_test_util.erl

@@ -44,6 +44,11 @@ save_docs(Db, Docs) ->
 make_docs(Count) ->
     [doc(I) || I <- lists:seq(1, Count)].
 
+
+make_docs(Since, Count) ->
+    [doc(I) || I <- lists:seq(Since, Count)].
+
+
 ddoc(changes) ->
     couch_doc:from_json_obj({[
         {<<"_id">>, <<"_design/bar">>},

src/couch_mrview_util.erl

@@ -13,6 +13,8 @@
 -module(couch_mrview_util).
 
 -export([get_view/4, get_view_index_pid/4]).
+-export([get_local_purge_doc_id/1, get_value_from_options/2]).
+-export([get_signature_from_filename/1]).
 -export([ddoc_to_mrst/2, init_state/4, reset_index/3]).
 -export([make_header/1]).
 -export([index_file/2, compaction_file/2, open_file/1]).
@@ -39,6 +41,27 @@
 -include_lib("couch_mrview/include/couch_mrview.hrl").
 
 
+get_local_purge_doc_id(Sig) ->
+    Version = "v" ++ config:get("purge", "version", "1") ++ "-",
+    ?l2b(?LOCAL_DOC_PREFIX ++ "purge-" ++ Version ++ "mrview-" ++ Sig).
+
+
+get_value_from_options(Key, Options) ->
+    case couch_util:get_value(Key, Options) of
+        undefined ->
+            Reason = binary_to_list(Key) ++ " must exist in Options.",
+            throw({bad_request, Reason});
+        Value -> Value
+    end.
+
+
+get_signature_from_filename(FileName) ->
+    FilePathList = filename:split(FileName),
+    [PureFN] = lists:nthtail(length(FilePathList) - 1, FilePathList),
+    PureFNExt = filename:extension(PureFN),
+    filename:basename(PureFN, PureFNExt).
+
+
 get_view(Db, DDoc, ViewName, Args0) ->
     {ok, Pid, Args2} = get_view_index_pid(Db, DDoc, ViewName, Args0),
     DbUpdateSeq = couch_util:with_db(Db, fun(WDb) ->

test/couch_mrview_purge_docs_fabric_tests.erl

@@ -0,0 +1,97 @@
+% 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_mrview_purge_docs_fabric_tests).
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl").
+
+-define(TIMEOUT, 1000).
+
+
+setup() ->
+    DbName = ?tempdb(),
+    ok = fabric:create_db(DbName, [?ADMIN_CTX]),
+    DbName.
+
+
+teardown(DbName) ->
+    ok = fabric:delete_db(DbName, [?ADMIN_CTX]).
+
+
+view_purge_fabric_test_() ->
+    {
+        "Map views",
+        {
+            setup,
+            fun() -> test_util:start_couch([fabric, mem3]) end,
+            fun test_util:stop_couch/1,
+            {
+                foreach,
+                fun setup/0, fun teardown/1,
+                [
+                    fun test_purge_verify_index/1
+                ]
+            }
+        }
+    }.
+
+
+test_purge_verify_index(DbName) ->
+    ?_test(begin
+        Docs1 = couch_mrview_test_util:make_docs(5),
+        {ok, _} = fabric:update_docs(DbName, Docs1, [?ADMIN_CTX]),
+        {ok, _} = fabric:update_doc(
+            DbName,
+            couch_mrview_test_util:ddoc(map),
+            [?ADMIN_CTX]
+        ),
+
+        purge_docs(DbName, [<<"1">>]),
+
+        Result2 = fabric:query_view(DbName, <<"bar">>, <<"baz">>, #mrargs{}),
+        Expect2 = {ok, [
+            {meta, [{total, 4}, {offset, 0}]},
+            {row, [{id, <<"2">>}, {key, 2}, {value, 2}]},
+            {row, [{id, <<"3">>}, {key, 3}, {value, 3}]},
+            {row, [{id, <<"4">>}, {key, 4}, {value, 4}]},
+            {row, [{id, <<"5">>}, {key, 5}, {value, 5}]}
+        ]},
+        ?assertEqual(Expect2, Result2),
+
+        {ok, DDoc} = fabric:open_doc(DbName, <<"_design/bar">>, []),
+        {ok, IdxState} = couch_mrview_util:ddoc_to_mrst(DbName, DDoc),
+        Sig = IdxState#mrst.sig,
+        HexSig = list_to_binary(couch_index_util:hexsig(Sig)),
+        DocId = couch_mrview_util:get_local_purge_doc_id(HexSig),
+        {ok, LocPurgeDoc} = fabric:open_doc(DbName, DocId, []),
+        {Props} = couch_doc:to_json_obj(LocPurgeDoc,[]),
+        {Options} = couch_util:get_value(<<"verify_options">>, Props),
+        ?assertEqual(true, couch_mrview_index:verify_index_exists(Options)),
+
+        ok
+    end).
+
+get_rev(#full_doc_info{} = FDI) ->
+    #doc_info{
+        revs = [#rev_info{} = PrevRev | _]
+    } = couch_doc:to_doc_info(FDI),
+    PrevRev#rev_info.rev.
+
+
+purge_docs(DbName, DocIds) ->
+    lists:foreach(fun(DocId) ->
+        FDI = fabric:get_full_doc_info(DbName, DocId, []),
+        Rev = get_rev(FDI),
+        {ok, [{ok, _}]} = fabric:purge_docs(DbName, [{DocId, [Rev]}], [])
+    end, DocIds).