Reset log after aborted snapshot installation.

kjnilsson · kjnilsson · commit 3e03e18ff57b · 2025-07-21T16:38:30.000+01:00
diff --git a/src/ra_kv_harness.erl b/src/ra_kv_harness.erl
@@ -181,6 +181,8 @@ start_single_peer_node(NodeId, Opts) ->
             erpc:call(NodeName, logger, set_primary_config, [level, warning]),
             erpc:call(NodeName, application, set_env, [sasl, sasl_error_logger, false]),
             erpc:call(NodeName, application, stop, [sasl]),
+            erpc:call(NodeName, application, set_env,
+                      [kernel, prevent_overlapping_partitions, false]),
             Dir = filename:join(BaseDir, NodeName),
             {ok, _} = erpc:call(NodeName, ra, start_in, [Dir]),
             % Set logger level to reduce verbosity on peer node
diff --git a/src/ra_log.erl b/src/ra_log.erl
@@ -732,6 +732,7 @@ set_last_index(Idx, #?MODULE{cfg = Cfg,
                              mem_table = Mt0,
                              last_written_index_term = {LWIdx0, _}} = State0) ->
     Cur = ra_snapshot:current(SnapState),
+    %% TODO: can a log recover to the right reset point? I doubt it
     case fetch_term(Idx, State0) of
         {undefined, State} when element(1, Cur) =/= Idx ->
             %% not found and Idx isn't equal to latest snapshot index
@@ -745,7 +746,7 @@ set_last_index(Idx, #?MODULE{cfg = Cfg,
             put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_WRITTEN_INDEX, Idx),
             {ok, State#?MODULE{range = ra_range:limit(Idx + 1, Range),
                                last_term = SnapTerm,
-                                mem_table = Mt,
+                               mem_table = Mt,
                                last_written_index_term = Cur}};
         {Term, State1} ->
             LWIdx = min(Idx, LWIdx0),
