Fix ra_log:commit/1 to roll back mt changes.

Else subsequent writes would crash the entire process.

Author: kjnilsson

src/ra_log.erl

@@ -388,9 +388,12 @@ commit_tx(#?MODULE{cfg = #cfg{uid = UId,
                               wal = Wal} = Cfg,
                    tx = {true, TxRange},
                    range = Range,
+                   pending = Pend0,
                    mem_table = Mt1} = State) ->
-    {Entries, Mt} = ra_mt:commit(Mt1),
-    Tid = ra_mt:tid(Mt),
+    %% TODO: staged could contain entries from previous? I don't think that is
+    %% ever the case as that would mean overwriting withing a single append batch
+    Entries = ra_mt:staged(Mt1),
+    Tid = ra_mt:tid(Mt1),
     WriterId = {UId, self()},
     PrevIdx = previous_wal_index(State),
     {WalCommands, Num, _} =
@@ -403,15 +406,21 @@ commit_tx(#?MODULE{cfg = #cfg{uid = UId,
 
     case ra_log_wal:write_batch(Wal, lists:reverse(WalCommands)) of
         {ok, Pid} ->
+            %% commit after send to WAL, else abort
+            {_, Mt} = ra_mt:commit(Mt1),
             ok = incr_counter(Cfg, ?C_RA_LOG_WRITE_OPS, Num),
             {ok, State#?MODULE{tx = false,
                                range = ra_range:add(TxRange, Range),
                                last_wal_write = {Pid, now_ms(), LastIdx},
                                mem_table = Mt}};
         {error, wal_down} ->
+            {Idx, _, _} = hd(Entries),
+            Mt = ra_mt:abort(Mt1),
             %% TODO: review this -  still need to return the state here
-            {error, wal_down, State#?MODULE{tx = false,
-                                            mem_table = Mt}}
+            {error, wal_down,
+             State#?MODULE{tx = false,
+                           pending = ra_seq:limit(Idx - 1, Pend0),
+                           mem_table = Mt}}
     end;
 commit_tx(#?MODULE{tx = false} = State) ->
     State.
@@ -815,7 +824,7 @@ handle_event({written, Term, WrittenSeq},
                     {State#?MODULE{last_written_index_term = {LastWrittenIdx, Term},
                                    pending = Pend}, []};
                 {error, not_prefix} ->
-                    ?DEBUG("~ts: ~p not prefix of ~p",
+                    ?DEBUG("~ts: ~w not prefix of ~w",
                            [Cfg#cfg.log_id, WrittenSeq, Pend0]),
                     {resend_pending(State0), []}
             end;

src/ra_log_wal.erl

@@ -543,18 +543,28 @@ incr_batch(#batch{num_writes = Writes,
                   #{Pid := #batch_writer{term = TERM,
                                          tid = MT_TID,
                                          seq = Seq0} = W} ->
-                      %% The Tid and term is the same so add to current batch_writer
-                      Seq = ra_seq:append(Idx, Seq0),
+                      %% The Tid and term is the same so add to
+                      %% current batch_writer
+                      Seq = case Idx > ra_seq:last(Seq0) of
+                                true ->
+                                    ra_seq:append(Idx, Seq0);
+                                false ->
+                                    %% this is a rewrite / resend
+                                    %% we need to limit the seq before
+                                    %% appending
+                                    ra_seq:append(Idx, ra_seq:limit(Idx - 1,
+                                                                    Seq0))
+                            end,
                       Waiting0#{Pid => W#batch_writer{seq = Seq,
                                                       smallest_live_idx = SmallestLiveIdx,
                                                       term = Term}};
                   _ ->
-                      %% The tid is different, open a new batch writer for the
-                      %% new tid and term
+                      %% The tid or term is different
+                      %% open a new batch writer for the new tid and term
                       PrevBatchWriter = maps:get(Pid, Waiting0, undefined),
                       Writer = #batch_writer{smallest_live_idx = SmallestLiveIdx,
                                              tid = MtTid,
-                                             seq = [Idx],
+                                             seq = ra_seq:append(Idx, []),
                                              uid = UId,
                                              term = Term,
                                              old = PrevBatchWriter},
