Stop search indexer when it is deleted.

Allow dreyfus to listen to ddoc_updated events and check if the index
is deleted as a result of update to it's design document. If the index
is deleted then it will stop the index if waitlist is empty, if the
waitlist is not empty then it will wait for a minute before shutting
down.

BugzId:85718

Author: brkolla

src/dreyfus_index.erl

@@ -157,6 +157,18 @@ handle_call(info, _From, State) -> % obsolete
     Reply = info_int(State#state.index_pid),
     {reply, Reply, State}.
 
+handle_cast({ddoc_updated, DDocResult},
+            #state{
+                index=#index{sig=Sig, dbname=DbName}=Index
+            }=State) ->
+    case check_if_index_is_deleted(DbName, DDocResult, Sig) of
+        true ->
+            couch_log:notice("Index ~s will be closed as it's not active or valid anymore", [index_name(Index)]),
+            {stop, {shutdown, ddoc_updated}, State};
+        false ->
+            {noreply, State}
+    end;
+
 handle_cast(_Msg, State) ->
     {noreply, State}.
 
@@ -214,13 +226,65 @@ handle_info({'DOWN',_,_,Pid,Reason}, #state{
     [gen_server:reply(P, {error, Reason}) || {P, _} <- WaitList],
     {stop, normal, State}.
 
-terminate(_Reason, _State) ->
-    ok.
+terminate(Reason, State) ->
+    case Reason of
+        {shutdown, ddoc_updated} ->
+            send_all(State#state.waiting_list, ddoc_updated),
+            ok;
+        _ ->
+            ok
+    end.
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
 
 % private functions.
+send_all(Waiters, Reply) ->
+    [gen_server:reply(From, Reply) || {From, _} <- Waiters].
+
+check_if_index_is_deleted(DbName, DDocResult, CurrentSig) ->
+    case DDocResult of
+        {not_found, deleted} ->
+            IndicesWithSig = indices_with_sig_from_other_ddocs(DbName, CurrentSig);
+        {ok, DDoc} ->
+            IndicesWithSig = indices_with_sig(DbName, DDoc, CurrentSig)
+    end,
+
+    case IndicesWithSig of
+        [] ->
+            true;
+        [_H | _T] ->
+            false
+    end.
+
+indices_with_sig(DbName, Doc, Sig) ->
+    case indices_matching_sig({Doc, Sig}) of
+        [] ->
+            % Index is deleted from this design doc, check if it's referred from any other design docs.
+            indices_with_sig_from_other_ddocs(DbName, Sig);
+        Indices ->
+            Indices
+    end.
+
+indices_with_sig_from_other_ddocs(DbName, CurrentSig) ->
+    {ok, DesignDocs} = fabric:design_docs(mem3:dbname(DbName)),
+    ActiveSigs = lists:usort(lists:flatmap(
+      fun indices_matching_sig/1,
+        [{couch_doc:from_json_obj(DD), CurrentSig} || DD <- DesignDocs])),
+    ActiveSigs.
+
+indices_matching_sig({#doc{body={Fields}}=Doc, CurrentSig}) ->
+    {RawIndexes} = couch_util:get_value(<<"indexes">>, Fields, {[]}),
+    {IndexNames, _} = lists:unzip(RawIndexes),
+    lists:filter(fun check_if_index_matches_sig/1, [{Doc, IndexName, CurrentSig} || IndexName <- IndexNames]).
+
+check_if_index_matches_sig({Doc, IndexName, Sig}) ->
+  case design_doc_to_index(Doc, IndexName) of
+      {ok, #index{sig=Sig_New}} when Sig_New =/= Sig ->
+          false;
+      {ok, #index{sig=Sig_New}} when Sig_New =:= Sig ->
+          true
+  end.
 
 open_index(DbName, #index{analyzer=Analyzer, sig=Sig}) ->
     Path = <<DbName/binary,"/",Sig/binary>>,

src/dreyfus_index_manager.erl

@@ -21,6 +21,7 @@
 
 -define(BY_SIG, dreyfus_by_sig).
 -define(BY_PID, dreyfus_by_pid).
+-define(BY_DB, dreyfus_by_db).
 
 % public api.
 -export([start_link/0, get_index/2, get_disk_size/2]).
@@ -44,8 +45,9 @@ get_disk_size(DbName, Index) ->
 % gen_server functions.
 
 init([]) ->
-    ets:new(?BY_SIG, [set, private, named_table]),
+    ets:new(?BY_SIG, [set, protected, named_table]),
     ets:new(?BY_PID, [set, private, named_table]),
+    ets:new(?BY_DB, [bag, protected, named_table]),
     couch_event:link_listener(?MODULE, handle_db_event, nil, [all_dbs]),
     process_flag(trap_exit, true),
     {ok, nil}.
@@ -69,12 +71,12 @@ handle_call({get_disk_size, DbName, #index{sig=Sig}=Index}, From, State) ->
     Reply = clouseau_rpc:disk_size(Path),
     {reply, Reply, State};
 
-handle_call({open_ok, DbName, Sig, NewPid}, {OpenerPid, _}, State) ->
+handle_call({open_ok, DbName, DDocId, Sig, NewPid}, {OpenerPid, _}, State) ->
     link(NewPid),
     [{_, WaitList}] = ets:lookup(?BY_SIG, {DbName, Sig}),
     [gen_server:reply(From, {ok, NewPid}) || From <- WaitList],
     ets:delete(?BY_PID, OpenerPid),
-    add_to_ets(NewPid, DbName, Sig),
+    add_to_ets(NewPid, DbName, DDocId, Sig),
     {reply, ok, State};
 
 handle_call({open_error, DbName, Sig, Error}, {OpenerPid, _}, State) ->
@@ -121,25 +123,49 @@ handle_db_event(DbName, created, _St) ->
 handle_db_event(DbName, deleted, _St) ->
     gen_server:cast(?MODULE, {cleanup, DbName}),
     {ok, nil};
+handle_db_event(<<"shards/", _/binary>> = DbName, {ddoc_updated, DDocId}, _St) ->
+    DDocResult = couch_util:with_db(DbName, fun(Db) ->
+        couch_db:open_doc(Db, DDocId, [ejson_body, ?ADMIN_CTX])
+    end),
+    couch_log:info("Received ddoc_updated event for ~s",[DDocId]),
+    DbShards = [mem3:name(Sh) || Sh <- mem3:local_shards(mem3:dbname(DbName))],
+    lists:foreach(fun(DbShard) ->
+        lists:foreach(fun({_DbShard, {_DDocId, Sig}}) ->
+            case ets:lookup(?BY_SIG, {DbShard, Sig}) of
+                [{_, IndexPid}] ->
+                    gen_server:cast(IndexPid, {ddoc_updated, DDocResult});
+                [] ->
+                    ok
+            end
+        end, ets:match_object(?BY_DB, {DbShard, {DDocId, '_'}}))
+    end, DbShards),
+    {ok, nil};
 handle_db_event(_DbName, _Event, _St) ->
     {ok, nil}.
 
-new_index(DbName, #index{sig=Sig}=Index) ->
+new_index(DbName, #index{ddoc_id=DDocId, sig=Sig}=Index) ->
     case (catch dreyfus_index:start_link(DbName, Index)) of
     {ok, NewPid} ->
-        Msg = {open_ok, DbName, Sig, NewPid},
+        Msg = {open_ok, DbName, DDocId, Sig, NewPid},
         ok = gen_server:call(?MODULE, Msg, infinity),
         unlink(NewPid);
     Error ->
         Msg = {open_error, DbName, Sig, Error},
         ok = gen_server:call(?MODULE, Msg, infinity)
     end.
 
-add_to_ets(Pid, DbName, Sig) ->
+add_to_ets(Pid, DbName, DDocId, Sig) ->
     true = ets:insert(?BY_PID, {Pid, {DbName, Sig}}),
-    true = ets:insert(?BY_SIG, {{DbName, Sig}, Pid}).
+    true = ets:insert(?BY_SIG, {{DbName, Sig}, Pid}),
+    true = ets:insert(?BY_DB, {DbName, {DDocId, Sig}}).
 
 delete_from_ets(Pid, DbName, Sig) ->
+    case ets:match_object(?BY_DB, {DbName, {'_', Sig}}) of
+        [{DbName, {DDocId, Sig}}] ->
+            true = ets:delete_object(?BY_DB, {DbName, {DDocId, Sig}});
+        _Else ->
+            true
+    end,
     true = ets:delete(?BY_PID, Pid),
     true = ets:delete(?BY_SIG, {DbName, Sig}).