wip

Author: kjnilsson

src/ra.erl

@@ -140,6 +140,7 @@ start(Params) when is_list(Params) ->
     [ok = application:set_env(ra, Param, Value)
      || {Param, Value} <- Params],
     Res = application:ensure_all_started(ra),
+    ra_env:configure_logger(logger),
     _ = ra_system:start_default(),
     Res.
 

src/ra_kv.erl

@@ -18,6 +18,7 @@
 -export([
          start_cluster/3,
          add_member/3,
+         member_overview/1,
 
          put/4,
          get/3,
@@ -62,7 +63,8 @@ start_cluster(System, ClusterName, #{members := ServerIds})
                    #{id => Id,
                      uid => UId,
                      cluster_name => ClusterName,
-                     log_init_args => #{uid => UId},
+                     log_init_args => #{uid => UId,
+                                        min_snapshot_interval => 0},
                      initial_members => ServerIds,
                      machine => Machine}
                end || Id <- ServerIds],
@@ -75,13 +77,22 @@ add_member(System, {Name, _} = Id, LeaderId) ->
     Config = #{id => Id,
                uid => UId,
                cluster_name => Name,
-               log_init_args => #{uid => UId},
+               log_init_args => #{uid => UId,
+                                  min_snapshot_interval => 0},
                initial_members => Members,
                machine => Machine},
     ok = ra:start_server(System, Config),
     {ok, _, _} =  ra:add_member(LeaderId, Id),
     ok.
 
+member_overview(ServerId) ->
+    case ra:member_overview(ServerId) of
+        {ok, O, _} ->
+            maps:with([log, machine], O);
+        Err ->
+            Err
+    end.
+
 
 %% client
 -spec put(ra:server_id(), key(), value(), non_neg_integer()) ->

src/ra_log.erl

@@ -725,9 +725,7 @@ set_last_index(Idx, #?MODULE{cfg = Cfg,
     {state(), [effect()]}.
 handle_event({written, Term, WrittenSeq},
              #?MODULE{cfg = Cfg,
-                      % range = Range,
                       snapshot_state = SnapState,
-                      % first_index = FirstIdx,
                       pending = Pend0} = State0) ->
     CurSnap = ra_snapshot:current(SnapState),
     %% gap detection
@@ -825,7 +823,7 @@ handle_event(major_compaction, #?MODULE{reader = Reader0,
                                         snapshot_state = SS} = State) ->
     case ra_snapshot:current(SS) of
         {SnapIdx, _} ->
-            Effs = ra_log_segments:schedule_compaction(major,SnapIdx,
+            Effs = ra_log_segments:schedule_compaction(major, SnapIdx,
                                                        LiveIndexes, Reader0),
             {State, Effs};
         _ ->
@@ -841,6 +839,7 @@ handle_event({snapshot_written, {SnapIdx, _} = Snap, LiveIndexes, SnapKind},
                       snapshot_state = SnapState0} = State0)
 %% only update snapshot if it is newer than the last snapshot
   when SnapIdx >= FstIdx ->
+    % ?assert(ra_snapshot:pending(SnapState0) =/= undefined),
     SnapState1 = ra_snapshot:complete_snapshot(Snap, SnapKind, LiveIndexes,
                                                SnapState0),
     case SnapKind of

src/ra_log_segments.erl

@@ -697,7 +697,7 @@ make_links(Dir, To, From)
          _ = prim_file:delete(SymFn),
          %% make a symlink from the compacted target segment to a new .link
          %% where the compacted indexes now can be found
-         ok = prim_file:make_symlink(To, SymFn),
+         ok = prim_file:make_symlink(filename:basename(To), SymFn),
          %% rename to link to replace original segment
          ok = prim_file:rename(SymFn, SegFn)
      end || FromFn <- From],

src/ra_log_snapshot_state.erl

@@ -28,7 +28,7 @@ smallest(Table, UId) when is_binary(UId) ->
     ets:lookup_element(Table, UId, 3, 0).
 
 -spec live_indexes(ets:table(), ra:uid()) ->
-    ra:index().
+    ra_seq:state().
 live_indexes(Table, UId) when is_binary(UId) ->
     ets:lookup_element(Table, UId, 4, []).
 

src/ra_snapshot.erl

@@ -362,7 +362,7 @@ last_index_for(UId) ->
     end.
 
 -spec begin_snapshot(meta(), MacModule :: module(),
-                     MacStateb :: term(), kind(), state()) ->
+                     MacState :: term(), kind(), state()) ->
     {state(), [effect()]}.
 begin_snapshot(#{index := Idx, term := Term} = Meta, MacMod, MacState, SnapKind,
                #?MODULE{module = Mod,

test/ra_log_2_SUITE.erl

@@ -50,6 +50,7 @@ all_tests() ->
      snapshot_written_after_installation,
      oldcheckpoints_deleted_after_snapshot_install,
      append_after_snapshot_installation,
+     release_cursor_after_snapshot_installation,
      written_event_after_snapshot_installation,
      update_release_cursor,
      update_release_cursor_with_machine_version,
@@ -1285,6 +1286,57 @@ append_after_snapshot_installation(Config) ->
     {[_, _], _} = ra_log_take(16, 17, Log),
     ok.
 
+release_cursor_after_snapshot_installation(Config) ->
+    Log0 = ra_log_init(Config, #{min_snapshot_interval => 0}),
+    {0, 0} = ra_log:last_index_term(Log0),
+    Log1 = assert_log_events(write_n(1, 16, 2, Log0),
+                             fun (L) ->
+                                     LW = ra_log:last_written(L),
+                                     {15, 2} == LW
+                             end),
+
+    Log2 = Log1,
+
+    %% create snapshot chunk
+    Meta = meta(15, 2, [?N1]),
+    Chunk = create_snapshot_chunk(Config, Meta, [1, 5, 10], #{}),
+    SnapState0 = ra_log:snapshot_state(Log2),
+    {ok, SnapState1} = ra_snapshot:begin_accept(Meta, SnapState0),
+    Machine = {machine, ?MODULE, #{}},
+    {SnapState, _, LiveIndexes, AEffs} = ra_snapshot:complete_accept(Chunk, 1, Machine,
+                                                                     SnapState1),
+    run_effs(AEffs),
+    {ok, Log3, Effs4} = ra_log:install_snapshot({15, 2}, ?MODULE, LiveIndexes,
+                                                ra_log:set_snapshot_state(SnapState, Log2)),
+
+    run_effs(Effs4),
+    {15, 2} = ra_snapshot:current(ra_log:snapshot_state(Log3)),
+
+    %% Write some entries
+    Log4 = assert_log_events(write_n(16, 20, 2, Log3),
+                             fun (L) ->
+                                     LW = ra_log:last_written(L),
+                                     {19, 2} == LW
+                             end),
+
+    %% then take a snapshot
+    {Log5, Effs5} = ra_log:update_release_cursor(19, #{?N1 => new_peer(),
+                                                       ?N2 => new_peer()},
+                                                 ?MODULE, [1, 5, 10, 17], Log4),
+
+    run_effs(Effs5),
+    %% ensure snapshot index has been updated and 1 segment deleted
+    Log = assert_log_events(Log5,
+                             fun (L) ->
+                                     {19, 2} == ra_log:snapshot_index_term(L)
+                                     % andalso
+                                     % length(find_segments(Config)) == 1
+                             end),
+
+    ct:pal("Log ~p", [Log]),
+
+    ok.
+
 written_event_after_snapshot_installation(Config) ->
     logger:set_primary_config(level, all),
     %% simulates scenario where a server receives a written event from the wal