wip

Author: kjnilsson

src/ra_log.erl

@@ -859,14 +859,20 @@ handle_event({segments, TidRanges, NewSegs},
           end,
     {State, [{bg_work, Fun, fun (_Err) -> ok end}]};
 handle_event({compaction_result, Result},
-             #?MODULE{reader = Reader0} = State) ->
+             #?MODULE{cfg = #cfg{log_id = LogId},
+                      reader = Reader0} = State) ->
+            ?DEBUG("~ts: compaction result ~p", [LogId, Result]),
     {Reader, Effs} = ra_log_segments:handle_compaction_result(Result, Reader0),
     {State#?MODULE{reader = Reader}, Effs};
-handle_event(major_compaction, #?MODULE{reader = Reader0,
+handle_event(major_compaction, #?MODULE{cfg = #cfg{log_id = LogId},
+                                        reader = Reader0,
                                         live_indexes = LiveIndexes,
                                         snapshot_state = SS} = State) ->
     case ra_snapshot:current(SS) of
         {SnapIdx, _} ->
+            ?DEBUG("~ts: ra_log: major_compaction requested at snapshot index ~b, "
+                   "~b live indexes",
+                   [LogId, SnapIdx, ra_seq:length(LiveIndexes)]),
             {Reader, Effs} = ra_log_segments:schedule_compaction(major, SnapIdx,
                                                                  LiveIndexes, Reader0),
             {State#?MODULE{reader = Reader}, Effs};
@@ -1159,6 +1165,7 @@ 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]),

src/ra_mt.erl

@@ -390,6 +390,12 @@ info(#?MODULE{tid = Tid,
       size => ets:info(Tid, size),
       name => ets:info(Tid, name),
       range => range(State),
+      previous => case Prev of
+                      undefined ->
+                          undefined;
+                      _ ->
+                          info(Prev)
+                  end,
       has_previous => Prev =/= undefined
      }.
 

src/ra_server.erl

@@ -1640,17 +1640,18 @@ handle_receive_snapshot(#install_snapshot_rpc{term = Term,
                                                               OldMeta,
                                                               OldMacState),
             State1 = update_term(Term,
-                                State0#{cfg => Cfg,
-                                        log => Log,
-                                        commit_index => SnapIndex,
-                                        last_applied => SnapIndex,
-                                        %% this may not be the actual cluster index
-                                        cluster_index_term => {SnapIndex,
-                                                               SnapTerm},
-                                        cluster => make_cluster(Id, ClusterIds),
-                                        membership =>
-                                            get_membership(ClusterIds, State0),
-                                        machine_state => MacState}),
+                                 State0#{cfg => Cfg,
+                                         log => Log,
+                                         commit_index => SnapIndex,
+                                         last_applied => SnapIndex,
+                                         %% this may not be the actual
+                                         %% cluster index
+                                         cluster_index_term => {SnapIndex,
+                                                                SnapTerm},
+                                         cluster => make_cluster(Id, ClusterIds),
+                                         membership =>
+                                             get_membership(ClusterIds, State0),
+                                         machine_state => MacState}),
             State = maps:remove(snapshot_phase, State1),
             put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_APPLIED, SnapIndex),
             %% it was the last snapshot chunk so we can revert back to

test/ra_log_2_SUITE.erl

@@ -1362,7 +1362,6 @@ snapshot_installation_with_live_indexes(Config) ->
                                      LW = ra_log:last_written(L),
                                      {9, 2} == LW
                              end),
-
     Log2 = Log1,
 
     %% create snapshot chunk
@@ -1373,10 +1372,10 @@ snapshot_installation_with_live_indexes(Config) ->
     Machine = {machine, ?MODULE, #{}},
 
     %% write  a sparse one
+    {ok, Log2b} = ra_log:write_sparse({14, 2, <<>>}, 9, Log2),
     {SnapState, _, LiveIndexes, AEffs} = ra_snapshot:complete_accept(Chunk, 1, Machine,
                                                                      SnapState1),
     run_effs(AEffs),
-    {ok, Log2b} = ra_log:write_sparse({14, 2, <<>>}, 9, Log2),
     {ok, Log3, Effs4} = ra_log:install_snapshot({15, 2}, ?MODULE, LiveIndexes,
                                                 ra_log:set_snapshot_state(SnapState, Log2b)),
 
@@ -1388,12 +1387,20 @@ snapshot_installation_with_live_indexes(Config) ->
     %% write the next index, bearning in mind the last index the WAL saw
     %% was 14
     {ok, Log4} = ra_log:write([{16, 2, <<>>}], Log3),
-    Log = assert_log_events(Log4,
-                             fun (L) ->
-                                     LW = ra_log:last_written(L),
-                                     {16, 2} == LW
-                             end),
+    Log5 = assert_log_events(Log4,
+                            fun (L) ->
+                                    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)]),
+    flush(),
     ok.
 
 snapshot_installation(Config) ->

test/ra_server_SUITE.erl

@@ -2312,7 +2312,7 @@ follower_installs_snapshot_with_pre(_Config) ->
 
     %% now send a pre message
     ISRpcPre = ISRpcInit#install_snapshot_rpc{chunk_state = {0, pre},
-                                           data = [{2, 1, <<"e1">>}]},
+                                              data = [{2, 1, <<"e1">>}]},
     {receive_snapshot, State3, [{reply, _}]} =
         ra_server:handle_receive_snapshot(ISRpcPre, State2),