snapshot api changes

Author: kjnilsson

src/ra.hrl

@@ -169,7 +169,7 @@
         {term :: ra_term(), % the leader's term
          leader_id :: ra_server_id(),
          meta :: snapshot_meta(),
-         chunk_state = {0, pre} :: {pos_integer(), chunk_flag()},
+         chunk_state = {0, pre} :: {non_neg_integer(), chunk_flag()},
          data :: term()
         }).
 

src/ra_log.erl

@@ -33,7 +33,7 @@
          next_index/1,
          snapshot_state/1,
          set_snapshot_state/2,
-         install_snapshot/3,
+         install_snapshot/4,
          recover_snapshot/1,
          snapshot_index_term/1,
          update_release_cursor/5,
@@ -60,7 +60,6 @@
 -define(MIN_CHECKPOINT_INTERVAL, 16384).
 -define(LOG_APPEND_TIMEOUT, 5000).
 -define(WAL_RESEND_TIMEOUT, 5000).
--define(ETSTBL, ra_log_snapshot_state).
 
 -type ra_meta_key() :: atom().
 -type segment_ref() :: {ra_range:range(), File :: file:filename_all()}.
@@ -926,9 +925,9 @@ snapshot_state(State) ->
 set_snapshot_state(SnapState, State) ->
     State#?MODULE{snapshot_state = SnapState}.
 
--spec install_snapshot(ra_idxterm(), module(), state()) ->
-    {ra_snapshot:meta(), MacState :: term(), state(), effects()}.
-install_snapshot({SnapIdx, SnapTerm} = IdxTerm, MacMod,
+-spec install_snapshot(ra_idxterm(), module(), ra_seq:state(), state()) ->
+    {ok, state(), effects()}.
+install_snapshot({SnapIdx, SnapTerm} = IdxTerm, MacMod, LiveIndexes,
                  #?MODULE{cfg = #cfg{uid = UId,
                                      names = Names} = Cfg,
                           snapshot_state = SnapState0,
@@ -951,30 +950,18 @@ install_snapshot({SnapIdx, SnapTerm} = IdxTerm, MacMod,
                            last_index = SnapIdx,
                            last_term = SnapTerm,
                            last_written_index_term = IdxTerm},
-    {Meta, MacState} = recover_snapshot(State),
-    LiveIndexes = ra_machine:live_indexes(MacMod, MacState),
-    %% TODO: it is not safe to write the indexes _after_ if we then treat
-    %% the persisted indexes as authoritative as if we crash in between
-    %% it may compact segments that still contain live indexes
-    SmallestLiveIndex = ra_seq:first(LiveIndexes),
-    SnapDir = ra_snapshot:current_snapshot_dir(SnapState),
-    ok = ra_snapshot:write_indexes(SnapDir, LiveIndexes),
-    %% TODO: more mt entries could potentially be cleared up here
+    %% TODO: more mt entries could potentially be cleared up in the
+    %% mem table here
+    SmallestLiveIndex = case ra_seq:first(LiveIndexes) of
+                            undefined ->
+                                SnapIdx + 1;
+                            I ->
+                                I
+                        end,
     {Spec, Mt} = ra_mt:set_first(SmallestLiveIndex, Mt0),
     ok = exec_mem_table_delete(Names, UId, Spec),
-    %% TODO: move this to install_snapshot so we can work out the
-    %% live indexes
-    SmallestIdx = case LiveIndexes of
-                      [] ->
-                          SnapIdx + 1;
-                      _ ->
-                          ra_seq:first(LiveIndexes)
-                  end,
-    ok = ra_log_snapshot_state:insert(?ETSTBL, UId, SnapIdx, SmallestIdx,
-                                      LiveIndexes),
-    {Meta, MacState, State#?MODULE{live_indexes = LiveIndexes,
-                                   mem_table = Mt
-                                   },
+    {ok, State#?MODULE{live_indexes = LiveIndexes,
+                       mem_table = Mt},
      CompEffs ++ CPEffects}.
 
 

src/ra_server.erl

@@ -1516,9 +1516,10 @@ handle_follower(Msg, State) ->
 handle_receive_snapshot(#install_snapshot_rpc{term = Term,
                                               meta = #{index := SnapIndex,
                                                        machine_version := SnapMacVer,
+                                                       cluster := ClusterIds,
                                                        term := SnapTerm} = SnapMeta,
                                               chunk_state = {Num, ChunkFlag},
