Support Erlang/OTP 29

.github/workflows/ci.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       matrix:
         os: ['ubuntu-24.04']
-        otp: ['28', '27', '26']
+        otp: ['29', '28', '27']
         rebar3: ['3.27']
     steps:
     - uses: actions/checkout@v4
@@ -40,7 +40,7 @@ jobs:
     strategy:
       matrix:
         os: ['ubuntu-24.04']
-        otp: ['28', '27', '26']
+        otp: ['29', '28', '27']
         rebar3: ['3.27']
     steps:
     - uses: actions/checkout@v4

.github/workflows/release.yml

@@ -15,7 +15,7 @@ jobs:
       - uses: actions/checkout@v6
       - uses: erlef/setup-beam@v1
         with:
-          otp-version: "28"
+          otp-version: "29"
           rebar3-version: "3.27"
       - run: rebar3 compile
       - run: rebar3 hex publish -r hexpm --yes

.github/workflows/revert-release.yml

@@ -18,7 +18,7 @@ jobs:
       - uses: actions/checkout@v6
       - uses: erlef/setup-beam@v1
         with:
-          otp-version: "28"
+          otp-version: "29"
           rebar3-version: "3.27"
       - run: rebar3 hex publish --revert ${{ inputs.version }} -r hexpm --yes
         env:

rebar.config

