Merge pull request #474 from cloudant/3364-fix-view-compactor-unknown_infoarchive/test-branchtest-branch

3364 fix view compactor unknown info

5 files changed, 67 insertions, 5 deletions

diff --git a/src/couch_index/src/couch_index.erl b/src/couch_index/src/couch_index.erl
index b339010a6..c86f5e122 100644
--- a/src/couch_index/src/couch_index.erl
+++ b/src/couch_index/src/couch_index.erl
@@ -192,7 +192,17 @@ handle_call({compacted, NewIdxState}, _From, State) ->
             end;
         false ->
             {reply, ok, commit_compacted(NewIdxState, State)}
-    end.
+    end;
+handle_call({compaction_failed, Reason}, _From, State) ->
+    #st{
+        mod = Mod,
+        idx_state = OldIdxState,
+        waiters = Waiters
+    } = State,
+    send_all(Waiters, Reason),
+    {ok, NewIdxState} = Mod:remove_compacted(OldIdxState),
+    NewState = State#st{idx_state = NewIdxState, waiters = []},
+    {reply, {ok, NewIdxState}, NewState}.
 
 handle_cast({trigger_update, UpdateSeq}, State) ->
     #st{
diff --git a/src/couch_index/src/couch_index_compactor.erl b/src/couch_index/src/couch_index_compactor.erl
index b5db058a5..61f406c1a 100644
--- a/src/couch_index/src/couch_index_compactor.erl
+++ b/src/couch_index/src/couch_index_compactor.erl
@@ -15,7 +15,7 @@
 
 
 %% API
--export([start_link/2, run/2, cancel/1, is_running/1]).
+-export([start_link/2, run/2, cancel/1, is_running/1, get_compacting_pid/1]).
 
 %% gen_server callbacks
 -export([init/1, terminate/2, code_change/3]).
@@ -47,6 +47,8 @@ cancel(Pid) ->
 is_running(Pid) ->
     gen_server:call(Pid, is_running).
 
+get_compacting_pid(Pid) ->
+    gen_server:call(Pid, get_compacting_pid).
 
 init({Index, Module}) ->
     process_flag(trap_exit, true),
