diff --git a/src/ra_log.erl b/src/ra_log.erl index be33ab2a..224ef575 100644 --- a/src/ra_log.erl +++ b/src/ra_log.erl @@ -845,6 +845,7 @@ set_last_index(Idx, #?MODULE{cfg = Cfg, range = Range, snapshot_state = SnapState, mem_table = Mt0, + pending = Pend0, last_written_index_term = {LWIdx0, _}} = State0) -> Cur = ra_snapshot:current(SnapState), %% After set_last_index, recovery depends on the snapshot state. @@ -853,7 +854,8 @@ set_last_index(Idx, #?MODULE{cfg = Cfg, {undefined, State} when element(1, Cur) =/= Idx -> %% not found and Idx isn't equal to latest snapshot index {not_found, State}; - {_, State} when element(1, Cur) =:= Idx -> + {_, State} when is_tuple(Cur) andalso + element(1, Cur) =:= Idx -> {_, SnapTerm} = Cur, %% Idx is equal to the current snapshot {ok, Mt} = ra_log_ets:new_mem_table_please(Cfg#cfg.names, @@ -863,6 +865,7 @@ set_last_index(Idx, #?MODULE{cfg = Cfg, {ok, State#?MODULE{range = ra_range:limit(Idx + 1, Range), last_term = SnapTerm, mem_table = Mt, + pending = ra_seq:limit(Idx, Pend0), last_written_index_term = Cur}}; {Term, State1} -> LWIdx = min(Idx, LWIdx0), @@ -885,6 +888,7 @@ set_last_index(Idx, #?MODULE{cfg = Cfg, {ok, State2#?MODULE{range = ra_range:limit(Idx + 1, Range), last_term = Term, mem_table = Mt, + pending = ra_seq:limit(Idx, Pend0), last_written_index_term = {LWIdx, LWTerm}}} end. diff --git a/src/ra_server.erl b/src/ra_server.erl index 13978915..1c6c7492 100644 --- a/src/ra_server.erl +++ b/src/ra_server.erl @@ -631,14 +631,13 @@ handle_leader({PeerId, #append_entries_reply{success = false, {Peer0#{match_index => PeerLastIdx, next_index => PeerLastIdx + 1}, Log1}; {EntryTerm, Log1} -> - % last_index has a different term or entry does not - % exist + % last_index has a different term % The peer must have uncommitted entries from a prior - % term that have now been overwritten by the current + % term that should now be overwritten by the current % leader. % Decrement next_index but don't go lower than % match index. - NextIndex = max(min(NI-1, PeerNextIdx), MI), + NextIndex = max(min(NI-1, PeerLastIdx), MI + 1), ?DEBUG("~ts: leader received last_index ~b" " from ~tw with term ~b " "- expected term ~b. Setting " diff --git a/src/ra_server_proc.erl b/src/ra_server_proc.erl index b92bcd99..9493ff1e 100644 --- a/src/ra_server_proc.erl +++ b/src/ra_server_proc.erl @@ -2146,6 +2146,9 @@ send_snapshots(Id, Term, {_, ToNode} = To, ChunkSize, "ra_log_snapshot_state for ~tw", [LogId, To]), ok; + {UID, SnapIdx, _SmallestIdx, []} -> + %% no live indexes, skip this phase + ok; {UID, SnapIdx, _SmallestIdx, LiveIndexes} -> %% first send the init phase try gen_statem:call(To, RPC, diff --git a/test/ra_log_2_SUITE.erl b/test/ra_log_2_SUITE.erl index bf84f47c..52d947ee 100644 --- a/test/ra_log_2_SUITE.erl +++ b/test/ra_log_2_SUITE.erl @@ -27,6 +27,7 @@ all_tests() -> handle_overwrite, handle_overwrite_append, follower_aer_stale_last_written, + snapshot_install_then_overwrite_stale_last_written, receive_segment, delete_during_segment_flush, read_one, @@ -38,6 +39,7 @@ all_tests() -> last_written_overwrite_2, last_index_reset, set_last_index_with_pending, + set_last_index_limits_pending_before_new_write, fold_after_sparse_mem_table, fold_after_sparse_segments, write_sparse_re_init, @@ -208,6 +210,60 @@ follower_aer_stale_last_written(Config) -> ok. +snapshot_install_then_overwrite_stale_last_written(Config) -> + %% Reproduces the metadata-store production incident. + %% + %% A follower installs a snapshot, then receives and writes an + %% *uncommitted* tail replicated by the leader. A new leader then + %% overwrites (truncates) that tail in a higher term. The follower must + %% pull last_written_index_term back to the truncation point. If it does + %% not, it reports a stale, higher last_written in its + %% append_entries_reply, the leader over-advances that peer's match_index + %% and prematurely commits entries that are not persisted on a durable + %% majority. + Log0 = ra_log_init(Config), + + %% install a snapshot at index 10, term 1 (received from the leader) + SnapIdx = 10, + SnapTerm = 1, + Meta = meta(SnapIdx, SnapTerm, [?N1]), + Chunk = create_snapshot_chunk(Config, Meta, #{}), + SnapState0 = ra_log:snapshot_state(Log0), + {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, Log1, _} = ra_log:install_snapshot({SnapIdx, SnapTerm}, ?MODULE, + LiveIndexes, + ra_log:set_snapshot_state(SnapState, + Log0)), + ?assertEqual({SnapIdx, SnapTerm}, ra_log:last_written(Log1)), + + %% leader replicates an uncommitted tail 11..15 in term 2 + {ok, Log2} = ra_log:write([{11, 2, <<"e">>}, + {12, 2, <<"e">>}, + {13, 2, <<"e">>}, + {14, 2, <<"e">>}, + {15, 2, <<"e">>}], Log1), + %% process the WAL write completion so last_written advances to 15 + Log3 = receive + {ra_log_event, {written, 2, _} = Evt} -> + element(1, ra_log:handle_event(Evt, Log2)) + after 2000 -> + flush(), + exit(written_timeout) + end, + ?assertEqual({15, 2}, ra_log:last_written(Log3)), + + %% a new leader in term 3 overwrites from index 13 -> log truncation + {ok, Log4} = ra_log:write([{13, 3, <<"new">>}], Log3), + + %% last_written must be pulled back to the last surviving entry (12, 2), + %% NOT left stale at (15, 2) + ?assertEqual({12, 2}, ra_log:last_written(Log4)), + ok. + handle_overwrite(Config) -> Log0 = ra_log_init(Config), {ok, Log1} = ra_log:write([{1, 1, "value"}, @@ -961,6 +1017,64 @@ set_last_index_with_pending(Config) -> ok. +%% set_last_index/2 rewinds the log without draining any writes that +%% are still pending WAL confirmation, and without limiting `pending` +%% to the new last index. The stale, higher indexes left behind in +%% `pending` still point at real entries in the mem table (the old +%% table is kept around as `prev` of the new one), so if the WAL later +%% asks the log to resend from one of those indexes, ra_log:resend_from/2 +%% happily resends the *discarded, truncated* data back into the WAL - +%% resurrecting log entries the reset was supposed to get rid of. +%% This is delivered as a genuine {ra_log_event, {resend_write, _}} event, +%% processed the same way ra_server_proc would process it - i.e. *after* +%% set_last_index, not as part of a new write. +set_last_index_limits_pending_before_new_write(Config) -> + Log0 = ra_log_init(Config), + %% write 1..10 without delivering any log events so 6..10 remain + %% in `pending`, unconfirmed by the WAL + Log1 = write_n(1, 11, 1, Log0), + 10 = ra_log:next_index(Log1) - 1, + {10, 1} = ra_log:last_index_term(Log1), + %% reset last_index back to 5 while 6..10 are still pending + {ok, Log2} = ra_log:set_last_index(5, Log1), + 6 = ra_log:next_index(Log2), + {5, 1} = ra_log:last_index_term(Log2), + %% drain the original batch's written confirmation(s) first - this + %% is unrelated to the bug under test and already resolves to {5, 1} + %% regardless of it, since fetch_term/2 is guarded by the (correctly + %% truncated) range + Log3 = assert_log_events(Log2, fun (L) -> + {5, 1} == ra_log:last_written(L) + end), + %% simulate the WAL asking the log to resend from 6, exactly as + %% ra_server_proc would process it after set_last_index and with no + %% new write in between + {Log4, []} = ra_log:handle_event({resend_write, 6}, Log3), + 6 = ra_log:next_index(Log4), + {5, 1} = ra_log:last_index_term(Log4), + %% without limiting pending, indexes 6..10 (term 1) are still + %% considered pending and get resent to the WAL, which will confirm + %% them with a written event - even though they were truncated away + %% and should never have been written again. With pending correctly + %% limited there is nothing left to resend, so no such event arrives. + Resurrected = receive + {ra_log_event, {written, _, _}} = Evt -> + Evt + after 200 -> + no_event + end, + ?assertEqual(no_event, Resurrected), + %% a genuinely new write at the reset point must still work + {ok, Log5} = ra_log:write([{6, 2, new6}], Log4), + Pred = fun (L) -> + {6, 2} == ra_log:last_written(L) + end, + Log6 = assert_log_events(Log5, Pred), + {6, 2} = ra_log:last_written(Log6), + {6, 2} = ra_log:last_index_term(Log6), + 7 = ra_log:next_index(Log6), + ok. + fold_after_sparse_mem_table(Config) -> Log0 = ra_log_init(Config), Log1 = write_n(1, 6, 1, Log0),