@@ -37,7 +37,7 @@
     {test, [
         {deps, [
             {logger_debug_h, "0.2.0"},
-            {meck, "1.0.0"}
+            {meck, "1.2.0"}
         ]},
         {plugins, [
             {rebar3_codecov, "0.7.0"}

src/cets.erl

@@ -563,7 +563,7 @@ handle_call(ping_all, From, State = #{other_servers := Servers}) ->
     proc_lib:spawn(fun() ->
         %% If ping crashes, the caller would not receive a reply.
         %% So, we have to use catch to still able to reply with ok.
-        Results = lists:map(fun(Server) -> {Server, catch ping(Server)} end, Servers),
+        Results = lists:map(fun ping_server/1, Servers),
         BadResults = [Res || {_Server, Result} = Res <- Results, Result =/= pong],
         case BadResults of
             [] ->
@@ -643,6 +643,15 @@ handle_send_dump(NewPids, JoinRef, PauseRef, Dump, State) ->
             {reply, {error, ignored}, State}
     end.
 
+-spec ping_server(server_pid()) -> {server_pid(), pong | {'EXIT', term()}}.
+ping_server(Server) ->
+    {Server,
+        try
+            ping(Server)
+        catch
+            _:Reason:Stacktrace -> {'EXIT', {Reason, Stacktrace}}
+        end}.
+
 -spec handle_down(reference(), pid(), term(), state()) -> state().
 handle_down(Mon, Pid, Reason, State = #{pause_monitors := Mons}) ->
     case lists:member(Mon, Mons) of
@@ -939,7 +948,7 @@ call_user_handle_down(RemotePid, #{tab := Tab, opts := Opts, is_leader := IsLead
                 remote_node => node(RemotePid)
             },
             %% Errors would be logged inside run_tracked
-            catch cets_long:run_tracked(Info, FF);
+            cets_long:run_ignore(fun() -> cets_long:run_tracked(Info, FF) end);
         _ ->
             ok
     end.
@@ -948,7 +957,7 @@ call_user_handle_down(RemotePid, #{tab := Tab, opts := Opts, is_leader := IsLead
 handle_wrong_leader(Op, From, #{opts := #{handle_wrong_leader := F}}) ->
     %% It is used for debugging/logging
     %% Do not do anything heavy here
-    catch F(#{from => From, op => Op, server => self()}),
+    cets_long:run_ignore(fun() -> F(#{from => From, op => Op, server => self()}) end),
     ok;
 handle_wrong_leader(_Op, _From, _State) ->
     ok.

src/cets_join.erl

@@ -149,9 +149,12 @@ join2(Info, LocalPid, RemotePid, JoinOpts) ->
     after
         checkpoint(before_unpause, JoinOpts),
         %% If unpause fails, there would be log messages
-        lists:foreach(fun({Pid, Ref}) -> catch cets:unpause(Pid, Ref) end, Paused)
+        lists:foreach(fun({Pid, Ref}) -> try_unpause(Pid, Ref) end, Paused)
     end.
 
+try_unpause(Pid, Ref) ->
+    cets_long:run_ignore(fun() -> cets:unpause(Pid, Ref) end).
+
 -spec pause_servers(AllPids :: [pid(), ...]) -> Paused :: [{pid(), cets:pause_monitor()}].
 pause_servers(AllPids) ->
     %% We should create a pause helper process on each node in the cluster.
@@ -175,7 +178,10 @@ pause_on_remote_node(JoinerPid, AllPids) ->
         %% Ignore pids on the current node
         %% (because we only interested in internode connections here).
         %% Catching because we can ignore losing some connections here.
-        _Pauses = [catch cets:pause(Pid) || Pid <- AllPids, node(Pid) =/= MyNode],
+        [
+            cets_long:run_ignore(fun() -> cets:pause(Pid) end)
+         || Pid <- AllPids, node(Pid) =/= MyNode
+        ],
         Self ! {ready, self()},
         receive
             {'DOWN', JoinerMon, process, JoinerPid, _Reason} ->
@@ -194,7 +200,12 @@ send_dump(Pid, Paused, Pids, JoinRef, Dump, JoinOpts) ->
     PauseRef = proplists:get_value(Pid, Paused),
     checkpoint({before_send_dump, Pid}, JoinOpts),
     %% Error reporting would be done by cets_long:call_tracked
-    Result = catch cets:send_dump(Pid, Pids, JoinRef, PauseRef, Dump),
+    Result =
+        try
+            cets:send_dump(Pid, Pids, JoinRef, PauseRef, Dump)
+        catch
+            _:Reason -> {error, Reason}
+        end,
     checkpoint({after_send_dump, Pid, Result}, JoinOpts),
     ok.
 

src/cets_long.erl

@@ -1,6 +1,6 @@
 %% @doc Helper to log long running operations.
 -module(cets_long).
--export([run_spawn/2, run_tracked/2]).
+-export([run_spawn/2, run_tracked/2, run_ignore/1]).
 
 -ifdef(TEST).
 -export([pinfo/2]).
@@ -45,6 +45,19 @@ run_spawn(Info, F) ->
             erlang:raise(Class, Reason, Stacktrace)
     end.
 
+%% @doc Runs `Fun' for its side effects, ignoring any error it raises.
+%%
+%% Used for best-effort calls where a failure is acceptable and either
+%% logged elsewhere or intentionally ignored.
+-spec run_ignore(task_fun()) -> ok.
+run_ignore(Fun) ->
+    try
+        _ = Fun(),
+        ok
+    catch
+        _:_ -> ok
+    end.
+
 %% @doc Runs function `Fun'.
 %%
 %% Logs errors.

test/cets_test_setup.erl

@@ -74,14 +74,22 @@ start_local(Name) ->
     start_local(Name, #{}).
 
 start_local(Name, Opts) ->
-    catch cets:stop(Name),
+    try
+        cets:stop(Name)
+    catch
+        _:_ -> ok
+    end,
     cets_test_wait:wait_for_name_to_be_free(node(), Name),
     {ok, Pid} = cets:start(Name, Opts),
     schedule_cleanup(Pid),
     {ok, Pid}.
 
 start(Node, Tab) ->
-    catch rpc(Node, cets, stop, [Tab]),
+    try
+        rpc(Node, cets, stop, [Tab])
+    catch
+        _:_ -> ok
+    end,
     cets_test_wait:wait_for_name_to_be_free(Node, Tab),
     {ok, Pid} = rpc(Node, cets, start, [Tab, #{}]),
     schedule_cleanup(Pid),
@@ -91,7 +99,11 @@ start_link_local(Name) ->
     start_link_local(Name, #{}).
 
 start_link_local(Name, Opts) ->
-    catch cets:stop(Name),
+    try
+        cets:stop(Name)
+    catch
+        _:_ -> ok
+    end,
     cets_test_wait:wait_for_name_to_be_free(node(), Name),
     {ok, Pid} = cets:start_link(Name, Opts),
     schedule_cleanup(Pid),
@@ -100,7 +112,11 @@ start_link_local(Name, Opts) ->
 start_disco(Node, Opts) ->
     case Opts of
         #{name := Name} ->
-            catch rpc(Node, cets, stop, [Name]),
+            try
+                rpc(Node, cets, stop, [Name])
+            catch
+                _:_ -> ok
+            end,
             cets_test_wait:wait_for_name_to_be_free(Node, Name);
         _ ->
             ok