This is an automated email from the ASF dual-hosted git repository.
nickva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/couchdb.git
The following commit(s) were added to refs/heads/main by this push:
new 0eb3ec67e Handle errors better in index cleanup
0eb3ec67e is described below
commit 0eb3ec67e3450e3ca71846e067227ae4668f0beb
Author: Nick Vatamaniuc <[email protected]>
AuthorDate: Tue Jul 28 19:36:56 2026 -0400
Handle errors better in index cleanup
Previously, if a node was down when cleaning up any indexes we'd get an
exception like:
```
{erpc,noconnection} : node@host
[<<"erpc:receive_response/3 L1442">>,<<"dreyfus_fabric_cleanup:recv/3
L54">>
[<<"erpc:receive_response/3 L1442">>,<<"dreyfus_fabric_cleanup:recv/3
L54">>,...
```
We can handle that better by anticipating a few more error conditions:
noconnection, exits, throws, timeouts.
To avoid copying the mess in three place let's just use a single helper
function in fabric.
Add tests for all the silly bits.
---
src/dreyfus/src/dreyfus_fabric_cleanup.erl | 14 +--
src/fabric/src/fabric_index_cleanup.erl | 35 +++++--
src/fabric/test/eunit/fabric_tests.erl | 143 ++++++++++++++++++++++++++++-
src/nouveau/src/nouveau_fabric_cleanup.erl | 14 +--
4 files changed, 170 insertions(+), 36 deletions(-)
diff --git a/src/dreyfus/src/dreyfus_fabric_cleanup.erl
b/src/dreyfus/src/dreyfus_fabric_cleanup.erl
index 0488211be..f01afc4f3 100644
--- a/src/dreyfus/src/dreyfus_fabric_cleanup.erl
+++ b/src/dreyfus/src/dreyfus_fabric_cleanup.erl
@@ -26,7 +26,7 @@ go(DbName) ->
erpc:send_request(Node, ?MODULE, go_local, [DbName, Dbs,
Sigs], Node, Acc)
end,
Reqs = maps:fold(Fun, erpc:reqids_new(), ByNode),
- recv(DbName, Reqs, fabric_util:abs_request_timeout());
+ fabric_index_cleanup:recv(?MODULE, DbName, Reqs,
fabric_util:abs_request_timeout());
Error ->
couch_log:error("~p : error fetching ddocs db:~p ~p", [?MODULE,
DbName, Error]),
Error
@@ -49,15 +49,3 @@ go_local(DbName, Dbs, #{} = Sigs) ->
error:database_does_not_exist ->
ok
end.
-
-recv(DbName, Reqs, Timeout) ->
- case erpc:receive_response(Reqs, Timeout, true) of
- {ok, _Lable, Reqs1} ->
- recv(DbName, Reqs1, Timeout);
- {Error, Label, Reqs1} ->
- ErrMsg = "~p : error cleaning dreyfus indexes db:~p req:~p
error:~p",
- couch_log:error(ErrMsg, [?MODULE, DbName, Label, Error]),
- recv(DbName, Reqs1, Timeout);
- no_request ->
- ok
- end.
diff --git a/src/fabric/src/fabric_index_cleanup.erl
b/src/fabric/src/fabric_index_cleanup.erl
index 13759ba1d..a329530ab 100644
--- a/src/fabric/src/fabric_index_cleanup.erl
+++ b/src/fabric/src/fabric_index_cleanup.erl
@@ -16,7 +16,8 @@
cleanup_all_nodes/0,
cleanup_all_nodes/1,
cleanup_this_node/0,
- cleanup_this_node/1
+ cleanup_this_node/1,
+ recv/4
]).
cleanup_all_nodes() ->
@@ -55,7 +56,7 @@ cleanup_indexes(DbName, Nodes) ->
Acc3
end,
Reqs = maps:fold(Fun, erpc:reqids_new(), ByNode),
- recv(DbName, Reqs, fabric_util:abs_request_timeout());
+ recv(?MODULE, DbName, Reqs, fabric_util:abs_request_timeout());
Error ->
couch_log:error("~p : error fetching ddocs db:~p ~p", [?MODULE,
DbName, Error]),
Error
@@ -68,14 +69,30 @@ send(Node, M, F, A, Reqs) ->
Label = {Node, M, F},
erpc:send_request(Node, M, F, A, Label, Reqs).
-recv(DbName, Reqs, Timeout) ->
- case erpc:receive_response(Reqs, Timeout, true) of
+% Receive responses for an erpc request collection built with
+% erpc:send_request/6. Used by this module and the dreyfus and nouveau cleanup
+% modules. Cleanup or timeout are best-effort we log them and keep going (a
+% node might be off for hardware replacement or something).
+%
+recv(Module, DbName, Reqs, Timeout) ->
+ try erpc:receive_response(Reqs, Timeout, true) of
{ok, _Label, Reqs1} ->
- recv(DbName, Reqs1, Timeout);
- {Error, Label, Reqs1} ->
- ErrMsg = "~p : error cleaning indexes db:~p req:~p error:~p",
- couch_log:error(ErrMsg, [?MODULE, DbName, Label, Error]),
- recv(DbName, Reqs1, Timeout);
+ recv(Module, DbName, Reqs1, Timeout);
+ {Res, Label, Reqs1} ->
+ log_error(Module, DbName, Label, Res),
+ recv(Module, DbName, Reqs1, Timeout);
no_request ->
ok
+ catch
+ error:{erpc, timeout} ->
+ Labels = [Label || {_ReqId, Label} <- erpc:reqids_to_list(Reqs)],
+ log_error(Module, DbName, Labels, timeout),
+ ok;
+ Class:{Reason, Label, Reqs1} when is_map(Reqs1) ->
+ log_error(Module, DbName, Label, {Class, Reason}),
+ recv(Module, DbName, Reqs1, Timeout)
end.
+
+log_error(Module, DbName, Label, Error) ->
+ ErrMsg = "~p : error cleaning indexes db:~p req:~p error:~p",
+ couch_log:error(ErrMsg, [Module, DbName, Label, Error]).
diff --git a/src/fabric/test/eunit/fabric_tests.erl
b/src/fabric/test/eunit/fabric_tests.erl
index daf018231..38c43e1d5 100644
--- a/src/fabric/test/eunit/fabric_tests.erl
+++ b/src/fabric/test/eunit/fabric_tests.erl
@@ -15,6 +15,7 @@
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
+-include_lib("mem3/include/mem3.hrl").
cleanup_index_files_test_() ->
{
@@ -29,7 +30,9 @@ cleanup_index_files_test_() ->
?TDEF_FE(t_cleanup_index_file_after_ddoc_update),
?TDEF_FE(t_cleanup_index_file_after_ddoc_delete),
?TDEF_FE(t_cleanup_empty_view_checkpoints),
- ?TDEF_FE(t_cleanup_disallowed_language_checkpoints)
+ ?TDEF_FE(t_cleanup_disallowed_language_checkpoints),
+ ?TDEF_FE(t_cleanup_index_files_with_node_down),
+ ?TDEF_FE(t_cleanup_search_indexes_with_node_down)
]
}.
@@ -223,6 +226,60 @@ t_cleanup_disallowed_language_checkpoints({_, DbName}) ->
purges(DbName)
).
+t_cleanup_index_files_with_node_down({_, DbName}) ->
+ FakeNode = 'bogus@nohost',
+ mock_node_down(DbName, FakeNode),
+ meck:new(mem3_util, [passthrough]),
+ meck:expect(mem3_util, live_nodes, fun() -> meck:passthrough([]) ++
[FakeNode] end),
+ ?assertEqual(ok, fabric:cleanup_index_files_all_nodes(DbName)),
+ ErpcError = {error, {erpc, noconnection}},
+ lists:foreach(
+ fun({M, F}) ->
+ Label = {FakeNode, M, F},
+ Args = [fabric_index_cleanup, DbName, Label, ErpcError],
+ ?assert(meck:called(couch_log, error, ['_', Args]))
+ end,
+ [
+ {couch_mrview_cleanup, cleanup},
+ {dreyfus_fabric_cleanup, go_local},
+ {nouveau_fabric_cleanup, go_local}
+ ]
+ ).
+
+t_cleanup_search_indexes_with_node_down({_, DbName}) ->
+ FakeNode = 'bogus@nohost',
+ mock_node_down(DbName, FakeNode),
+ ErpcError = {error, {erpc, noconnection}},
+ ?assertEqual(ok, dreyfus_fabric_cleanup:go(DbName)),
+ Args1 = [dreyfus_fabric_cleanup, DbName, FakeNode, ErpcError],
+ ?assert(meck:called(couch_log, error, ['_', Args1])),
+ ?assertEqual(ok, nouveau_fabric_cleanup:go(DbName)),
+ Args2 = [nouveau_fabric_cleanup, DbName, FakeNode, ErpcError],
+ ?assert(meck:called(couch_log, error, ['_', Args2])).
+
+% Jump through hoops to pretend we have a node down. We mock DbName's shard map
+% to return an extra shard copy on a bogus down node.
+%
+mock_node_down(DbName, FakeNode) ->
+ DDocRes = fabric_util:get_design_doc_records(DbName),
+ meck:new(fabric_util, [passthrough]),
+ meck:expect(fabric_util, get_design_doc_records, fun(Db) ->
+ case Db =:= DbName of
+ true -> DDocRes;
+ false -> meck:passthrough([Db])
+ end
+ end),
+ meck:new(mem3, [passthrough]),
+ meck:expect(mem3, shards, fun(Db) ->
+ Shards = meck:passthrough([Db]),
+ case Db =:= DbName of
+ true -> Shards ++ [(hd(Shards))#shard{node = FakeNode}];
+ false -> Shards
+ end
+ end),
+ meck:new(couch_log, [passthrough]),
+ meck:expect(couch_log, error, 2, ok).
+
shard_names(DbName) ->
[mem3:name(S) || S <- mem3:local_shards(DbName)].
@@ -494,3 +551,87 @@ teardown_fabric(Ctx) ->
clear_shards_db() ->
ShardsDb = ?l2b(config:get("mem3", "shards_db", "_dbs")),
couch_server:delete(ShardsDb, [?ADMIN_CTX]).
+
+index_cleanup_recv_test_() ->
+ {
+ foreach,
+ fun() -> meck:expect(couch_log, error, 2, ok) end,
+ fun(_) -> meck:unload() end,
+ [
+ ?TDEF_FE(t_recv_no_requests),
+ ?TDEF_FE(t_recv_ok_responses),
+ ?TDEF_FE(t_recv_logs_non_ok_responses),
+ ?TDEF_FE(t_recv_handles_noconnection),
+ ?TDEF_FE(t_recv_handles_remote_exceptions),
+ ?TDEF_FE(t_recv_handles_timeout)
+ ]
+ }.
+
+t_recv_no_requests(_) ->
+ ?assertEqual(ok, recv(erpc:reqids_new(), 5000)),
+ ?assertEqual(0, meck:num_calls(couch_log, error, 2)).
+
+t_recv_ok_responses(_) ->
+ Reqs0 = erpc:reqids_new(),
+ Reqs1 = send_fun(node(), req1, fun() -> ok end, Reqs0),
+ Reqs2 = send_fun(node(), req2, fun() -> ok end, Reqs1),
+ ?assertEqual(ok, recv(Reqs2, 5000)),
+ ?assertEqual(0, meck:num_calls(couch_log, error, 2)).
+
+t_recv_logs_non_ok_responses(_) ->
+ Reqs1 = send_fun(node(), req1, fun() -> {error, potato} end,
erpc:reqids_new()),
+ ?assertEqual(ok, recv(Reqs1, 5000)),
+ ?assert(meck:called(couch_log, error, ['_', ['_', '_', req1, {error,
potato}]])).
+
+t_recv_handles_noconnection(_) ->
+ Self = self(),
+ Reqs0 = erpc:reqids_new(),
+ Reqs1 = erpc:send_request('bogus@totallybogus', erlang, node, [],
down_node, Reqs0),
+ Reqs2 = send_fun(
+ node(),
+ req2,
+ fun() ->
+ Self ! req2_ran,
+ ok
+ end,
+ Reqs1
+ ),
+ ?assertEqual(ok, recv(Reqs2, 5000)),
+ % Log the bogus one but keep going otherwise
+ NoConn = {error, {erpc, noconnection}},
+ ?assert(meck:called(couch_log, error, ['_', ['_', '_', down_node,
NoConn]])),
+ ?assertEqual(1, meck:num_calls(couch_log, error, 2)),
+ receive
+ req2_ran -> ok
+ end.
+
+t_recv_handles_remote_exceptions(_) ->
+ % A variety of failures on the other side: exits, errors and throws
+ Reqs0 = erpc:reqids_new(),
+ Reqs1 = send_fun(node(), err_req, fun() -> error(potato) end, Reqs0),
+ Reqs2 = send_fun(node(), throw_req, fun() -> throw(potato) end, Reqs1),
+ Reqs3 = send_fun(node(), exit_req, fun() -> exit(potato) end, Reqs2),
+ Reqs4 = send_fun(node(), ok_req, fun() -> ok end, Reqs3),
+ ?assertEqual(ok, recv(Reqs4, 5000)),
+ ?assert(
+ meck:called(couch_log, error, ['_', ['_', '_', err_req, {error,
{exception, potato, '_'}}]])
+ ),
+ ?assert(meck:called(couch_log, error, ['_', ['_', '_', throw_req, {throw,
potato}]])),
+ ?assert(
+ meck:called(couch_log, error, ['_', ['_', '_', exit_req, {exit,
{exception, potato}}]])
+ ),
+ ?assertEqual(3, meck:num_calls(couch_log, error, 2)).
+
+t_recv_handles_timeout(_) ->
+ Reqs1 = send_fun(node(), slow_req, fun() -> timer:sleep(10000) end,
erpc:reqids_new()),
+ ?assertEqual(ok, recv(Reqs1, 100)),
+ % Test the global timeout
+ ?assert(meck:called(couch_log, error, ['_', ['_', '_', [slow_req],
timeout]])).
+
+recv(Reqs, TimeoutMSec) ->
+ % Note: this is a standard erpc format see
https://www.erlang.org/doc/apps/kernel/erpc.html
+ Timeout = {abs, erlang:monotonic_time(millisecond) + TimeoutMSec},
+ fabric_index_cleanup:recv(?MODULE, <<"db">>, Reqs, Timeout).
+
+send_fun(Node, Label, Fun, Reqs) ->
+ erpc:send_request(Node, erlang, apply, [Fun, []], Label, Reqs).
diff --git a/src/nouveau/src/nouveau_fabric_cleanup.erl
b/src/nouveau/src/nouveau_fabric_cleanup.erl
index 75c2190b8..d67c6f9c9 100644
--- a/src/nouveau/src/nouveau_fabric_cleanup.erl
+++ b/src/nouveau/src/nouveau_fabric_cleanup.erl
@@ -26,7 +26,7 @@ go(DbName) ->
erpc:send_request(Node, ?MODULE, go_local, [DbName, Dbs,
Sigs], Node, Acc)
end,
Reqs = maps:fold(Fun, erpc:reqids_new(), ByNode),
- recv(DbName, Reqs, fabric_util:abs_request_timeout());
+ fabric_index_cleanup:recv(?MODULE, DbName, Reqs,
fabric_util:abs_request_timeout());
Error ->
couch_log:error("~p : error fetching ddocs db:~p ~p", [?MODULE,
DbName, Error]),
Error
@@ -51,15 +51,3 @@ go_local(DbName, Dbs, Sigs) ->
error:database_does_not_exist ->
ok
end.
-
-recv(DbName, Reqs, Timeout) ->
- case erpc:receive_response(Reqs, Timeout, true) of
- {ok, _Label, Reqs1} ->
- recv(DbName, Reqs1, Timeout);
- {Error, Label, Reqs1} ->
- ErrMsg = "~p : error cleaning nouveau indexes db:~p node: ~p
error:~p",
- couch_log:error(ErrMsg, [?MODULE, DbName, Label, Error]),
- recv(DbName, Reqs1, Timeout);
- no_request ->
- ok
- end.
