Make sure view compaction terminates

If a view group is compacting and the corresponding database
is shutdown by the LRU system, then the view compaction is
aborted because its couch view group process shutdowns.
This could lead to situations where the number of active
databases is much higher than max_dbs_open and making it
impossible to compact view groups.

Issue reported and patch tested by Mike Leddy. Thanks.

COUCHDB-1283

This is a backport of revision 1171328 from branch 1.2.x

git-svn-id: https://svn.apache.org/repos/asf/couchdb/branches/1.1.x@1171329 13f79535-47bb-0310-9956-ffa450edef68

src/couchdb/couch_httpd_db.erl

@@ -128,7 +128,7 @@ handle_changes_req1(Req, Db) ->
 handle_compact_req(#httpd{method='POST',path_parts=[DbName,_,Id|_]}=Req, Db) ->
     ok = couch_db:check_is_admin(Db),
     couch_httpd:validate_ctype(Req, "application/json"),
-    ok = couch_view_compactor:start_compact(DbName, Id),
+    {ok, _} = couch_view_compactor:start_compact(DbName, Id),
     send_json(Req, 202, {[{ok, true}]});
 
 handle_compact_req(#httpd{method='POST'}=Req, Db) ->

src/couchdb/couch_view_compactor.erl

@@ -20,7 +20,7 @@
 %% @doc Compacts the views.  GroupId must not include the _design/ prefix
 start_compact(DbName, GroupId) ->
     Pid = couch_view:get_group_server(DbName, <<"_design/",GroupId/binary>>),
-    gen_server:cast(Pid, {start_compact, fun compact_group/3}).
+    gen_server:call(Pid, {start_compact, fun compact_group/3}).
 
 %%=============================================================================
 %% internal functions
@@ -42,7 +42,6 @@ compact_group(Group, EmptyGroup, DbName) ->
 
     {ok, Db} = couch_db:open_int(DbName, []),
     {ok, {Count, _}} = couch_btree:full_reduce(Db#db.fulldocinfo_by_id_btree),
-    couch_db:close(Db),
 
     <<"_design", ShortName/binary>> = GroupId,
     TaskName = <<DbName/binary, ShortName/binary>>,
@@ -77,9 +76,23 @@ compact_group(Group, EmptyGroup, DbName) ->
         views=NewViews,
         current_seq=Seq
     },
+    maybe_retry_compact(Db, GroupId, NewGroup).
 
+maybe_retry_compact(#db{name = DbName} = Db, GroupId, NewGroup) ->
     Pid = couch_view:get_group_server(DbName, GroupId),
-    gen_server:cast(Pid, {compact_done, NewGroup}).
+    case gen_server:call(Pid, {compact_done, NewGroup}) of
+    ok ->
+        couch_db:close(Db);
+    update ->
+        {ok, Db2} = couch_db:reopen(Db),
+        {_, Ref} = erlang:spawn_monitor(fun() ->
+            couch_view_updater:update(nil, NewGroup, Db2)
+        end),
+        receive
+        {'DOWN', Ref, _, _, {new_group, NewGroup2}} ->
+            maybe_retry_compact(Db2, GroupId, NewGroup2)
+        end
+    end.
 
 %% @spec compact_view(View, EmptyView, Retry) -> CompactView
 compact_view(View, EmptyView) ->

src/couchdb/couch_view_group.erl

@@ -151,9 +151,9 @@ handle_call({request_group, RequestSeq}, From,
 
 handle_call(request_group_info, _From, State) ->
     GroupInfo = get_group_info(State),
-    {reply, {ok, GroupInfo}, State}.
+    {reply, {ok, GroupInfo}, State};
 
-handle_cast({start_compact, CompactFun}, #group_state{compactor_pid=nil}
+handle_call({start_compact, CompactFun}, _From, #group_state{compactor_pid=nil}
         = State) ->
     #group_state{
         group = #group{name = GroupId, sig = GroupSig} = Group,
@@ -165,12 +165,12 @@ handle_cast({start_compact, CompactFun}, #group_state{compactor_pid=nil}
     NewGroup = reset_file(Db, Fd, DbName, Group),
     couch_db:close(Db),
     Pid = spawn_link(fun() -> CompactFun(Group, NewGroup, DbName) end),
-    {noreply, State#group_state{compactor_pid = Pid}};
-handle_cast({start_compact, _}, State) ->
+    {reply, {ok, Pid}, State#group_state{compactor_pid = Pid}};
+handle_call({start_compact, _}, _From, #group_state{compactor_pid=Pid} = State) ->
     %% compact already running, this is a no-op
-    {noreply, State};
+    {reply, {ok, Pid}, State};
 
-handle_cast({compact_done, #group{current_seq=NewSeq} = NewGroup},
+handle_call({compact_done, #group{current_seq=NewSeq} = NewGroup}, _From,
         #group_state{group = #group{current_seq=OldSeq}} = State)
         when NewSeq >= OldSeq ->
     #group_state{
@@ -206,31 +206,20 @@ handle_cast({compact_done, #group{current_seq=NewSeq} = NewGroup},
     {ok, NewRefCounter} = couch_ref_counter:start([NewGroup#group.fd]),
 
     self() ! delayed_commit,
-    {noreply, State#group_state{
+    {reply, ok, State#group_state{
         group=NewGroup,
         ref_counter=NewRefCounter,
         compactor_pid=nil,
         updater_pid=NewUpdaterPid
     }};
-handle_cast({compact_done, NewGroup}, State) ->
+handle_call({compact_done, NewGroup}, _From, State) ->
     #group_state{
         group = #group{name = GroupId, current_seq = CurrentSeq},
         init_args={_RootDir, DbName, _}
     } = State,
     ?LOG_INFO("View index compaction still behind for ~s ~s -- current: ~p " ++
         "compact: ~p", [DbName, GroupId, CurrentSeq, NewGroup#group.current_seq]),
-    Pid = spawn_link(fun() ->
-        {_,Ref} = erlang:spawn_monitor(fun() ->
-            couch_view_updater:update(nil, NewGroup, DbName)
-        end),
-        receive
-            {'DOWN', Ref, _, _, {new_group, NewGroup2}} ->
-                #group{name=GroupId} = NewGroup2,
-                Pid2 = couch_view:get_group_server(DbName, GroupId),
-                gen_server:cast(Pid2, {compact_done, NewGroup2})
-        end
-    end),
-    {noreply, State#group_state{compactor_pid = Pid}};
+    {reply, update, State}.
 
 handle_cast({partial_update, Pid, NewGroup}, #group_state{updater_pid=Pid}
         = State) ->

src/couchdb/couch_view_updater.erl

@@ -18,15 +18,22 @@
 
 -spec update(_, #group{}, Dbname::binary()) -> no_return().
 
-update(Owner, Group, DbName) ->
+update(Owner, Group, DbName) when is_binary(DbName) ->
+    {ok, Db} = couch_db:open_int(DbName, []),
+    try
+        update(Owner, Group, Db)
+    after
+        couch_db:close(Db)
+    end;
+
+update(Owner, Group, #db{name = DbName} = Db) ->
     #group{
         name = GroupName,
         current_seq = Seq,
         purge_seq = PurgeSeq
     } = Group,
     couch_task_status:add_task(<<"View Group Indexer">>, <<DbName/binary," ",GroupName/binary>>, <<"Starting index update">>),
 
-    {ok, Db} = couch_db:open_int(DbName, []),
     DbPurgeSeq = couch_db:get_purge_seq(Db),
     Group2 =
     if DbPurgeSeq == PurgeSeq ->
@@ -36,7 +43,6 @@ update(Owner, Group, DbName) ->
         purge_index(Group, Db);
     true ->
         couch_task_status:update(<<"Resetting view index due to lost purge entries.">>),
-        couch_db:close(Db),
         exit(reset)
     end,
     {ok, MapQueue} = couch_work_queue:new(
@@ -74,7 +80,6 @@ update(Owner, Group, DbName) ->
     couch_task_status:set_update_frequency(0),
     couch_task_status:update("Finishing."),
     couch_work_queue:close(MapQueue),
-    couch_db:close(Db),
     receive {new_group, NewGroup} ->
         exit({new_group,
                 NewGroup#group{current_seq=couch_db:get_update_seq(Db)}})

test/etap/200-view-group-no-db-leaks.t

@@ -165,7 +165,7 @@ wait_db_compact_done(N) ->
     end.
 
 compact_view_group() ->
-    ok = couch_view_compactor:start_compact(test_db_name(), ddoc_name()),
+    {ok, _} = couch_view_compactor:start_compact(test_db_name(), ddoc_name()),
     wait_view_compact_done(10).
 
 wait_view_compact_done(0) ->