From 51f9810e0168dfec45b0780d293ab8e88beb3813 Mon Sep 17 00:00:00 2001 From: Karl Nilsson Date: Fri, 3 Jul 2026 10:53:18 +0100 Subject: [PATCH] Forward follower's append_entries_reply when it already has the snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A follower whose last_applied is at or ahead of an incoming snapshot's index rejects the install_snapshot_rpc with an append_entries_reply rather than an install_snapshot_result. The snapshot sender treated this as fatal (exit(unexpected_install_snapshot_result)), so the reply never reached the leader and next_index was never corrected — the leader re-sent the same snapshot indefinitely, until the next election. The sender now forwards that reply to the leader and exits normally. The leader's existing success = false handler advances next_index to the position the follower indicated, breaking the loop. Adds ra_SUITE test snapshot_installation_target_already_has_snapshot. --- src/ra_server_proc.erl | 14 ++++++++ test/ra_SUITE.erl | 79 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 93 insertions(+) diff --git a/src/ra_server_proc.erl b/src/ra_server_proc.erl index d7ef06d0..b92bcd99 100644 --- a/src/ra_server_proc.erl +++ b/src/ra_server_proc.erl @@ -2152,6 +2152,20 @@ send_snapshots(Id, Term, {_, ToNode} = To, ChunkSize, {dirty_timeout, InstallTimeout}) of #install_snapshot_result{} -> ok; + #append_entries_reply{} = AER -> + %% the target's last applied index is at or + %% ahead of the snapshot index so it rejected + %% the snapshot with an append_entries_reply. + %% Forward it to the leader so it can correct + %% next_index for this peer, then finish + %% normally rather than crashing and entering + %% a pointless snapshot retry loop. + ?DEBUG("~ts: ~tw is at or ahead of snapshot " + "index ~b, forwarding its " + "append_entries_reply to the leader", + [LogId, To, SnapIdx]), + ok = gen_statem:cast(Id, {To, AER}), + exit(normal); Unexp -> ?INFO("~ts: ~tw returned an unexpected" " install snapshot result ~w", diff --git a/test/ra_SUITE.erl b/test/ra_SUITE.erl index 275c2630..a1d201b7 100644 --- a/test/ra_SUITE.erl +++ b/test/ra_SUITE.erl @@ -63,6 +63,7 @@ all_tests() -> server_catches_up, snapshot_installation, snapshot_installation_with_call_crash, + snapshot_installation_target_already_has_snapshot, add_member, queue_example, ramp_up_and_ramp_down, @@ -1065,6 +1066,84 @@ snapshot_installation_with_call_crash(Config) -> ok. +snapshot_installation_target_already_has_snapshot(Config) -> + %% Regression test: if a leader sends a snapshot to a follower whose + %% last_applied is already at or ahead of the snapshot index, the follower + %% rejects it with an append_entries_reply (rather than an + %% install_snapshot_result). The snapshot sender process must forward that + %% reply to the leader so it can correct next_index for the peer, instead + %% of crashing with unexpected_install_snapshot_result and entering a + %% pointless snapshot retry loop. + N1 = nth_server_name(Config, 1), + N2 = nth_server_name(Config, 2), + N3 = nth_server_name(Config, 3), + Name = ?config(test_name, Config), + Servers = [N1, N2, N3], + Mac = {module, ra_queue, #{}}, + %% sentinel term that will never occur naturally so the forwarded reply + %% can be identified in the leader's call history + Sentinel = 424242, + {ok, [Leader0, _, Down], []} = ra:start_cluster(default, Name, Mac, Servers), + ok = ra:stop_server(?SYS, Down), + {ok, _, Leader} = ra:members(Leader0), + %% process enough commands to trigger a snapshot on the leader + [begin + ok = ra:pipeline_command(Leader, {enq, N}, no_correlation, normal), + ok = ra:pipeline_command(Leader, deq, no_correlation, normal) + end || N <- lists:seq(1, 2500)], + {ok, _, _} = ra:process_command(Leader, deq), + + meck:new(ra_server, [passthrough]), + %% simulate the target already being at or ahead of the snapshot index: + %% it rejects the install_snapshot_rpc with an append_entries_reply + meck:expect(ra_server, handle_follower, + fun (#install_snapshot_rpc{term = T, + meta = #{index := Idx}}, State) -> + Reply = #append_entries_reply{term = T, + success = false, + next_index = Idx + 1, + last_index = Idx, + last_term = Sentinel}, + {follower, State, [{reply, Reply}]}; + (A, B) -> + meck:passthrough([A, B]) + end), + %% wrap handle_leader so its calls are captured in the meck history + meck:expect(ra_server, handle_leader, + fun (A, B) -> meck:passthrough([A, B]) end), + + %% catchup will attempt to send a snapshot which the target rejects + ok = ra:restart_server(?SYS, Down), + + %% the rejecting reply must reach the leader's handle_leader (i.e. the + %% sender forwarded it rather than crashing) + ForwardedToLeader = + fun () -> + lists:any( + fun ({_P, {ra_server, handle_leader, + [{Peer, #append_entries_reply{last_term = LT}} | _]}, + _R}) + when Peer == Down andalso LT == Sentinel -> + true; + (_) -> + false + end, meck:history(ra_server)) + end, + ?assert(try_n_times(ForwardedToLeader, 100)), + + meck:unload(ra_server), + + %% once the target stops rejecting it installs the snapshot for real and + %% the cluster converges - i.e. the leader recovered without getting stuck + ?assert(try_n_times( + fun () -> + {ok, {N1Idx, _}, _} = ra:local_query(N1, fun ra_lib:id/1), + {ok, {N2Idx, _}, _} = ra:local_query(N2, fun ra_lib:id/1), + {ok, {N3Idx, _}, _} = ra:local_query(N3, fun ra_lib:id/1), + (N1Idx == N2Idx) and (N1Idx == N3Idx) + end, 200)), + ok. + try_n_times(_Fun, 0) -> false; try_n_times(Fun, N) ->