Fix multiple issues around interrupted snapshot replication

with sparse entries

Author: kjnilsson

src/ra.erl

@@ -155,9 +155,10 @@ start_in(DataDir) ->
     ra_env:configure_logger(logger),
     LogFile = filename:join(DataDir, "ra.log"),
     SaslFile = filename:join(DataDir, "ra_sasl.log"),
-    logger:set_primary_config(level, debug),
+    logger:remove_handler(ra_handler),
+    ok = logger:set_primary_config(level, debug),
     Config = #{config => #{file => LogFile}},
-    logger:add_handler(ra_handler, logger_std_h, Config),
+    ok = logger:add_handler(ra_handler, logger_std_h, Config),
     application:load(sasl),
     application:set_env(sasl, sasl_error_logger, {file, SaslFile}),
     application:stop(sasl),

src/ra_log.erl

@@ -21,6 +21,7 @@
          append_sync/2,
          write_sync/2,
          fold/5,
+         fold/6,
          sparse_read/2,
          partial_read/3,
          execute_read_plan/4,
@@ -491,7 +492,7 @@ write([{Idx, _, _} | _], #?MODULE{cfg = #cfg{uid = UId},
                                       [UId, Idx, Range])),
     {error, {integrity_error, Msg}}.
 
--spec write_sparse(log_entry(), ra:index(), state()) ->
+-spec write_sparse(log_entry(), option(ra:index()), state()) ->
     {ok, state()} | {error, wal_down | gap_detected}.
 write_sparse({Idx, Term, _} = Entry, PrevIdx0,
              #?MODULE{cfg = #cfg{uid = UId,
@@ -533,11 +534,18 @@ write_sparse({Idx, Term, _} = Entry, PrevIdx0,
 -spec fold(FromIdx :: ra_index(), ToIdx :: ra_index(),
            fun((log_entry(), Acc) -> Acc), Acc, state()) ->
     {Acc, state()} when Acc :: term().
+fold(From0, To0, Fun, Acc0, State) ->
+    fold(From0, To0, Fun, Acc0, State, error).
+
+-spec fold(FromIdx :: ra_index(), ToIdx :: ra_index(),
+           fun((log_entry(), Acc) -> Acc), Acc, state(),
+            MissingKeyStrategy :: error | return) ->
+    {Acc, state()} when Acc :: term().
 fold(From0, To0, Fun, Acc0,
      #?MODULE{cfg = Cfg,
               mem_table = Mt,
               range = {StartIdx, EndIdx},
-              reader = Reader0} = State)
+              reader = Reader0} = State, MissingKeyStrat)
   when To0 >= From0 andalso
        To0 >= StartIdx ->
 
@@ -550,22 +558,27 @@ fold(From0, To0, Fun, Acc0,
     case MtOverlap of
         {undefined, {RemStart, RemEnd}} ->
             {Reader, Acc} = ra_log_segments:fold(RemStart, RemEnd, Fun,
-                                               Acc0, Reader0),
+                                                 Acc0, Reader0,
+                                                 MissingKeyStrat),
             {Acc, State#?MODULE{reader = Reader}};
         {{MtStart, MtEnd}, {RemStart, RemEnd}} ->
             {Reader, Acc1} = ra_log_segments:fold(RemStart, RemEnd, Fun,
-                                                Acc0, Reader0),
-            Acc = ra_mt:fold(MtStart, MtEnd, Fun, Acc1, Mt),
+                                                  Acc0, Reader0,
+                                                  MissingKeyStrat),
+            Acc = ra_mt:fold(MtStart, MtEnd, Fun, Acc1, Mt, MissingKeyStrat),
             NumRead = MtEnd - MtStart + 1,
             ok = incr_counter(Cfg, ?C_RA_LOG_READ_MEM_TBL, NumRead),
             {Acc, State#?MODULE{reader = Reader}};
         {{MtStart, MtEnd}, undefined} ->
-            Acc = ra_mt:fold(MtStart, MtEnd, Fun, Acc0, Mt),
+            Acc = ra_mt:fold(MtStart, MtEnd, Fun, Acc0, Mt, MissingKeyStrat),
+            %% TODO: if fold is short circuited with MissingKeyStrat == return
+            %% this count isn't correct, it doesn't massively matter so leaving
+            %% for now
             NumRead = MtEnd - MtStart + 1,
             ok = incr_counter(Cfg, ?C_RA_LOG_READ_MEM_TBL, NumRead),
             {Acc, State}
     end;
-fold(_From, _To, _Fun, Acc, State) ->
+fold(_From, _To, _Fun, Acc, State, _) ->
     {Acc, State}.
 
 %% @doc Reads a list of indexes.

src/ra_log_segment.erl

@@ -12,6 +12,7 @@
          append/4,
          sync/1,
          fold/6,
+         fold/7,
          is_modified/1,
          read_sparse/4,
          read_sparse_no_checks/4,
@@ -300,7 +301,22 @@ fold(#state{cfg = #cfg{mode = read} = Cfg,
             cache = Cache,
             index = Index},
      FromIdx, ToIdx, Fun, AccFun, Acc) ->