@@ -69,6 +71,8 @@ handle_call(cancel, _From, #st{pid=Pid}=State) ->
     unlink(Pid),
     exit(Pid, kill),
     {reply, ok, State#st{pid=undefined}};
+handle_call(get_compacting_pid, _From, #st{pid=Pid}=State) ->
+    {reply, {ok, Pid}, State};
 handle_call(is_running, _From, #st{pid=Pid}=State) when is_pid(Pid) ->
     {reply, true, State};
 handle_call(is_running, _From, State) ->
@@ -81,6 +85,14 @@ handle_cast(_Mesg, State) ->
 
 handle_info({'EXIT', Pid, normal}, #st{pid=Pid}=State) ->
     {noreply, State#st{pid=undefined}};
+handle_info({'EXIT', Pid, Reason}, #st{pid = Pid} = State) ->
+    #st{idx = Idx, mod = Mod} = State,
+    {ok, IdxState} = gen_server:call(Idx, {compaction_failed, Reason}),
+    DbName = Mod:get(db_name, IdxState),
+    IdxName = Mod:get(idx_name, IdxState),
+    Args = [DbName, IdxName, Reason],
+    couch_log:error("Compaction failed for db: ~s idx: ~s reason: ~p", Args),
+    {noreply, State#st{pid = undefined}};
 handle_info({'EXIT', _Pid, normal}, State) ->
     {noreply, State};
 handle_info({'EXIT', Pid, _Reason}, #st{idx=Pid}=State) ->
diff --git a/src/couch_mrview/src/couch_mrview_compactor.erl b/src/couch_mrview/src/couch_mrview_compactor.erl
index 9ef79b664..fabe2894c 100644
--- a/src/couch_mrview/src/couch_mrview_compactor.erl
+++ b/src/couch_mrview/src/couch_mrview_compactor.erl
@@ -15,7 +15,7 @@
 -include_lib("couch/include/couch_db.hrl").
 -include_lib("couch_mrview/include/couch_mrview.hrl").
 
--export([compact/3, swap_compacted/2]).
+-export([compact/3, swap_compacted/2, remove_compacted/1]).
 
 -record(acc, {
    btree = nil,
@@ -294,6 +294,13 @@ swap_compacted(OldState, NewState) ->
     {ok, NewState#mrst{fd_monitor=Ref}}.
 
 
+remove_compacted(#mrst{sig = Sig, db_name = DbName} = State) ->
+    RootDir = couch_index_util:root_dir(),
+    CompactFName = couch_mrview_util:compaction_file(DbName, Sig),
+    ok = couch_file:delete(RootDir, CompactFName),
+    {ok, State}.
+
+
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
 
diff --git a/src/couch_mrview/src/couch_mrview_index.erl b/src/couch_mrview/src/couch_mrview_index.erl
index 0473d5dd2..eaec5cc52 100644
--- a/src/couch_mrview/src/couch_mrview_index.erl
+++ b/src/couch_mrview/src/couch_mrview_index.erl
@@ -16,7 +16,7 @@
 -export([get/2]).
 -export([init/2, open/2, close/1, reset/1, delete/1]).
 -export([start_update/3, purge/4, process_doc/3, finish_update/1, commit/1]).
--export([compact/3, swap_compacted/2]).
+-export([compact/3, swap_compacted/2, remove_compacted/1]).
 -export([index_file_exists/1]).
 
 -include_lib("couch/include/couch_db.hrl").
@@ -184,6 +184,10 @@ swap_compacted(OldState, NewState) ->
     couch_mrview_compactor:swap_compacted(OldState, NewState).
 
 
+remove_compacted(State) ->
+    couch_mrview_compactor:remove_compacted(State).
+
+
 index_file_exists(State) ->
     #mrst{
         db_name=DbName,
diff --git a/src/couch_mrview/test/couch_mrview_compact_tests.erl b/src/couch_mrview/test/couch_mrview_compact_tests.erl
index 079639f5d..40877c80e 100644
--- a/src/couch_mrview/test/couch_mrview_compact_tests.erl
+++ b/src/couch_mrview/test/couch_mrview_compact_tests.erl
@@ -20,9 +20,11 @@
 
 setup() ->
     {ok, Db} = couch_mrview_test_util:init_db(?tempdb(), map, 1000),
+    ok = meck:new(couch_mrview_compactor, [passthrough]),
     Db.
 
 teardown(Db) ->
+    meck:unload(),
     couch_db:close(Db),
     couch_server:delete(Db#db.name, [?ADMIN_CTX]),
     ok.
@@ -38,7 +40,8 @@ compaction_test_() ->
                 foreach,
                 fun setup/0, fun teardown/1,
                 [
-                    fun should_swap/1
+                    fun should_swap/1,
+                    fun should_remove/1
                 ]
             }
         }
@@ -71,6 +74,32 @@ should_swap(Db) ->
     end).
 
 
+should_remove(Db) ->
+    ?_test(begin
+        DDoc = <<"_design/bar">>,
+        {ok, _Results} = couch_mrview:query_view(Db, DDoc, <<"baz">>),
+        {ok, IndexPid} = couch_index_server:get_index(couch_mrview_index, Db, DDoc),
+        ok = couch_index:compact(IndexPid, []),
+        {ok, CompactorPid} = couch_index:get_compactor_pid(IndexPid),
+        {ok, CompactingPid} = couch_index_compactor:get_compacting_pid(CompactorPid),
+        MonRef = erlang:monitor(process, CompactingPid),
+        exit(CompactingPid, crash),
+        receive
+            {'DOWN', MonRef, process, _, crash} ->
+                meck:wait(couch_mrview_compactor, remove_compacted, '_', 100),
+                ?assertEqual(1, meck:num_calls(
+                    couch_mrview_compactor, remove_compacted, '_', IndexPid)),
+                ?assert(is_process_alive(IndexPid)),
+                ?assert(is_process_alive(CompactorPid))
+        after ?TIMEOUT ->
+            erlang:error(
+                {assertion_failed, [
+                    {module, ?MODULE}, {line, ?LINE},
+                    {reason, "compaction didn't exit :/"}]})
+        end
+    end).
+
+
 start_query(Db) ->
     Self = self(),
     Pid = spawn(fun() ->