@@ -947,8 +948,8 @@ handle_event({snapshot_written, {SnapIdx, _} = Snap, LiveIndexes, SnapKind},
                                    current_snapshot = Snap,
                                    snapshot_state = SnapState},
             {Reader, CompEffs} = ra_log_segments:schedule_compaction(minor, SnapIdx,
-                                                                    LiveIndexes,
-                                                                    State#?MODULE.reader),
+                                                                     LiveIndexes,
+                                                                     State#?MODULE.reader),
             Effects = CompEffs ++ Effects0,
             {State#?MODULE{reader = Reader}, Effects};
         checkpoint ->
@@ -1165,7 +1166,6 @@ assert(#?MODULE{cfg = #cfg{log_id = LogId},
                 current_snapshot = CurrSnap,
                 live_indexes = LiveIndexes
                } = State) ->
-    ra_log_segments:range/1
     %% TODO: remove this at some point?
     ?DEBUG("~ts: ra_log: asserting Range ~p Snapshot ~p",
            [LogId, Range, CurrSnap]),
diff --git a/src/ra_server.erl b/src/ra_server.erl
@@ -1571,11 +1571,8 @@ handle_receive_snapshot(#install_snapshot_rpc{term = Term,
             %% the snapshot sending must have been interrupted and restarted
             %% during the init or pre-phase
             %% abort the snapshot, and revert to follower
-            SnapState0 = ra_log:snapshot_state(Log00),
-            SnapState = ra_snapshot:abort_accept(SnapState0),
-            Log = ra_log:set_snapshot_state(SnapState, Log00),
-            {follower, maps:remove(snapshot_phase, State0#{log => Log}),
-             [{next_event, Rpc}]};
+            State = abort_receive(State0),
+            {follower, State, [{next_event, Rpc}]};
         pre when is_list(ChunkOrEntries) ->
             [{_FstIdx, _, _} | _] = ChunkOrEntries,
             % ?DEBUG("~ts: receiving snapshot chunk pre first index ~b snap index ~b, term ~b",
@@ -1664,71 +1661,50 @@ handle_receive_snapshot(#install_snapshot_rpc{term = Term,
     end;
 handle_receive_snapshot(#append_entries_rpc{term = Term} = Msg,
                         #{current_term := CurTerm,
-                          cfg := #cfg{log_id = LogId},
-                          log := Log0} = State)
+                          cfg := #cfg{log_id = LogId}} = State0)
   when Term > CurTerm ->
     ?INFO("~ts: follower receiving snapshot saw append_entries_rpc from ~w for term ~b "
           "abdicates term: ~b!",
           [LogId, Msg#append_entries_rpc.leader_id,
            Term, CurTerm]),
-    SnapState0 = ra_log:snapshot_state(Log0),
-    SnapState = ra_snapshot:abort_accept(SnapState0),
-    Log = ra_log:set_snapshot_state(SnapState, Log0),
-    {follower, maps:remove(snapshot_phase,
-                           update_term(Term,
-                                       clear_leader_id(State#{log => Log}))),
-     [{next_event, Msg}]};
+    State = abort_receive(State0),
+    {follower, update_term(Term, State), [{next_event, Msg}]};
 handle_receive_snapshot({ra_log_event, Evt},
                         #{cfg := #cfg{log_id = _LogId},
                           log := Log0} = State) ->
-    % ?DEBUG("~ts: ~s ra_log_event received: ~w",
-    %       [LogId, ?FUNCTION_NAME, Evt]),
     % simply forward all other events to ra_log
     % whilst the snapshot is being received
     {Log, Effects} = ra_log:handle_event(Evt, Log0),
     {receive_snapshot, State#{log => Log}, Effects};
 handle_receive_snapshot(receive_snapshot_timeout,
-                        #{cfg := #cfg{log_id = LogId},
-                          log := Log0} = State) ->
+                        #{cfg := #cfg{log_id = LogId}} = State0) ->
     ?INFO("~ts: ~s receive snapshot timed out.",
           [LogId, ?FUNCTION_NAME]),
-    SnapState0 = ra_log:snapshot_state(Log0),
-    SnapState = ra_snapshot:abort_accept(SnapState0),
-    Log = ra_log:set_snapshot_state(SnapState, Log0),
-    {follower, maps:remove(snapshot_phase, State#{log => Log}), []};
+    State = abort_receive(State0),
+    {follower, State, []};
 handle_receive_snapshot(#info_rpc{term = Term} = Msg,
                         #{current_term := CurTerm,
-                          cfg := #cfg{log_id = LogId},
-                          log := Log0} = State)
+                          cfg := #cfg{log_id = LogId}} = State0)
   when CurTerm < Term ->
     ?INFO("~ts: follower receiving snapshot saw info_rpc from ~w for term ~b "
-          "abdicates term: ~b!",
+          "current term: ~b!",
           [LogId, Msg#info_rpc.from,
            Term, CurTerm]),
-    SnapState0 = ra_log:snapshot_state(Log0),
-    SnapState = ra_snapshot:abort_accept(SnapState0),
-    Log = ra_log:set_snapshot_state(SnapState, Log0),
-    {follower, maps:remove(snapshot_phase,
-                           update_term(Term, clear_leader_id(State#{log => Log}))),
-     [{next_event, Msg}]};
+    State = abort_receive(State0),
+    {follower, update_term(Term, State), [{next_event, Msg}]};
 handle_receive_snapshot(#info_rpc{} = InfoRpc, State) ->
     InfoReplyEffect = empty_info_reply_effect(State, InfoRpc),
     {receive_snapshot, State, [InfoReplyEffect]};
 handle_receive_snapshot(#info_reply{term = Term} = Msg,
                         #{current_term := CurTerm,
-                          cfg := #cfg{log_id = LogId},
-                          log := Log0} = State)
+                          cfg := #cfg{log_id = LogId}} = State0)
   when CurTerm < Term ->
     ?INFO("~ts: follower receiving snapshot saw info_reply from ~w for term ~b "
           "abdicates term: ~b!",
           [LogId, Msg#info_reply.from,
            Term, CurTerm]),
-    SnapState0 = ra_log:snapshot_state(Log0),
-    SnapState = ra_snapshot:abort_accept(SnapState0),
-    Log = ra_log:set_snapshot_state(SnapState, Log0),
-    {follower, maps:remove(snapshot_phase,
-                           update_term(Term, clear_leader_id(State#{log => Log}))),
-     [{next_event, Msg}]};
+    State = abort_receive(State0),
+    {follower, update_term(Term, State), [{next_event, Msg}]};
 handle_receive_snapshot(#info_reply{}, State) ->
     {receive_snapshot, State, []};
 handle_receive_snapshot(Msg, State) ->
@@ -1737,6 +1713,21 @@ handle_receive_snapshot(Msg, State) ->
     %% TODO: work out what else to handle
     {receive_snapshot, State, [{reply, {error, {unsupported_call, Msg}}}]}.
 
+abort_receive(#{snapshot_phase := Phase,
+                last_applied := LastApplied,
+                log := Log0} = State) ->
+    SnapState0 = ra_log:snapshot_state(Log0),
+    SnapState = ra_snapshot:abort_accept(SnapState0),
+    Log1 = ra_log:set_snapshot_state(SnapState, Log0),
+    Log = case Phase of
+              pre ->
+                  {ok, Log2} = ra_log:set_last_index(LastApplied, Log1),
+                  Log2;
+              _ ->
+                  Log1
+          end,
+    clear_leader_id(maps:remove(snapshot_phase, State#{log => Log})).
+
 -spec handle_await_condition(ra_msg(), ra_server_state()) ->
     {ra_state(), ra_server_state(), effects()}.
 handle_await_condition(#request_vote_rpc{} = Msg, State) ->
diff --git a/src/ra_server_proc.erl b/src/ra_server_proc.erl
@@ -938,6 +938,14 @@ receive_snapshot(EventType, {aux_command, Cmd}, State0) ->
         ?HANDLE_EFFECTS(Effects, EventType,
                         State0#state{server_state = ServerState}),
     {keep_state, State, Actions};
+receive_snapshot(info, {'DOWN', MRef, process, _Pid, _Info},
+         #state{leader_monitor = MRef} = State) ->
+    %% leader is down
+    ?INFO("~ts: receive_snapshot - Leader monitor down. Aborting snapshot receive. "
+          "Entering follower state.",
+          [log_id(State)]),
+    receive_snapshot(info, receive_snapshot_timeout,
+                     State#state{leader_monitor = undefined});
 receive_snapshot(EventType, Msg, State0) ->
     case handle_receive_snapshot(Msg, State0) of
         {receive_snapshot, State1, Effects} ->
diff --git a/src/ra_snapshot.erl b/src/ra_snapshot.erl
@@ -204,9 +204,18 @@ find_snapshots(#?MODULE{uid = UId,
             State;
         Current0 ->
             Current = filename:join(SnapshotsDir, Current0),
-            {ok, #{index := Idx, term := Term}} = Module:read_meta(Current),
-            %% TODO: recover live indexes and record that
-            ok = ra_log_snapshot_state:insert(?ETSTBL, UId, Idx, Idx+1, []),
+            {ok, #{index := Idx,
+                   term := Term}} = Module:read_meta(Current),
+            %% recover live indexes and record that
+            {ok, Indexes} = indexes(Current),
+            SmallestLiveIdx = case ra_seq:first(Indexes) of
+                                  undefined ->
+                                      Idx+1;
+                                  First ->
+                                      First
+                              end,
+            ok = ra_log_snapshot_state:insert(?ETSTBL, UId, Idx, SmallestLiveIdx,
+                                              Indexes),
 
             ok = delete_snapshots(SnapshotsDir, lists:delete(Current0, Snaps)),
             %% delete old snapshots if any
@@ -494,7 +503,7 @@ complete_snapshot({Idx, _} = IdxTerm, snapshot, LiveIndexes,
                       I ->
                           I
                   end,
-    %% TODO live indexes
+    %% live indexes
     ok = ra_log_snapshot_state:insert(?ETSTBL, UId, Idx, SmallestIdx,
                                       LiveIndexes),
     State#?MODULE{pending = undefined,
diff --git a/test/ra_log_2_SUITE.erl b/test/ra_log_2_SUITE.erl
@@ -1381,7 +1381,6 @@ snapshot_installation_with_live_indexes(Config) ->
 
 
     run_effs(Effs4),
-    ct:pal("o ~p", [ra_log:overview(Log3)]),
     {15, _} = ra_log:last_index_term(Log3),
     {15, _} = ra_log:last_written(Log3),
     %% write the next index, bearning in mind the last index the WAL saw
@@ -1392,15 +1391,22 @@ snapshot_installation_with_live_indexes(Config) ->
                                     LW = ra_log:last_written(L),
                                     {16, 2} == LW
                             end),
-    ct:pal("o ~p", [ra_log:overview(Log5)]),
     ra_log_wal:force_roll_over(ra_log_wal),
     Log = assert_log_events(Log5,
                             fun (L) ->
                                     #{mem_table_range := R} = ra_log:overview(L),
                                     R == undefined
                             end),
     ct:pal("o ~p", [ra_log:overview(Log)]),
+    UId = ?config(uid, Config),
+    ?assertEqual(LiveIndexes, ra_log_snapshot_state:live_indexes(
+                                ra_log_snapshot_state, UId)),
+    ra_log:close(Log),
     flush(),
+    _LogAfter = ra_log_init(Config),
+    %% validate recovery recovers the live indexes correctly
+    ?assertEqual(LiveIndexes, ra_log_snapshot_state:live_indexes(
+                                ra_log_snapshot_state, UId)),
     ok.
 
 snapshot_installation(Config) ->
diff --git a/test/ra_log_memory.erl b/test/ra_log_memory.erl
@@ -284,7 +284,7 @@ write_meta_f(_Key, _Value, State) ->
 can_write(_Log) ->
     true.
 
-overview(Log) ->
+overview(#state{} = Log) ->
     #{type => ?MODULE,
       last_index => Log#state.last_index,
       last_written => Log#state.last_written,
diff --git a/test/ra_server_SUITE.erl b/test/ra_server_SUITE.erl
@@ -65,6 +65,7 @@ all() ->
      leader_receives_install_snapshot_rpc,
      follower_installs_snapshot,
      follower_installs_snapshot_with_pre,
+     follower_aborts_snapshot_with_pre,
      follower_ignores_installs_snapshot_with_higher_machine_version,
      follower_receives_stale_snapshot,
      follower_receives_snapshot_lower_than_last_applied,
@@ -182,6 +183,7 @@ setup_log() ->
                                       ra_log_memory:fold(A, B, C, D, E)
                               end),
     meck:expect(ra_log, release_resources, fun ra_log_memory:release_resources/3),
+    meck:expect(ra_log, overview, fun ra_log_memory:overview/1),
     meck:expect(ra_log, append_sync,
                 fun({Idx, Term, _} = E, L0) ->
                         L1 = ra_log_memory:append(E, L0),
@@ -2339,6 +2341,42 @@ follower_installs_snapshot_with_pre(_Config) ->
         ra_server:handle_receive_snapshot(ISRpc1, State3),
     ok.
 
+follower_aborts_snapshot_with_pre(_Config) ->
+    N1 = ?N1, N2 = ?N2, N3 = ?N3,
+    #{N3 := {_, State = #{cluster := Config}, _}} =
+        init_servers([N1, N2, N3], {module, ra_queue, #{}}),
+    LastTerm = 1, % snapshot term
+    Term = 2, % leader term
+    Idx = 3,
+    ISRpcInit = #install_snapshot_rpc{term = Term, leader_id = N1,
+                                      meta = snap_meta(Idx, LastTerm, Config),
+                                      chunk_state = {0, init},
+                                      data = []},
+    %% the init message starts the process
+    {receive_snapshot, State1,
+     [{next_event, ISRpc}, {record_leader_msg, _}]} =
+        ra_server:handle_follower(ISRpcInit, State#{current_term => Term}),
+
+    %% actually process init message in the correct state
+    {receive_snapshot, State2, [{reply, _}]} =
+        ra_server:handle_receive_snapshot(ISRpc, State1),
+
+    %% now send a pre message
+    ISRpcPre = ISRpcInit#install_snapshot_rpc{chunk_state = {0, pre},
+                                              data = [{2, 1, <<"e1">>}]},
+    {receive_snapshot, State3, [{reply, _}]} =
+        ra_server:handle_receive_snapshot(ISRpcPre, State2),
+    ?assertMatch(#{log := #{last_index := 2}},
+                 ra_server:overview(State3)),
+
+    {follower, State3b, []} =
+        ra_server:handle_receive_snapshot(receive_snapshot_timeout, State3),
+    ?assertNot(maps:is_key(snapshot_phase, State3b)),
+    %% assert the aborted install reset the log
+    ?assertMatch(#{log := #{last_index := 0}},
+                 ra_server:overview(State3b)),
+    ok.
+
 follower_ignores_installs_snapshot_with_higher_machine_version(_Config) ->
     %% currently followers cannot correctly handle snapshots with a higher
     %% machine version so have to ignore them