-    fold0(Cfg, Cache, FromIdx, ToIdx, Index, Fun, AccFun, Acc).
+    fold0(Cfg, Cache, FromIdx, ToIdx, Index, Fun, AccFun, Acc,
+          error).
+
+-spec fold(state(),
+           FromIdx :: ra_index(),
+           ToIdx :: ra_index(),
+           fun((binary()) -> term()),
+           fun(({ra_index(), ra_term(), term()}, Acc) -> Acc), Acc,
+            MissingKeyStrat :: error | return) ->
+    Acc when Acc :: term().
+fold(#state{cfg = #cfg{mode = read} = Cfg,
+            cache = Cache,
+            index = Index},
+     FromIdx, ToIdx, Fun, AccFun, Acc, MissingKeyStrat) ->
+    fold0(Cfg, Cache, FromIdx, ToIdx, Index, Fun, AccFun, Acc,
+          MissingKeyStrat).
 
 -spec is_modified(state()) -> boolean().
 is_modified(#state{cfg = #cfg{fd = Fd},
@@ -403,10 +419,10 @@ term_query(#state{index = Index}, Idx) ->
         _ -> undefined
     end.
 
-fold0(_Cfg, _Cache, Idx, FinalIdx, _, _Fun, _AccFun, Acc)
+fold0(_Cfg, _Cache, Idx, FinalIdx, _, _Fun, _AccFun, Acc, _)
   when Idx > FinalIdx ->
     Acc;
-fold0(Cfg, Cache0, Idx, FinalIdx, Index, Fun, AccFun, Acc0) ->
+fold0(Cfg, Cache0, Idx, FinalIdx, Index, Fun, AccFun, Acc0, MissingKeyStrat) ->
     case Index of
         #{Idx := {Term, Offset, Length, Crc} = IdxRec} ->
             case pread(Cfg, Cache0, Offset, Length) of
@@ -415,7 +431,8 @@ fold0(Cfg, Cache0, Idx, FinalIdx, Index, Fun, AccFun, Acc0) ->
                     case validate_checksum(Crc, Data) of
                         true ->
                             Acc = AccFun({Idx, Term, Fun(Data)}, Acc0),
-                            fold0(Cfg, Cache, Idx+1, FinalIdx, Index, Fun, AccFun, Acc);
+                            fold0(Cfg, Cache, Idx+1, FinalIdx,
+                                  Index, Fun, AccFun, Acc, MissingKeyStrat);
                         false ->
                             %% CRC check failures are irrecoverable
                             exit({ra_log_segment_crc_check_failure, Idx, IdxRec,
@@ -426,8 +443,10 @@ fold0(Cfg, Cache0, Idx, FinalIdx, Index, Fun, AccFun, Acc0) ->
                     exit({ra_log_segment_unexpected_eof, Idx, IdxRec,
                           Cfg#cfg.filename})
             end;
-        _ ->
-            exit({missing_key, Idx, Cfg#cfg.filename})
+        _ when  MissingKeyStrat == error ->
+            exit({missing_key, Idx, Cfg#cfg.filename});
+        _ when  MissingKeyStrat == return ->
+            Acc0
     end.
 
 -spec range(state()) -> option({ra_index(), ra_index()}).

src/ra_log_segments.erl

@@ -21,7 +21,7 @@
          range/1,
          num_open_segments/1,
          update_first_index/2,
-         fold/5,
+         fold/6,
          sparse_read/3,
          read_plan/2,
          exec_read_plan/6,
@@ -285,14 +285,15 @@ compaction_conf(#?STATE{cfg = #cfg{compaction_conf = Conf}}) ->
 num_open_segments(#?STATE{open_segments = Open}) ->
      ra_flru:size(Open).
 
--spec fold(ra_index(), ra_index(), fun(), term(), state()) ->
+-spec fold(ra_index(), ra_index(), fun(), term(), state(),
+                                       MissingKeyStrategy :: error | return) ->
     {state(), term()}.
 fold(FromIdx, ToIdx, Fun, Acc,
-    #?STATE{cfg = #cfg{} = Cfg} = State0)
+    #?STATE{cfg = #cfg{} = Cfg} = State0, MissingKeyStrat)
   when ToIdx >= FromIdx ->
     ok = incr_counter(Cfg, ?C_RA_LOG_READ_SEGMENT, ToIdx - FromIdx + 1),
-    segment_fold(State0, FromIdx, ToIdx, Fun, Acc);
-fold(_FromIdx, _ToIdx, _Fun, Acc, #?STATE{} = State) ->
+    segment_fold(State0, FromIdx, ToIdx, Fun, Acc, MissingKeyStrat);
+fold(_FromIdx, _ToIdx, _Fun, Acc, #?STATE{} = State, _Strat) ->
     {State, Acc}.
 
 -spec sparse_read(state(), [ra_index()], [log_entry()]) ->
@@ -453,15 +454,15 @@ segment_fold_plan(SegRefs, {_ReqStart, ReqEnd} = ReqRange, Acc) ->
 segment_fold(#?STATE{segment_refs = SegRefs,
                      open_segments = OpenSegs,
                      cfg = Cfg} = State,
-             RStart, REnd, Fun, Acc) ->
+             RStart, REnd, Fun, Acc, MissingKeyStrat) ->
     Plan = segment_fold_plan(SegRefs, {RStart, REnd}, []),
     {Op, A} =
         lists:foldl(
           fun ({Fn, {Start, End}}, {Open0, Ac0}) ->
                   {Seg, Open} = get_segment(Cfg, Open0, Fn),
                   {Open, ra_log_segment:fold(Seg, Start, End,
                                              fun binary_to_term/1,
-                                             Fun, Ac0)}
+                                             Fun, Ac0, MissingKeyStrat)}
           end, {OpenSegs, Acc}, Plan),
     {State#?MODULE{open_segments = Op}, A}.
 

src/ra_mt.erl

@@ -6,7 +6,6 @@
 %% @hidden
 -module(ra_mt).
 
--include_lib("stdlib/include/assert.hrl").
 -include("ra.hrl").
 
 -export([
@@ -22,6 +21,7 @@
          lookup_term/2,
          tid_for/3,
          fold/5,
+         fold/6,
          get_items/2,
          record_flushed/3,
          set_first/2,
@@ -249,12 +249,6 @@ lookup_term(Idx, #?MODULE{tid = Tid,
         Term ->
             Term
     end.
-  % when ?IN_RANGE(Idx, Seq) ->
-  %   ets:lookup_element(Tid, Idx, 2);
-% lookup_term(Idx, #?MODULE{prev = #?MODULE{} = Prev}) ->
-  %   lookup_term(Idx, Prev);
-% lookup_term(_Idx, _State) ->
-  %   undefined.
 
 -spec tid_for(ra:index(), ra_term(), state()) ->
     undefined | ets:tid().
@@ -269,16 +263,28 @@ tid_for(Idx, Term, State) ->
             tid_for(Idx, Term, State#?MODULE.prev)
     end.
 
+-spec fold(ra:index(), ra:index(),
+           fun(), term(), state(), MissingKeyStrategy :: error | return) ->
+    term().
+fold(From, To, Fun, Acc, State, MissingKeyStrat)
+  when is_atom(MissingKeyStrat) andalso
+       To >= From ->
+    case lookup(From, State) of
+        undefined when MissingKeyStrat == error ->
+            error({missing_key, From, Acc});
+        undefined when MissingKeyStrat == return ->
+            Acc;
+        E ->
+            fold(From + 1, To, Fun, Fun(E, Acc),
+                 State, MissingKeyStrat)
+    end;
+fold(_From, _To, _Fun, Acc, _State, _Strat) ->
+    Acc.
+
 -spec fold(ra:index(), ra:index(), fun(), term(), state()) ->
     term().
-fold(To, To, Fun, Acc, State) ->
-    E = lookup(To, State),
-    Fun(E, Acc);
-fold(From, To, Fun, Acc, State)
-  when To > From ->
-    E = lookup(From, State),
-    ?assert(E =/= undefined),
-    fold(From + 1, To, Fun, Fun(E, Acc), State).
+fold(From, To, Fun, Acc, State) ->
+    fold(From, To, Fun, Acc, State, error).
 
 -spec get_items([ra:index()], state()) ->
     {[log_entry()],

src/ra_server.erl

@@ -452,9 +452,11 @@ recover(#{cfg := #cfg{log_id = LogId,
     FromScan = CommitIndex + 1,
     {ToScan, _} = ra_log:last_index_term(Log0),
     ?DEBUG("~ts: scanning for cluster changes ~b:~b ", [LogId, FromScan, ToScan]),
+    %% if we're recovering after a partial sparse write phase this will fail
+    %%
     {State, Log} = ra_log:fold(FromScan, ToScan,
                                fun cluster_scan_fun/2,
-                               State1, Log0),
+                               State1, Log0, return),
 
     put_counter(Cfg, ?C_RA_SVR_METRIC_COMMIT_LATENCY, 0),
     State#{log => Log,
@@ -1557,8 +1559,6 @@ handle_receive_snapshot(#install_snapshot_rpc{term = Term,
                                {ok, L} = ra_log:write_sparse(E, LstIdx, L0),
                                {L, I}
                        end, {Log00, LastIdx}, ChunkOrEntries),
-            ?DEBUG("~ts: receiving snapshot log last index ~p",
-                   [LogId, ra_log:last_index_term(Log)]),
             State = update_term(Term, State0#{log => Log}),
             {receive_snapshot, State, [{reply, Reply}]};
         next ->