[couchdb] 03/03: don't crash if index crashes during open or if event listener crashes

[rn...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

rnewson pushed a commit to branch couch_index_fixes
in repository https://gitbox.apache.org/repos/asf/couchdb.git

commit c81c5c5a990ff4b3536f82af69cb9f9aa10dd955
Author: Robert Newson <rn...@apache.org>
AuthorDate: Wed Mar 22 14:31:01 2023 +0000

    don't crash if index crashes during open or if event listener crashes
---
 src/couch_index/src/couch_index_server.erl | 23 ++++++++++++++++++-----
 1 file changed, 18 insertions(+), 5 deletions(-)

diff --git a/src/couch_index/src/couch_index_server.erl b/src/couch_index/src/couch_index_server.erl
index 2e368bfc2..0a2af85a1 100644
--- a/src/couch_index/src/couch_index_server.erl
+++ b/src/couch_index/src/couch_index_server.erl
@@ -41,7 +41,8 @@
     server_name,
     by_sig,
     by_pid,
-    by_db
+    by_db,
+    event_listener
 }).
 
 start_link(N) ->
@@ -143,8 +144,8 @@ init([N]) ->
         by_db = by_db(N)
     },
     ok = config:listen_for_changes(?MODULE, St),
-    couch_event:link_listener(?MODULE, handle_db_event, St, [all_dbs]),
-    {ok, St}.
+    {ok, Pid} = couch_event:link_listener(?MODULE, handle_db_event, St, [all_dbs]),
+    {ok, St#st{event_listener = Pid}}.
 
 terminate(_Reason, State) ->
     Pids = [Pid || {Pid, _} <- ets:tab2list(State#st.by_pid)],
@@ -154,7 +155,8 @@ terminate(_Reason, State) ->
 handle_call({get_index, {_Mod, _IdxState, DbName, Sig} = Args}, From, State) ->
     case ets:lookup(State#st.by_sig, {DbName, Sig}) of
         [] ->
-            spawn_link(fun() -> new_index(Args) end),
+            Pid = spawn_link(fun() -> new_index(Args) end),
+            ets:insert(State#st.by_pid, {Pid, opening, {DbName, Sig}}),
             ets:insert(State#st.by_sig, {{DbName, Sig}, [From]}),
             {noreply, State};
         [{_, Waiters}] when is_list(Waiters) ->
@@ -169,9 +171,10 @@ handle_call({async_open, {DbName, DDocId, Sig}, {ok, Pid}}, _From, State) ->
     link(Pid),
     add_to_ets(DbName, Sig, DDocId, Pid, State),
     {reply, ok, State};
-handle_call({async_error, {DbName, _DDocId, Sig}, Error}, _From, State) ->
+handle_call({async_error, {DbName, _DDocId, Sig}, Error}, {OpenerPid, _}, State) ->
     [{_, Waiters}] = ets:lookup(State#st.by_sig, {DbName, Sig}),
     [gen_server:reply(From, Error) || From <- Waiters],
+    ets:delete(State#st.by_pid, OpenerPid),
     ets:delete(State#st.by_sig, {DbName, Sig}),
     {reply, ok, State};
 handle_call({reset_indexes, DbName}, _From, State) ->
@@ -194,6 +197,10 @@ handle_cast({rem_from_ets, [DbName, DDocId, Sig]}, State) ->
     ets:delete_object(State#st.by_db, {DbName, {DDocId, Sig}}),
     {noreply, State}.
 
+handle_info({'EXIT', Pid, Reason}, #st{event_listener = Pid} = State) ->
+    couch_log:warning("~p: update notifier died ~p", [?MODULE, Reason]),
+    {ok, Pid1} = start_event_listener(),
+    {noreply, State#st{event_listener = Pid1}};
 handle_info({'EXIT', Pid, Reason}, Server) ->
     case ets:lookup(Server#st.by_pid, Pid) of
         [{Pid, {DbName, Sig}}] ->
@@ -203,6 +210,9 @@ handle_info({'EXIT', Pid, Reason}, Server) ->
                     ets:match_object(Server#st.by_db, {DbName, {'$1', Sig}})
             ],
             rem_from_ets(DbName, Sig, DDocIds, Pid, Server);
+        [{_, opening, {DbName, Sig}}] when Reason /= normal ->
+            Msg = {async_error, {DbName, nil, Sig}, Reason},
+            {reply, ok, _} = handle_call(Msg, {Pid, nil}, Server);
         [] when Reason /= normal ->
             exit(Reason);
         _Else ->
@@ -394,3 +404,6 @@ aggregate_queue_len() ->
      || Name <- Names
     ],
     lists:sum([X || {_, X} <- MQs]).
+
+start_event_listener() ->
+    couch_event:link_listener(?MODULE, handle_db_event, nil, [all_dbs]).