@@ -968,9 +978,9 @@ should_roll_wal(#state{conf = #conf{max_entries = MaxEntries},
 smallest_live_index(#conf{ra_log_snapshot_state_tid = Tid}, ServerUId) ->
     ra_log_snapshot_state:smallest(Tid, ServerUId).
 
-update_ranges(Ranges, UId, MtTid, _SmallestIdx, AddSeq) ->
+update_ranges(Ranges, UId, MtTid = MT_TID, _SmallestIdx, AddSeq) ->
     case Ranges of
-        #{UId := [{MtTid, Seq0} | Seqs]} ->
+        #{UId := [{MT_TID, Seq0} | Seqs]} ->
             %% limit the old range by the add end start as in some resend
             %% cases we may have got back before the prior range.
             Seq = ra_seq:add(AddSeq, Seq0),

src/ra_mt.erl

@@ -16,7 +16,7 @@
          insert_sparse/3,
          stage/2,
          commit/1,
-         % abort/1,
+         abort/1,
          lookup/2,
          lookup_term/2,
          tid_for/3,
@@ -182,6 +182,19 @@ stage({Idx, _, _} = _Entry,
             exit({unexpected_sparse_stage, Idx, Seq})
     end.
 
+-spec staged(state()) -> [log_entry()].
+staged(#?MODULE{staged = undefined}) ->
+    [];
+staged(#?MODULE{staged = {_, Staged0},
+                prev = Prev0}) ->
+    PrevStaged = case Prev0 of
+                     undefined ->
+                         [];
+                     _ ->
+                         staged(Prev0)
+                 end,
+    PrevStaged ++ lists:reverse(Staged0).
+
 -spec commit(state()) -> {[log_entry()], state()}.
 commit(#?MODULE{staged = undefined} = State) ->
     {[], State};
@@ -200,6 +213,16 @@ commit(#?MODULE{tid = Tid,
     {PrevStaged ++ Staged, State#?MODULE{staged = undefined,
                                          prev = Prev}}.
 
+-spec abort(state()) -> state().
+abort(#?MODULE{staged = undefined} = State) ->
+    State;
+abort(#?MODULE{indexes = Seq,
+               staged = {_, Staged0}} = State) ->
+    {Idx, _, _} = lists:last(Staged0),
+    State#?MODULE{staged = undefined,
+                  indexes = ra_seq:limit(Idx - 1, Seq)}.
+
+
 -spec lookup(ra:index(), state()) ->
     log_entry() | undefined.
 lookup(Idx, #?MODULE{staged = {FstStagedIdx, Staged}})
@@ -375,11 +398,11 @@ range(_State) ->
 tid(#?MODULE{tid = Tid}) ->
     Tid.
 
--spec staged(state()) -> [log_entry()].
-staged(#?MODULE{staged = {_, Staged}}) ->
-    Staged;
-staged(#?MODULE{staged = undefined}) ->
-    [].
+% -spec staged(state()) -> [log_entry()].
+% staged(#?MODULE{staged = {_, Staged}}) ->
+%     Staged;
+% staged(#?MODULE{staged = undefined}) ->
+%     [].
 
 -spec is_active(ets:tid(), state()) -> boolean().
 is_active(Tid, State) ->

src/ra_seq.erl

@@ -93,6 +93,8 @@ limit(_CeilIdxIncl, Seq) ->
 
 %% @doc adds two sequences together where To is
 %% the "lower" sequence
+%% TODO: optimise to avoid the fold which could be expensive
+%% for very large sequences containing large ranges
 -spec add(Add :: state(), To :: state()) -> state().
 add([], To) ->
     To;

src/ra_server.erl

@@ -1709,6 +1709,9 @@ handle_receive_snapshot(#install_snapshot_rpc{term = Term,
             put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_APPLIED, SnapIndex),
             %% it was the last snapshot chunk so we can revert back to
             %% follower status
+            %% TODO: consider waiting until all pending WAL writes (sparse)
+            %% have been confirmed written by the WAL or segment writer until
+            %% returning to follower state
             {follower, persist_last_applied(State), [{reply, Reply} |
                                                      Effs0 ++ Effs ++
                                                      SnapInstalledEffs]}

test/ra_log_2_SUITE.erl

@@ -22,6 +22,7 @@ all_tests() ->
      resend_write_lost_in_wal_crash,
      resend_after_written_event_lost_in_wal_crash,
      resend_write_after_tick,
+     snapshot_before_written,
      handle_overwrite,
      handle_overwrite_append,
      receive_segment,
@@ -52,7 +53,7 @@ all_tests() ->
      wal_down_read_availability,
      wal_down_append_throws,
      wal_down_write_returns_error_wal_down,
-
+     wal_down_stage,
      detect_lost_written_range,
      snapshot_installation,
      snapshot_written_after_installation,
@@ -141,6 +142,39 @@ end_per_testcase(_, Config) ->
 -define(N2, {n2, node()}).
 % -define(N3, {n3, node()}).
 
+snapshot_before_written(Config) ->
+    %% snapshot_written comes in before written event for lower entries.
+
+    Log0 = write_n(1, 6, 1,
+                   ra_log_init(Config, #{min_snapshot_interval => 0})),
+    receive
+        {ra_log_event, {written, 1, _}} -> ok
+    after 2000 ->
+              exit(written_timeout)
+    end,
+    %% write 10 entries
+    Log1 = write_n(6, 20, 1, Log0),
+    SnapIdx = 10,
+    %% do snapshot in
+    {Log2, Effs} = ra_log:update_release_cursor(SnapIdx, #{}, macctx(),
+                                                [10, 4], Log1),
+    run_effs(Effs),
+    {Log3, Effs3} = receive
+                        {ra_log_event, {snapshot_written, {10, 1}, _,
+                                        snapshot, _} = Evt} ->
+                            ra_log:handle_event(Evt, Log2)
+                    after 5000 ->
+                              flush(),
+                              exit(snapshot_written_timeout)
+                    end,
+    run_effs(Effs3),
+
+    _Log4 = assert_log_events(Log3,
+                              fun (L) ->
+                                      {19, 1} == ra_log:last_written(L)
+                              end),
+    ok.
+
 handle_overwrite(Config) ->
     Log0 = ra_log_init(Config),
     {ok, Log1} = ra_log:write([{1, 1, "value"},
@@ -154,9 +188,11 @@ handle_overwrite(Config) ->
     % ensure immediate truncation
     {1, 2} = ra_log:last_index_term(Log3),
     {ok, Log4} = ra_log:write([{2, 2, "value"}], Log3),
+
     % simulate the first written event coming after index 20 has already
     % been written in a new term
     {Log, _} = ra_log:handle_event({written, 1, [2, 1]}, Log4),
+    ct:pal("Log ~p", [Log]),
     % ensure last written has not been incremented
     {0, 0} = ra_log:last_written(Log),
     {2, 2} = ra_log:last_written(
@@ -1253,6 +1289,27 @@ wal_down_write_returns_error_wal_down(Config) ->
     {error, wal_down} = ra_log:write([{1, 1, hi}], Log0),
     ok.
 
+wal_down_stage(Config) ->
+    [SupPid] = [P || {ra_log_wal_sup, P, _, _}
+                     <- supervisor:which_children(ra_log_sup)],
+
+    Log0 = ra_log:begin_tx(ra_log_init(Config)),
+    Log1 = ra_log:append({2, 1, ho},
+                         ra_log:append({1, 1, hi}, Log0)),
+    ct:pal("o1 ~p", [ra_log:overview(Log1)]),
+    ok = supervisor:terminate_child(SupPid, ra_log_wal),
+    {error, wal_down, Log2} = ra_log:commit_tx(Log1),
+    {ok, _} = supervisor:restart_child(SupPid, ra_log_wal),
+    %% check it is possible to write the next  entries
+    ?assertEqual(1, ra_log:next_index(Log2)),
+    Log3 = ra_log:append({1, 1, hi}, ra_log:begin_tx(Log2)),
+    {ok, Log4} = ra_log:commit_tx(Log3),
+    ra_log_wal:force_roll_over(ra_log_wal),
+    ct:pal("o4 ~p", [ra_log:overview(Log4)]),
+    Log = deliver_all_log_events(Log4, 200),
+    ct:pal("o ~p", [ra_log:overview(Log)]),
+    ok.
+
 detect_lost_written_range(Config) ->
     Log0 = ra_log_init(Config, #{wal => ra_log_wal}),
     {0, 0} = ra_log:last_index_term(Log0),

test/ra_log_wal_SUITE.erl

@@ -24,6 +24,7 @@ all() ->
 all_tests() ->
     [
      basic_log_writes,
+     rewrite,
      sparse_writes,
      sparse_write_same_batch,
      sparse_write_overwrite,
@@ -156,6 +157,22 @@ basic_log_writes(Config) ->
     meck:unload(),
     ok.
 
+rewrite(Config) ->
+    meck:new(ra_log_segment_writer, [passthrough]),
+    meck:expect(ra_log_segment_writer, await, fun(_) -> ok end),
+    Conf = ?config(wal_conf, Config),
+    {_UId, _} = WriterId = ?config(writer_id, Config),
+    Tid = ets:new(?FUNCTION_NAME, []),
+    {ok, Pid} = ra_log_wal:start_link(Conf),
+    suspend_process(Pid),
+    {ok, _} = ra_log_wal:write(Pid, WriterId, Tid, 12, 1, "value"),
+    {ok, _} = ra_log_wal:write(Pid, WriterId, Tid, 12, 1, "value"),
+    erlang:resume_process(Pid),
+    flush(),
+    proc_lib:stop(Pid),
+    meck:unload(),
+    ok.
+
 sparse_writes(Config) ->
     meck:new(ra_log_segment_writer, [passthrough]),
     meck:expect(ra_log_segment_writer, await, fun(_) -> ok end),

test/ra_mt_SUITE.erl

@@ -36,6 +36,7 @@ all_tests() ->
      successor,
      successor_below,
      stage_commit,
+     stage_abort,
      range_overlap,
      stage_commit_2,
      perf,
@@ -464,6 +465,31 @@ stage_commit(_Config) ->
     || I <- lists:seq(1, 10)],
     ok.
 
+stage_abort(_Config) ->
+    Tid = ets:new(t1, [set, public]),
+    {ok, Mt0} = ra_mt:insert({0, 0, hi}, ra_mt:init(Tid)),
+    ?assertMatch({0, 0}, ra_mt:range(Mt0)),
+    Mt1 = lists:foldl(
+            fun (I, Acc) ->
+                    element(2, ra_mt:stage({I, 1, <<"banana">>}, Acc))
+            end, Mt0, lists:seq(1, 10)),
+    ?assertMatch({0, 10}, ra_mt:range(Mt1)),
+    [{I, _, _} = ra_mt:lookup(I, Mt1)
+    || I <- lists:seq(1, 10)],
+    Mt2 = ra_mt:abort(Mt1),
+    ?assertMatch({0, 0}, ra_mt:range(Mt2)),
+    Mt3 = lists:foldl(
+            fun (I, Acc) ->
+                    element(2, ra_mt:stage({I, 1, <<"banana">>}, Acc))
+            end, Mt2, lists:seq(1, 10)),
+    {Entries, Mt4}= ra_mt:commit(Mt3),
+    ?assertMatch({0, 10}, ra_mt:range(Mt4)),
+    ?assertEqual(10, length(Entries)),
+    ?assertMatch([{1, 1, _} | _], Entries),
+    [{I, _, _} = ra_mt:lookup(I, Mt4)
+    || I <- lists:seq(1, 10)],
+    ok.
+
 range_overlap(_Config) ->
     Tid = ets:new(t1, [set, public]),
     Mt0 = ra_mt:init(Tid),