-                                              data = Data},
+                                              data = ChunkOrEntries},
                         #{cfg := #cfg{id = Id,
                                       log_id = LogId,
                                       effective_machine_version = CurEffMacVer,
@@ -1536,7 +1537,7 @@ handle_receive_snapshot(#install_snapshot_rpc{term = Term,
                                      last_term = SnapTerm,
                                      last_index = SnapIndex},
     case ChunkFlag of
-        pre ->
+        pre when is_list(ChunkOrEntries) ->
             %% TODO: we may need to reset the log here to
             %% the last applied index as we
             %% dont know for sure indexes after last applied
@@ -1546,22 +1547,20 @@ handle_receive_snapshot(#install_snapshot_rpc{term = Term,
                        fun ({I, _, _} = E, {L0, LstIdx}) ->
                                {ok, L} = ra_log:write_sparse(E, LstIdx, L0),
                                {L, I}
-                       end, {Log00, LastIndex}, Data),
+                       end, {Log00, LastIndex}, ChunkOrEntries),
             State = update_term(Term, State0#{log => Log0}),
             {receive_snapshot, State, [{reply, Reply}]};
         next ->
             SnapState0 = ra_log:snapshot_state(Log00),
-            {ok, SnapState, Effs0} =
-                ra_snapshot:accept_chunk(Data, Num, ChunkFlag,
-                                         SnapState0),
+            SnapState = ra_snapshot:accept_chunk(ChunkOrEntries, Num, SnapState0),
             Log0 = ra_log:set_snapshot_state(SnapState, Log00),
             State = update_term(Term, State0#{log => Log0}),
-            {receive_snapshot, State, [{reply, Reply} | Effs0]};
+            {receive_snapshot, State, [{reply, Reply}]};
         last ->
             SnapState0 = ra_log:snapshot_state(Log00),
-            {ok, SnapState, Effs0} =
-                ra_snapshot:accept_chunk(Data, Num, ChunkFlag,
-                                         SnapState0),
+            {SnapState, MacState, LiveIndexes, Effs0} =
+                ra_snapshot:complete_accept(ChunkOrEntries, Num, Machine,
+                                            SnapState0),
             Log0 = ra_log:set_snapshot_state(SnapState, Log00),
             %% if the machine version of the snapshot is higher
             %% we also need to update the current effective machine configuration
@@ -1579,9 +1578,9 @@ handle_receive_snapshot(#install_snapshot_rpc{term = Term,
                           Cfg0
                   end,
             %% this is the last chunk so we can "install" it
-            {#{cluster := ClusterIds},
-             MacState, Log, Effs} = ra_log:install_snapshot({SnapIndex, SnapTerm},
-                                                            EffMacMod, Log0),
+            {ok, Log, Effs} = ra_log:install_snapshot({SnapIndex, SnapTerm},
+                                                      EffMacMod,
+                                                      LiveIndexes, Log0),
             OldServerIds = maps:map(fun (_, V) ->
                                             maps:with([voter_status], V)
                                     end, Cluster),

src/ra_server_proc.erl

@@ -1912,6 +1912,8 @@ send_snapshots(Id, Term, {_, ToNode} = To, ChunkSize,
         ra_snapshot:begin_read(SnapState, Context),
 
     %% only send the snapshot if the target server can accept it
+    %% TODO: grab the last_applied index also and use this to floor
+    %% the live indexes
     TheirMacVer = erpc:call(ToNode, ra_machine, version, [Machine]),
 
     %% rpc the check what their

src/ra_snapshot.erl

@@ -8,6 +8,7 @@
 
 -include("ra.hrl").
 
+-include_lib("stdlib/include/assert.hrl").
 -type file_err() :: ra_lib:file_err().
 
 %% alias
@@ -33,7 +34,8 @@
          complete_snapshot/4,
 
          begin_accept/2,
-         accept_chunk/4,
+         accept_chunk/3,
+         complete_accept/4,
          abort_accept/1,
 
          context/2,
@@ -70,6 +72,7 @@
 -record(accept, {%% the next expected chunk
                  next = 1 :: non_neg_integer(),
                  state :: term(),
+                 machine_version :: non_neg_integer(),
                  idxterm :: ra_idxterm()}).
 
 -record(?MODULE,
@@ -503,26 +506,30 @@ complete_snapshot(IdxTerm, checkpoint, _LiveIndexes,
 
 -spec begin_accept(meta(), state()) ->
     {ok, state()}.
-begin_accept(#{index := Idx, term := Term} = Meta,
+begin_accept(#{index := Idx,
+               machine_version := SnapMacVer,
+               term := Term} = Meta,
              #?MODULE{module = Mod,
                       snapshot_directory = Dir} = State) ->
     SnapDir = make_snapshot_dir(Dir, Idx, Term),
     ok = ra_lib:make_dir(SnapDir),
     {ok, AcceptState} = Mod:begin_accept(SnapDir, Meta),
     {ok, State#?MODULE{accepting = #accept{idxterm = {Idx, Term},
+                                           machine_version = SnapMacVer,
                                            state = AcceptState}}}.
 
--spec accept_chunk(term(), non_neg_integer(), chunk_flag(), state()) ->
-    {ok, state(), [effect()]}.
-accept_chunk(Chunk, Num, last,
-             #?MODULE{uid = _UId,
-                      module = Mod,
-                      snapshot_directory = Dir,
-                      current = Current,
-                      pending = Pending,
-                      accepting = #accept{next = Num,
-                                          idxterm = {_Idx, _} = IdxTerm,
-                                          state = AccState}} = State) ->
+-spec complete_accept(Chunk :: term(), Num :: non_neg_integer(),
+                      Machine :: ra_machine:machine(), state()) ->
+    {state(), MacState :: term(), ra_seq:state(), [effect()]}.
+complete_accept(Chunk, Num, Machine,
+                #?MODULE{uid = UId,
+                         module = Mod,
+                         snapshot_directory = Dir,
+                         current = Current,
+                         pending = Pending,
+                         accepting = #accept{next = Num,
+                                             idxterm = {Idx, Term} = IdxTerm,
+                                             state = AccState}} = State0) ->
     %% last chunk
     ok = Mod:complete_accept(Chunk, AccState),
     %% run validate here?
@@ -536,25 +543,44 @@ accept_chunk(Chunk, Num, last,
     Eff = {bg_work,
            fun() -> [delete(Dir, Del) || Del <- Dels] end,
            fun (_) -> ok end},
+    State = State0#?MODULE{accepting = undefined,
+                           %% reset any pending snapshot writes
+                           pending = undefined,
+                           current = IdxTerm},
+    {ok, #{machine_version := SnapMacVer}, MacState} = recover(State),
+    SnapMacMod  = ra_machine:which_module(Machine, SnapMacVer),
+    LiveIndexes = ra_machine:live_indexes(SnapMacMod, MacState),
+    SnapDir = make_snapshot_dir(Dir, Idx, Term),
+    ok = write_indexes(SnapDir, LiveIndexes),
+    %% delete accepting marker file
+    AcceptMarker = filename:join(SnapDir, <<"accepting">>),
+    _ = prim_file:delete(AcceptMarker),
+    %% assert accepting marker is no longer there
+    ?assertNot(filelib:is_file(AcceptMarker)),
+    SmallestIdx = case ra_seq:first(LiveIndexes) of
+                      undefined ->
+                          Idx + 1;
+                      I ->
+                          I
+                  end,
+    ok = ra_log_snapshot_state:insert(?ETSTBL, UId, Idx, SmallestIdx,
+                                      LiveIndexes),
+    {State, MacState, LiveIndexes, [Eff]}.
 
-    {ok, State#?MODULE{accepting = undefined,
-                       %% reset any pending snapshot writes
-                       pending = undefined,
-                       current = IdxTerm}, [Eff]};
-accept_chunk(Chunk, Num, next,
-             #?MODULE{module = Mod,
-                      accepting =
-                      #accept{state = AccState0,
-                              next = Num} = Accept} = State) ->
+-spec accept_chunk(Chunk :: term(), Num :: non_neg_integer(), state()) ->
+    state().
+accept_chunk(Chunk, Num, #?MODULE{module = Mod,
+                                  accepting =
+                                  #accept{state = AccState0,
+                                          next = Num} = Accept} = State) ->
     {ok, AccState} = Mod:accept_chunk(Chunk, AccState0),
-    {ok, State#?MODULE{accepting = Accept#accept{state = AccState,
-                                                 next = Num + 1}},
-     []};
-accept_chunk(_Chunk, Num, _ChunkFlag,
+    State#?MODULE{accepting = Accept#accept{state = AccState,
+                                            next = Num + 1}};
+accept_chunk(_Chunk, Num,
              #?MODULE{accepting = #accept{next = Next}} = State)
   when Next > Num ->
     %% this must be a resend - we can just ignore it
-    {ok, State, []}.
+    State.
 
 -spec abort_accept(state()) -> state().
 abort_accept(#?MODULE{accepting = undefined} = State) ->

test/ra_kv_SUITE.erl

@@ -73,40 +73,56 @@ basics(_Config) ->
     ?assertMatch({ok, #{machine := #{num_keys := 2}}, KvId},
                  ra:member_overview(KvId)),
     ra_log_wal:force_roll_over(ra_log_wal),
-    timer:sleep(1000),
+    %% wait for rollover processing
+    ra_log_wal:last_writer_seq(ra_log_wal, <<>>),
+    %% wait for segment writer to process
+    ra_log_segment_writer:await(ra_log_segment_writer),
+    %% promt ra_kv to take a snapshot
     ok = ra:aux_command(KvId, take_snapshot),
-    timer:sleep(1000),
+    %% wait for snapshot to complete
+    ok = ra_lib:retry(
+           fun () ->
+                   {ok, #{log := #{snapshot_index := SnapIdx,
+                                   num_segments := NumSegments,
+                                   last_index := LastIdx}}, _} =
+                       ra:member_overview(KvId),
+                   SnapIdx == LastIdx andalso NumSegments == 2
+           end, 100, 100),
+    %% restart server to test recovery
     ok = ra:stop_server(default, KvId),
     ok = ra:restart_server(default, KvId),
     {ok, #{index := LastIdx}} = ra_kv:put(KvId, <<"k3">>, <<"k3">>, 5000),
-    ct:pal("overview after ~p", [ra:member_overview(KvId)]),
-    {ok, #{machine := #{live_indexes := Live}}, _} = ra:member_overview(KvId),
+    {ok, #{machine := #{live_indexes := Live},
+           log := #{last_index := KvIdLastIdx}}, _} = ra:member_overview(KvId),
     {ok, {Reads, _}} = ra_server_proc:read_entries(KvId, [LastIdx | Live],
                                                    undefined, 1000),
-    ct:pal("ReadRes ~p", [Reads]),
-
-    % debugger:start(),
-    % int:i(ra_log),
-    % int:i(ra_snapshot),
-    % int:i(ra_server_proc),
-    % int:break(ra_server_proc, 1922),
-    % int:break(ra_log, 873),
-    % int:break(ra_log, 1002),
-    % int:break(ra_log, 1328),
+    ?assertEqual(3, map_size(Reads)),
+    % ct:pal("ReadRes ~p", [Reads]),
     KvId2 = {kv2, node()},
     ok = ra_kv:add_member(?SYS, KvId2, KvId),
-    timer:sleep(1000),
+    ok = ra_lib:retry(
+           fun () ->
+                   {ok, #{log := #{last_index := Last}}, _} =
+                       ra:member_overview(KvId2),
+                   Last >= KvIdLastIdx
+           end, 100, 100),
     {ok, {Reads2, _}} = ra_server_proc:read_entries(KvId2, [LastIdx | Live],
                                                     undefined, 1000),
-    ct:pal("ReadRes2 ~p", [Reads2]),
-    ct:pal("overview ~p", [ra:member_overview(KvId2)]),
     ?assertEqual(3, map_size(Reads2)),
     ra_log_wal:force_roll_over(ra_log_wal),
-    timer:sleep(1000),
+    ra_log_wal:last_writer_seq(ra_log_wal, <<>>),
+    ra_log_segment_writer:await(ra_log_segment_writer),
     {ok, {Reads3, _}} = ra_server_proc:read_entries(KvId2, [LastIdx | Live],
                                                     undefined, 1000),
     ct:pal("ReadRes3 ~p", [Reads3]),
-    ct:pal("overview3 ~p", [ra:member_overview(KvId2)]),
+    % ct:pal("overview3 ~p", [ra:member_overview(KvId2)]),
     ?assertEqual(3, map_size(Reads3)),
 
+    %% TODO: test recovery of kv
+    ok = ra:stop_server(default, KvId2),
+    ok = ra:restart_server(default, KvId2),
+    {ok, {Reads4, _}} = ra_server_proc:read_entries(KvId2, [LastIdx | Live],
+                                                    undefined, 1000),
+
+    ?assertEqual(3, map_size(Reads4)),
     ok.