b0VIM 8.1X\9E"Aknilssonsnowman.local~knilsson/code/rabbitmq/ra/src/ra_server.erlutf-8 U3210#"! UtppUq;,8Db:RPJM E F, Lr e V#RyTWXveJ3T}U9&GDaOQOHLDE*KpdU!QwSVWtdI1S{T9$adC a?W P  n D  s B s E u E  ^NE o'd:z7Io$l.~D " next_index ~b for ~w", ?DEBUG("~w: setting last index to ~b, " {LastTerm, L} when LastIdx >= MI -> % entry exists we can forward next_index => NextIdx}, L}; {Peer0#{match_index => LastIdx, [Id, PeerId, NextIdx]), ?DEBUG("~w: setting next index for ~w ~b", % entry was not found - simply set next index to {undefined, L} -> {Peer, Log} = case ra_log:fetch_term(LastIdx, Log0) of % match_index and update next_index directly % if the last_index exists and has a matching term we can forward next_index := NI}} = Nodes, #{PeerId := Peer0 = #{match_index := MI, State0 = #{id := Id, cluster := Nodes, log := Log0}) -> last_term = LastTerm}}, last_index = LastIdx, next_index = NextIdx,handle_leader({PeerId, #append_entries_reply{success = false, end; {follower, update_term(Term, State0), []} [Id, Term, CurTerm]), "abdicates term: ~b!~n", ?NOTICE("~w leader saw append_entries_reply for term ~b " _ -> {leader, State0, []}; [Id, PeerId]), ?WARN("~w saw append_entries_reply from unknown peer ~w~n", undefined -> case peer(PeerId, State0) of id := Id} = State0) when Term > CurTerm -> #{current_term := CurTerm,handle_leader({PeerId, #append_entries_reply{term = Term}}, end; end {leader, State, Effects} _ -> {stop, State, Effects}; ?INFO("~s: leader not in new cluster - goodbye", [LogId]), % time to say goodbye % config has been committed % leader is not in the cluster and the new cluster w log_id := LogId} #{commit_index := CI, cluster_index_term := {CITIndex, _}, {leader, State, Effects}; % leader is in the cluster #{id := Id, cluster := #{Id := _}} -> case State of Effects = Effects0 ++ RpcEffects, end, RpcEffects0 false -> RpcEffects0]; [{next_event, info, pipeline_rpcs} | true -> RpcEffects = case More of % after sending rpcs but before actioning the machine effects % a chance that effects will never be issued if the leader crashes % rpcs need to be issued _AFTER_ machine effects or there is [{3, Applied}]}]), make_pipelined_rpc_effects(State2, [{incr_metrics, ra_metrics, {State, More, RpcEffects0} = {State2, Effects0, Applied} = evaluate_quorum(State1, []), State1 = update_peer(PeerId, Peer, State0), next_index => max(NI, NextIdx)}, Peer = Peer0#{match_index => max(MI, LastIdx), Peer0 = #{match_index := MI, next_index := NI} -> {leader, State0, []}; [Id, PeerId]), ?WARN("~w saw append_entries_reply from unknown peer ~w~n", undefined -> case peer(PeerId, State0) ofad 9{zJjO? a Q H G *  t \ A 8 7  | { c R 6  X 6  -endif. ok. 3 = agreed_commit([4, 2, 3]), % 3 servers - only leader has seen new commit 4 = agreed_commit([3, 4, 4]), % only other servers have seen new commit 4 = agreed_commit([4, 4, 3]), % 3 servers - leader + 1 server has seen new commit 4 = agreed_commit([4, 4, 4]), % 2 servers - all servers have seen new commit 3 = agreed_commit([4, 3]), % 2 servers - only leader has seen new commit 4 = agreed_commit([4]), % one serveragreed_commit_test() ->-include_lib("eunit/include/eunit.hrl").-ifdef(TEST).%%% ===================%%% Internal unit tests%%% =================== {cast, To, {From, Msg}}.cast_reply(From, To, Msg) -> end. {Log, Entries} {false, Log} -> drop_existing({Log, Tail}); {true, Log} -> case ra_log:exists({Idx, Trm}, Log0) ofdrop_existing({Log0, [{Idx, Trm, _} | Tail] = Entries}) -> {Log0, []};drop_existing({Log0, []}) -> end. end {error, Reason, Log} _:Reason -> catch fold_log_from(From + ?FOLD_LOG_BATCH_SIZE, Folder, {St1, Log}) St1 = lists:foldl(Folder, St, Entries), try {Entries, Log} -> {ok, {St, Log}}; {[], Log} -> case ra_log:take(From, ?FOLD_LOG_BATCH_SIZE, Log0) offold_log_from(From, Folder, {St, Log0}) -> ?WARN("~w ~w received unhandled msg: ~W~n", [Id, RaState, Msg, 6]).log_unhandled_msg(RaState, Msg, #{id := Id}) -> lists:nth(Nth, SortedIdxs). Nth = trunc(length(SortedIdxs) / 2) + 1, SortedIdxs = lists:sort(fun erlang:'>'/2, Indexes),adHDC D 7 z Q  P  L * D |7l&`T2m5h_$rLhH end {S, E} _ -> {S, [{reply, From , {I, T}} | E]}; {_, #{from := From}, _, after_log_append} -> case C of {ok, I, T, S} = append_log_leader(C, S0), lists:foldl(fun(C, {S0, E}) -> {State0, Effects0} = %% TODO: refactor to use wal batch API?handle_leader({commands, Cmds}, State00 = #{id := _Id}) -> end; {leader, State, Effects} end, Effects1 _ -> [{reply, From, {Idx, Term}} | Effects1]; {_, #{from := From}, _, _} -> Effects1; {_, #{from := undefined}, _, _} -> Effects1; {_, _, _, await_consensus} -> Effects = case Cmd of % TODO: refactor - can this be made a bit nicer/more explicit? % check if a reply is required. Effects1 = [{incr_metrics, ra_metrics, [{2, 1}]} | Effects0], {State, _, Effects0} = make_pipelined_rpc_effects(State0, []), {ok, Idx, Term, State0} -> {leader, State, []}; "cluster_change_permitted ~w~n", [LogId, Cmd, 5, CCP]), ?WARN("~s command ~W NOT appended to log, " {not_appended, State = #{cluster_change_permitted := CCP}} -> case append_log_leader(Cmd, State00) ofhandle_leader({command, Cmd}, State00 = #{log_id := LogId}) -> {leader, State, Effects}; {State, _, Effects} = make_pipelined_rpc_effects(State1, []), State1 = State0#{cluster => Nodes#{PeerId => Peer}, log => Log}, end, {Peer0#{next_index => NextIndex}, L} % match index. % decrement next_index but don't go lower than % entry at the same index in a different term. % leader and the current leader wrote a different % The peer must have received an entry from a previous % exist % last_index has a different term or entry does not NextIndex]), [LogId, LastIdx, PeerId, LastTerm, _EntryTerm, "next_index to ~b~n", "- expected term ~b. Setting" " from ~w with term ~b " ?DEBUG("~s: leader received last_index ~b" NextIndex = max(min(NI-1, LastIdx), MI), {_EntryTerm, L} -> next_index => LastIdx + 1}, L}; {Peer0#{match_index => LastIdx, [LogId, LastIdx, LastTerm, MI]), "Resetting peer's state to last_index.~n", " lower than recorded match index [~b]." ?WARN("~s: leader saw peer with last_index [~b in ~b]" % is detected % should they turn-into non-voters when this sitution % non-persistent. % TODO: this can only really happen when peers are {_Term, L} when LastIdx < MI -> next_index => NextIdx}, L}; {Peer0#{match_index => LastIdx, [LogId, LastIdx, NextIdx, PeerId]),ad 5 H0G  u 7  w 8 Y 5   I @ N - U lG8 g!FbAX4d #{current_term := CurTerm} = State0) when Term > CurTerm ->handle_leader(#append_entries_rpc{term = Term} = Msg, end; {follower, update_term(Term, State0), [{next_event, Evt}]} Term, CurTerm]), [id(State0), Evt#install_snapshot_rpc.leader_id, "abdicates term: ~b!~n", ?INFO("~w leader saw install_snapshot_rpc from ~w for term ~b " _ -> {leader, State0, []}; [Id, Leader]), ?WARN("~w saw install_snapshot_rpc from unknown leader ~w~n", undefined -> case peer(Leader, State0) of id := Id} = State0) when Term > CurTerm -> #{current_term := CurTerm, leader_id = Leader} = Evt,handle_leader(#install_snapshot_rpc{term = Term, {leader, State, Effects}; end, Effects0 false -> [{next_event, info, pipeline_rpcs} | Effects0]; true -> Effects = case More of {State, More, Effects0} = make_pipelined_rpc_effects(State0, []),handle_leader(pipeline_rpcs, State0) -> end; {leader, State, Effects} {State, _, Effects} = make_pipelined_rpc_effects(State1, Effects0), end, _ -> [] [{demonitor, process, Pid}]; #{status := {sending_snapshot, Pid}} -> Effects0 = case Peer0 of %% we can now demonitor the process State0), next_index => LastIndex + 1}, commit_index_sent => LastIndex, match_index => LastIndex, Peer0#{status => normal, State1 = update_peer(PeerId, Peer0 -> {leader, State0, []}; [Id, PeerId]), ?WARN("~w saw install_snapshot_result from unknown peer ~w~n", undefined -> case peer(PeerId, State0) of #{id := Id} = State0) ->handle_leader({PeerId, #install_snapshot_result{last_index = LastIndex}}, end; {follower, update_term(Term, State0), []} " abdicates term: ~b!~n", [Id, Term, CurTerm]), ?DEBUG("~w: leader saw install_snapshot_result for term ~b" _ -> [LId, PeerId] [LogId, PeerId]), ?WARN("~s: saw install_snapshot_result from unknown peer ~w~n", undefined -> case peer(PeerId, State0) of when Term > CurTerm -> #{log_id := LogId, current_term := CurTerm} = State0)handle_leader({PeerId, #install_snapshot_result{term = Term}}, handle_aux(leader, Type, Cmd, State0);handle_leader({aux_command, Type, Cmd}, State0) -> {leader, State#{log => Log1}, Effects}; {Log1, Effects} = ra_log:handle_event(Evt, Log0),handle_leader({ra_log_event, Evt}, State = #{log := Log0}) -> {leader, State, [{incr_metrics, ra_metrics, [{3, Applied}]} | Effects]}; {State, _, Effects} = make_pipelined_rpc_effects(State1, Effects1), Effects0), {State1, Effects1, Applied} = evaluate_quorum(State0#{log => Log}, {Log, Effects0} = ra_log:handle_event(Evt, Log0),handle_leader({ra_log_event, {written, _} = Evt}, State0 = #{log := Log0}) -> {leader, State, Effects}; %% TOOD: ra_metrics Effects0), {State, _, Effects} = make_pipelined_rpc_effects(length(Cmds), State0, end, {State00, []}, Cmds),ad}RbF~A  u N  i .  o b  B 9 A  cD"QHl<i>%$#]$P H? } ?INFO("~w: candidate request_vote_result with higher term" when Term > CurTerm -> #{current_term := CurTerm} = State0)handle_candidate(#request_vote_result{term = Term}, end; {candidate, State0#{votes => NewVotes}, []} _ -> [{next_event, cast, {command, noop}} | Effects]}; {leader, maps:without([votes, leader_id], State), % Effects = ra_machine:leader_effects(Machine, MacState), {State, Effects} = make_all_rpcs(initialise_peers(State0)), NewVotes -> case trunc(maps:size(Nodes) / 2) + 1 of NewVotes = Votes + 1, cluster := Nodes} = State0) -> #{current_term := Term, votes := Votes,handle_candidate(#request_vote_result{term = Term, vote_granted = true}, {ra_state(), ra_server_state(), ra_effects()}.-spec handle_candidate(ra_msg() | election_timeout, ra_server_state()) -> {leader, State, []}. log_unhandled_msg(leader, Msg, State),handle_leader(Msg, State) -> {leader, State, []}; %% handle to avoid logging as unhandledhandle_leader(#pre_vote_result{}, State) -> {leader, State, []}; %% handle to avoid logging as unhandledhandle_leader(#request_vote_result{}, State) -> {leader, State, Effects}; {State, Effects} = make_all_rpcs(State0), % enforce leadership when Term =< CurTerm -> #{current_term := CurTerm} = State0)handle_leader(#pre_vote_rpc{term = Term}, end; {follower, update_term(Term, State0), [{next_event, Msg}]} " abdicates term: ~b!~n", [Id, Term, CurTerm]), ?INFO("~w leader saw pre_vote_rpc for term ~b" _ -> {leader, State0, []}; [Id, Cand]), ?WARN("~w leader saw pre_vote_rpc for unknown peer ~w~n", undefined -> case peer(Cand, State0) of id := Id} = State0) when Term > CurTerm -> #{current_term := CurTerm,handle_leader(#pre_vote_rpc{term = Term, candidate_id = Cand} = Msg, {leader, State, [{reply, Reply}]}; Reply = #request_vote_result{term = Term, vote_granted = false},handle_leader(#request_vote_rpc{}, State = #{current_term := Term}) -> end; {follower, update_term(Term, State0), [{next_event, Msg}]} Term, CurTerm]), [id(State0), Msg#request_vote_rpc.candidate_id, "abdicates term: ~b!~n", ?INFO("~w leader saw request_vote_rpc from ~w for term ~b " _ -> {leader, State0, []}; [Id, Cand]), ?WARN("~w leader saw request_vote_rpc for unknown peer ~w~n", undefined -> case peer(Cand, State0) of id := Id} = State0) when Term > CurTerm -> #{current_term := CurTerm,handle_leader(#request_vote_rpc{term = Term, candidate_id = Cand} = Msg, {leader, State0, [cast_reply(Id, LeaderId, Reply)]}; Reply = append_entries_reply(CurTerm, false, State0), id := Id} = State0) -> #{current_term := CurTerm,handle_leader(#append_entries_rpc{leader_id = LeaderId}, exit(leader_saw_append_entries_rpc_in_same_term); " this should not happen!~n", [Id, Term]), ?ERR("~w leader saw append_entries_rpc for same term ~b" id := Id}) ->handle_leader(#append_entries_rpc{term = Term}, #{current_term := Term, {follower, update_term(Term, State0), [{next_event, Msg}]}; Term, CurTerm]), [id(State0), Msg#append_entries_rpc.leader_id, "abdicates term: ~b!~n", ?INFO("~w leader saw append_entries_rpc from ~w for term ~b "adtPt-X, Q  G t > % z N   W % j@_CvV(  j5f1j5q<"t token = Token},handle_pre_vote(#pre_vote_result{term = Term, vote_granted = true, {follower, State0#{votes => 0}, [{next_event, ISR}]}; when Term >= CurTerm -> #{current_term := CurTerm} = State0)handle_pre_vote(#install_snapshot_rpc{term = Term} = ISR, {follower, State#{votes => 0}, []}; State = update_term(Term, State0), % higher term always reverts? when Term > CurTerm -> #{current_term := CurTerm} = State0)handle_pre_vote(#pre_vote_result{term = Term}, {follower, State#{votes => 0}, [{next_event, Msg}]}; % revert to follower state State = update_term(Term, State0), when Term > CurTerm -> #{current_term := CurTerm} = State0)handle_pre_vote(#request_vote_rpc{term = Term} = Msg, {follower, State#{votes => 0}, [{next_event, Msg}]}; % revert to follower state State = update_term(Term, State0), when Term >= CurTerm -> #{current_term := CurTerm} = State0)handle_pre_vote(#append_entries_rpc{term = Term} = Msg, {ra_state(), ra_server_state(), ra_effects()}.-spec handle_pre_vote(ra_msg(), ra_server_state()) -> {candidate, State, []}. log_unhandled_msg(candidate, Msg, State),handle_candidate(Msg, State) -> call_for_election(candidate, State);handle_candidate(election_timeout, State) -> {candidate, State, []}; %% handle to avoid logging as unhandledhandle_candidate(#pre_vote_result{}, State) -> {candidate, State, []}; %% handle to avoid logging as unhandledhandle_candidate(#request_vote_result{}, State) -> {candidate, State, []}; %% just ignore pre_votes that aren't of a higher termhandle_candidate(#pre_vote_rpc{}, State) -> {candidate, State, [{reply, Reply}]}; Reply = #request_vote_result{term = Term, vote_granted = false},handle_candidate(#request_vote_rpc{}, State = #{current_term := Term}) -> {follower, State, [{next_event, Msg}]}; State = update_meta(Term, undefined, State0), [id(State0), CurTerm, Term]), ?INFO("~w: candidate pre_vote_rpc with higher term received ~b -> ~b~n", when Term > CurTerm -> #{current_term := CurTerm} = State0)handle_candidate(#pre_vote_rpc{term = Term} = Msg, {follower, State, [{next_event, Msg}]}; State = update_meta(Term, undefined, State0), [id(State0), CurTerm, Term]), ?INFO("~w: candidate request_vote_rpc with higher term received ~b -> ~b~n", when Term > CurTerm -> #{current_term := CurTerm} = State0)handle_candidate(#request_vote_rpc{term = Term} = Msg, {follower, State, []}; State = update_meta(Term, undefined, State0), [id(State0), CurTerm, Term]), " term received ~b -> ~b~n", ?INFO("~w: candidate append_entries_reply with higher" when Term > CurTerm -> #{current_term := CurTerm} = State0)handle_candidate({_PeerId, #append_entries_reply{term = Term}}, {candidate, State, [{cast, LeaderId, {id(State), Reply}}]}; Reply = append_entries_reply(CurTerm, false, State), % term must be older return success=false #{current_term := CurTerm} = State) ->handle_candidate(#append_entries_rpc{leader_id = LeaderId}, {follower, State, [{next_event, Msg}]}; State = update_meta(Term, undefined, State0), #{current_term := CurTerm} = State0) when Term >= CurTerm ->handle_candidate(#append_entries_rpc{term = Term} = Msg, {candidate, State, []};handle_candidate(#request_vote_result{vote_granted = false}, State) -> {follower, State, []}; State = update_meta(Term, undefined, State0), " received ~b -> ~b", [id(State0), CurTerm, Term]),ad#sMlR+wn, ] + o C  v I * w F R  H*S'M{U1 f:~Ap:9s {FirstIdx, State0}, Entries), fun pre_append_log_follower/2, {LastIdx, State1} = lists:foldl( [{FirstIdx, _, _} | _] -> % FirstTerm evaluate_commit_index_follower(State1, []); % commit index for previously written entries % evaluate commit index as we may have received an updated leader_id => LeaderId}, log => Log2, State1 = State0#{commit_index => min(Idx, LeaderCommit), {Idx, _} = ra_log:last_index_term(Log2), % and the leader commit % entry seen (but not necessarily written) % update commit index to be the min of the last end, Log1 _ -> L; {ok, L} = ra_log:set_last_index(PLIdx, Log1), [PLIdx]), ?DEBUG("ra: resetting last index to ~b~n", %% last index to match the leader %% if no entries were sent we need to reset [] when element(1, LastIdx) > PLIdx -> Log2 = case Entries0 of LastIdx = ra_log:last_index_term(Log1), [] -> case Entries of {Log1, Entries} = drop_existing({Log0, Entries0}), % filter entries already seen {entry_ok, State0} -> case has_log_entry_or_snapshot(PLIdx, PLTerm, State00) of State00 = update_term(Term, State000), when Term >= CurTerm -> State000 = #{id := Id, log := Log0, current_term := CurTerm}) entries = Entries0}, prev_log_term = PLTerm, prev_log_index = PLIdx, leader_commit = LeaderCommit, leader_id = LeaderId,handle_follower(#append_entries_rpc{term = Term, {ra_state(), ra_server_state(), ra_effects()}.-spec handle_follower(ra_msg(), ra_server_state()) -> {pre_vote, State, []}. log_unhandled_msg(pre_vote, Msg, State),handle_pre_vote(Msg, State) -> {pre_vote, State#{log => Log}, Effects}; {Log, Effects} = ra_log:handle_event(Evt, Log0), % simply forward all other events to ra_loghandle_pre_vote({ra_log_event, Evt}, State = #{log := Log0}) -> call_for_election(pre_vote, State);handle_pre_vote(election_timeout, State) -> {pre_vote, State, []}; %% handle to avoid logging as unhandledhandle_pre_vote(#pre_vote_result{}, State) -> {pre_vote, State, []}; %% handle to avoid logging as unhandledhandle_pre_vote(#request_vote_result{}, State) -> process_pre_vote(pre_vote, PreVote, State);handle_pre_vote(#pre_vote_rpc{} = PreVote, State) -> {pre_vote, State, []}; %% just handle negative results to avoid printing an unhandled message loghandle_pre_vote(#pre_vote_result{vote_granted = false}, State) -> end; {pre_vote, State#{votes => NewVotes}, []} _ -> call_for_election(candidate, State); NewVotes -> case trunc(maps:size(Nodes) / 2) + 1 of State = update_term(Term, State0), NewVotes = Votes + 1, cluster := Nodes} = State0) -> pre_vote_token := Token, #{current_term := Term, votes := Votes,ad6E3O* >  g & s U ) d 6 K  mCn5w3s4]3W#m-f6 process_pre_vote(follower, PreVote, State);handle_follower(#pre_vote_rpc{} = PreVote, State) -> {follower, State#{log => Log}, Effects}; {Log, Effects} = ra_log:handle_event(Evt, Log0), % simply forward all other events to ra_loghandle_follower({ra_log_event, Evt}, State = #{log := Log0}) -> evaluate_commit_index_follower(State, Effects); State = State0#{log => Log}, {Log, Effects} = ra_log:handle_event(Evt, Log0), State0 = #{log := Log0}) ->handle_follower({ra_log_event, {written, _} = Evt}, {follower, State, [cast_reply(Id, LeaderId, Reply)]}; [Id, _Term, CurTerm]), ?DEBUG("~w: follower request_vote_rpc in ~b but current term ~b~n", Reply = append_entries_reply(CurTerm, false, State), % the term is lower than current term #{id := Id, current_term := CurTerm} = State) ->handle_follower(#append_entries_rpc{term = _Term, leader_id = LeaderId}, end; condition_timeout_effects => Effects}, Effects} % repeat reply effect on condition timeout condition => fun follower_catchup_cond/2, State#{leader_id => LeaderId, {await_condition, Effects = [cast_reply(Id, LeaderId, Reply)], State0), {Reply, State} = mismatch_append_entries_reply(Term, CommitIndex, % simplest way to proceed % This _may_ overwrite some valid entries but is probably the % and commit_index + 1 as the next expected. % is rewind back and use the commit index as the last index % which their next expected entry is the best we can do here % As the follower is responsible for telling the leader % that also has made progress % that it cannot commit then gets replaced by a majority leader % This situation arises when a minority leader replicates entries [Id, PLIdx, OtherTerm, PLTerm, CommitIndex + 1]), "Asking leader to resend from ~b~n", "but not with term ~b~n" ?INFO("~w: term mismatch - follower had entry at ~b with term ~b " CommitIndex = maps:get(commit_index, State0), {term_mismatch, OtherTerm, State0} -> condition_timeout_effects => Effects}, Effects}; % repeat reply effect on condition timeout condition => fun follower_catchup_cond/2, State0#{leader_id => LeaderId, {await_condition, Effects = [cast_reply(Id, LeaderId, Reply)], [Id, PLIdx, PLTerm, Reply#append_entries_reply.next_index]), " Requesting from ~b~n", ?INFO("~w: follower did not have entry at ~b in ~b." Reply = append_entries_reply(Term, false, State0), {missing, State0} -> end; end exit(Err) {error, _} = Err -> State#{condition => fun wal_down_condition/2}, []}; {await_condition, {error, wal_down} -> {follower, State#{log => Log}, []}; {ok, Log} -> case ra_log:write(Entries, Log1) of leader_id => LeaderId}, State = State1#{commit_index => min(LeaderCommit, LastIdx), % processed % last_applied will be incremented when the written event is % anything at this point. % Increment only commit_index here as we are not applyingad4Fs: K" i 4  w 3 h w ; Z Q zLQ#u7XGc'Nb4 SnapState0 = ra_log:snapshot_state(Log0), [Id, Idx, Term]), ?DEBUG("~w: begin_accept snapshot at index ~b in term ~b~n", %% index. %% only begin snapshot procedure if Idx is higher than the last_applied when Term >= CurTerm andalso Idx > LastApplied -> current_term := CurTerm} = State0) last_applied := LastApplied, #{id := Id, log := Log0, chunk_state = {1, _ChunkFlag}} = Rpc, leader_id = LeaderId, last_term = SnapTerm, last_index = Idx, last_config = Cluster,handle_follower(#install_snapshot_rpc{term = Term,%% TODO: must abort pending if for some reason we need to do so%% need to check if it's the first or last rpc {follower, State, [{reply, Reply}]}; last_index = LastIndex}, last_term = LastTerm, Reply = #install_snapshot_result{term = CurTerm, % follower receives a snapshot from an old term [Id, LastIndex, LastTerm]), ?DEBUG("~w: install_snapshot old term ~b in ~b~n", when Term < CurTerm -> State = #{id := Id, current_term := CurTerm}) last_term = LastTerm}, last_index = LastIndex,handle_follower(#install_snapshot_rpc{term = Term, {follower, update_term(Term, State), []}; State = #{current_term := CurTerm}) when Term > CurTerm ->handle_follower({_PeerId, #append_entries_reply{term = Term}}, {follower, State, [{reply, Reply}]}; Reply = #request_vote_result{term = CurTerm, vote_granted = false}, [id(State), _Cand, Term, CurTerm]), ?INFO("~w declining vote to ~w for term ~b, current term ~b~n", when Term < CurTerm -> State = #{current_term := CurTerm})handle_follower(#request_vote_rpc{term = Term, candidate_id = _Cand}, end; {follower, State#{current_term => Term}, [{reply, Reply}]} Reply = #request_vote_result{term = Term, vote_granted = false}, [id(State0), Cand, Term, {LLIdx, LLTerm}, {LastIdxTerm}]), " last log entry idxterm seen was: ~w~n", " candidate last log index term was: ~w~n" ?INFO("~w: declining vote for ~w for term ~b," false -> [{reply, Reply}]}; {follower, State#{voted_for => Cand, current_term => Term}, Reply = #request_vote_result{term = Term, vote_granted = true}, [id(State0), Cand, {LLIdx, LLTerm}, Term, CurTerm]), "for term ~b previous term was ~b~n", ?INFO("~w: granting vote for ~w with last indexterm ~w" true -> case is_candidate_log_up_to_date(LLIdx, LLTerm, LastIdxTerm) of LastIdxTerm = last_idx_term(State), State = update_term(Term, State0), when Term >= CurTerm -> #{current_term := CurTerm} = State0) last_log_term = LLTerm}, last_log_index = LLIdx,handle_follower(#request_vote_rpc{term = Term, candidate_id = Cand, {follower, State, [{reply, Reply}]}; Reply = #request_vote_result{term = Term, vote_granted = false}, [id(State), Cand, VotedFor, Term]), ?DEBUG("~w: follower request_vote_rpc for ~w already voted for ~w in ~b", % already voted for another in this term when VotedFor /= undefined andalso VotedFor /= Cand -> #{current_term := Term, voted_for := VotedFor} = State)handle_follower(#request_vote_rpc{candidate_id = Cand, term = Term},adiLZ/i; T 9 8 s % { a + T  x 7 Tg lC _*zDkAq? i #{condition_timeout_effects := Effects} = State) ->handle_await_condition(await_condition_timeout, call_for_election(pre_vote, State);handle_await_condition(election_timeout, State) -> {follower, State, [{next_event, cast, Msg}]};handle_await_condition(#request_vote_rpc{} = Msg, State) -> {ra_state(), ra_server_state(), ra_effects()}.-spec handle_await_condition(ra_msg(), ra_server_state()) -> {receive_snapshot, State, []}. %% TODO: work out what else to handle %% drop all other events?? log_unhandled_msg(receive_snapshot, Msg, State),handle_receive_snapshot(Msg, State) -> {follower, State#{log => Log}, []}; Log = ra_log:set_snapshot_state(SnapState, Log0), SnapState = ra_snapshot:abort_accept(SnapState0), SnapState0 = ra_log:snapshot_state(Log0),handle_receive_snapshot(receive_snapshot_timeout, #{log := Log0} = State) -> {receive_snapshot, State#{log => Log}, Effects}; {Log, Effects} = ra_log:handle_event(Evt, Log0), % whilst the snapshot is being written % simply forward all other events to ra_loghandle_receive_snapshot({ra_log_event, Evt}, State = #{log := Log0}) -> end; {receive_snapshot, State, [{reply, Reply}]} State = State0#{log => Log}, Log = ra_log:set_snapshot_state(SnapState, Log0), next -> {follower, persist_last_applied(State), [{reply, Reply}]}; %% follower status %% it was the last snapshot chunk so we can revert back to machine_state => MacState}, cluster => make_cluster(Id, ClusterNodes), last_applied => LastIndex, commit_index => LastIndex, current_term => Term, State = State0#{log => Log, {{_, _, ClusterNodes}, MacState} = ra_log:recover_snapshot(Log), SnapState, Log0), Log = ra_log:install_snapshot({LastIndex, LastTerm}, %% this is the last chunk so we can "install" it last -> case ChunkFlag of last_index = LastIndex}, last_term = LastTerm, Reply = #install_snapshot_result{term = CurTerm, {ok, SnapState} = ra_snapshot:accept_chunk(Data, Num, ChunkFlag, SnapState0), SnapState0 = ra_log:snapshot_state(Log0), [Id, Num, ChunkFlag]), ?DEBUG("~w: receiving snapshot chunk: ~b / ~b~n", when Term >= CurTerm -> current_term := CurTerm} = State0) #{id := Id, log := Log0, data = Data}, chunk_state = {Num, ChunkFlag}, last_term = LastTerm, last_index = LastIndex,handle_receive_snapshot(#install_snapshot_rpc{term = Term, {follower, State, []}. log_unhandled_msg(follower, Msg, State),handle_follower(Msg, State) -> call_for_election(pre_vote, State);handle_follower(election_timeout, State) -> {follower, State, []}; %% handle to avoid logging as unhandledhandle_follower(#pre_vote_result{}, State) -> {follower, State, []}; %% handle to avoid logging as unhandledhandle_follower(#request_vote_result{}, State) -> leader_id => LeaderId}, [{next_event, Rpc}]}; {receive_snapshot, State0#{log => Log, Log = ra_log:set_snapshot_state(SS, Log0), SnapState0), {ok, SS} = ra_snapshot:begin_accept({Idx, SnapTerm, Cluster},adewGt@/ U .  ^  c 1 R 7   r M ! _9 PV2oZ6-,+*[9`I&%Y State#{log => ra_log:release_resources(1, Log0)}; %% time %% followers should only ever need a single segment open at any onebecome(follower, #{log := Log0} = State) -> State#{log => Log}; Log = ra_log:release_resources(maps:size(Cluster), Log0),become(leader, #{cluster := Cluster, log := Log0} = State) ->% Internal maps:get(current_term, State).current_term(State) ->-spec current_term(ra_server_state()) -> maybe(ra_term()). maps:get(leader_id, State, undefined).leader_id(State) ->-spec leader_id(ra_server_state()) -> maybe(ra_server_id()). ra_machine:module(Machine).machine(#{machine := Machine}) ->-spec machine(ra_server_state()) -> module().uid(#{uid := UId}) -> UId.-spec uid(ra_server_state()) -> ra_uid().id(#{id := Id}) -> Id.-spec id(ra_server_state()) -> ra_server_id().% property helpers end. {RaftState, State0, []} undefined -> {RaftState, State0#{log => Log, aux_state => Aux}, []}; {no_reply, Aux, Log} -> [{reply, Reply}]}; {RaftState, State0#{log => Log, aux_state => Aux}, {reply, Reply, Aux, Log} -> Log0, MacState0) of case ra_machine:handle_aux(Machine, RaftState, Type, Cmd, Aux0, machine_state := MacState0} = State0) -> machine := Machine,handle_aux(RaftState, Type, Cmd, #{aux_state := Aux0, log := Log0, end. MinMI >= CI MinMI = lists:min([M || #{match_index := M} <- Peers]), Peers -> [] -> true; % there is only one server case maps:values(peers(State)) ofis_fully_replicated(#{commit_index := CI} = State) ->-spec is_fully_replicated(ra_server_state()) -> boolean(). LastWritten =:= LastIdxTerm. LastIdxTerm = ra_log:last_index_term(Log), LastWritten = ra_log:last_written(Log),is_fully_persisted(#{log := Log}) ->-spec is_fully_persisted(ra_server_state()) -> boolean(). ra_log:next_index(Log) =:= 1.is_new(#{log := Log}) ->-spec is_new(ra_server_state()) -> boolean(). machine => MacOverview}. O#{log => LogOverview, MacOverview = ra_machine:overview(Machine, MacState), LogOverview = ra_log:overview(Log), cluster, leader_id, voted_for], State), O = maps:with([uid, current_term, commit_index, last_applied, machine_state := MacState} = State) ->overview(#{log := Log, machine := Machine,-spec overview(ra_server_state()) -> map(). ra_machine:state_enter(Machine, RaftState, MacState)}. {become(RaftState, State), machine_state := MacState} = State) ->handle_state_enter(RaftState, #{machine := Machine, {ra_server_state() | eol, ra_effects()}.-spec handle_state_enter(ra_state() | eol, ra_server_state()) -> ra_machine:tick(Machine, Now, MacState). Now = os:system_time(millisecond),tick(#{machine := Machine, machine_state := MacState}) ->-spec tick(ra_server_state()) -> ra_effects(). end. {await_condition, State, []} % log_unhandled_msg(await_condition, Msg, State), false -> {follower, State, [{next_event, Msg}]}; true -> case Cond(Msg, State) ofhandle_await_condition(Msg, #{condition := Cond} = State) -> {await_condition, State#{log => Log}, Effects}; {Log, Effects} = ra_log:handle_event(Evt, Log0), % simply forward all other events to ra_loghandle_await_condition({ra_log_event, Evt}, State = #{log := Log0}) -> {follower, State#{condition_timeout_effects => []}, Effects};ad>V^kYL:1 a G k P O  W 6 c @   L  [_poKZ7GR aC"make_pipelined_rpc_effects(State, Effects) -> end, [], Effects)). Acc (_, Acc) -> end; [Filtered | Acc] Filtered -> [] -> Acc; case filter_follower_effects(L) of %% nested case - recurse (L, Acc) when is_list(L) -> [C | Acc]; %% by followers %% only machine monitors should not be emitted when Comp =/= machine -> ({monitor, process, Comp, _} = C, Acc) [C | Acc]; ({delete_snapshot, _} = C, Acc) -> [C | Acc]; (garbage_collection = C, Acc) -> [C | Acc]; ({aux, _} = C, Acc) -> [C | Acc]; ({incr_metrics, _, _} = C, Acc) -> [C | Acc]; lists:reverse(lists:foldl(fun ({release_cursor, _, _} = C, Acc) ->filter_follower_effects(Effects) -> {follower, State, Effects}. %% when no leader is knownevaluate_commit_index_follower(State, Effects) -> end; | Effects]} {incr_metrics, ra_metrics, [{3, Applied}]} {follower, State, [cast_reply(Id, LeaderId, Reply), Reply = append_entries_reply(Term, true, State), Effects = filter_follower_effects(Effects1), % filter the effects that should be applied on a follower {State, Effects1, Applied} -> filter_follower_effects(Effects)]}; [cast_reply(Id, LeaderId, Reply) | {delete_and_terminate, State1, Reply = append_entries_reply(Term, true, State1), {delete_and_terminate, State1, Effects} -> case catch apply_to(EffectiveCommitIndex, State0, Effects0) of % neet catch termination throw EffectiveCommitIndex = min(Idx, CommitIndex), {Idx, _} = ra_log:last_written(Log), % than the last entry received % Take the smaller of the two values as commit index may be higher % in the log as they may not have been fully persisted yet % as writes are async we can't use the index of the last available entry when LeaderId =/= undefined -> log := Log} = State0, Effects0) current_term := Term, id := Id, leader_id := LeaderId,evaluate_commit_index_follower(#{commit_index := CommitIndex, ra_log:can_write(Log).wal_down_condition(_Msg, #{log := Log}) -> false.follower_catchup_cond(_Msg, _State) -> PLIdx >= ra_log:next_index(Log); % index seen % term is ok - check if the snapshot index is greater than the last when Term >= CurTerm -> log := Log}) #{current_term := CurTerm, last_index = PLIdx},follower_catchup_cond(#install_snapshot_rpc{term = Term, end; false _ -> true; {entry_ok, _State} -> case has_log_entry_or_snapshot(PLIdx, PLTerm, State0) of when Term >= CurTerm -> State0 = #{current_term := CurTerm}) prev_log_term = PLTerm}, prev_log_index = PLIdx,follower_catchup_cond(#append_entries_rpc{term = Term, State.become(_RaftState, State) ->adRA rc<# p ] 6  w : b 0 u 7 g J $ n@?I%Z16wCXr0 {send_snapshot, PeerId, {SnapState, Id, Term}}, {LastIdx, %% and we don't pipeline entries until after snapshot %% that once the snapshot is fully replicated %% don't increment the next index here as we will do SnapState = ra_log:snapshot_state(Log), {LastIdx, _} -> State#{log => Log}); PrevTerm, MaxBatchSize, make_append_entries_rpc(PeerId, PrevIdx, % Previous index is the same as snapshot index {PrevIdx, PrevTerm} -> case ra_log:snapshot_index_term(Log) of % to send a snapshot. % The assumption here is that a missing entry means we need {undefined, Log} -> State#{log => Log}); make_append_entries_rpc(PeerId, PrevIdx, PrevTerm, MaxBatchSize, {PrevTerm, Log} when is_integer(PrevTerm) -> case ra_log:fetch_term(PrevIdx, Log0) of PrevIdx = Next - 1, current_term := Term} = State) -> #{id := Id, log := Log0,make_rpc_effect(PeerId, Next, MaxBatchSize, end, {State, []}, Peers). {S, [Eff | Effs]} make_rpc_effect(PeerId, Next, ?AER_CHUNK_SIZE, S0), {_, Eff, S} = maps:fold(fun(PeerId, #{next_index := Next}, {S0, Effs}) ->make_rpcs_for(Peers, State) -> make_rpcs_for(peers_not_sending_snapshots(State), State).make_all_rpcs(State) ->% makes empty append entries for peers that aren't pipelineable make_rpcs_for(stale_peers(State), State).make_rpcs(State) -> end, {State, false, Effects}, Cluster). Acc (_, _, Acc) -> end; Acc false -> [Eff | Effs]}; More, {update_peer(PeerId, Peer, S), NextIdx - MI < ?MAX_PIPELINE_DISTANCE), More = More0 orelse (NextIdx < NextLogIdx andalso %% is there more potentially pipelining commit_index_sent => CommitIndex}, Peer = Peer0#{next_index => NextIdx, MaxBatchSize, S0), make_rpc_effect(PeerId, NI, {NextIdx, Eff, S} = true -> case NI - MI < ?MAX_PIPELINE_DISTANCE of % next index % check if the match index isn't too far behind the % there are unsent items or a new commit index when NI < NextLogIdx orelse CI < CommitIndex -> {S0, More0, Effs} = Acc) match_index := MI} = Peer0, commit_index_sent := CI, (PeerId, #{next_index := NI, Acc; %% we should not pipeline %% if a peers is currently receiving a snapshot (_, #{status := {sending_snapshot, _}}, Acc) -> Acc; %% oneself fun (I, _, Acc) when I =:= Id -> maps:fold( NextLogIdx = ra_log:next_index(Log), Effects) -> cluster := Cluster} = State, log := Log, commit_index := CommitIndex,make_pipelined_rpc_effects(MaxBatchSize, #{id := Id, make_pipelined_rpc_effects(?AER_CHUNK_SIZE, State, Effects).ad(T>eK [ '  f * S 9 8 r K  P - , r\?>3xwv;xBAU0n`MCB[$ State); {down, Pid, Info}, noreply}}, handle_leader({command, {'$usr', #{from => undefined}, %% commit command to be processed by state machinehandle_down(leader, machine, Pid, Info, State) -> {ra_state(), ra_server_state(), ra_effects()}. pid(), term(), ra_server_state()) ->-spec handle_down(ra_state(), machine | snapshot_sender | snapshot_writer, end. State _ -> update_peer(PeerId, Peer#{status => normal}, State); [{PeerId, Peer}] -> case PeerKv of end, Peers)), (_, _) -> false true; when Pid =:= SnapshotPid -> maps:filter(fun(_, #{status := {sending_snapshot, Pid}}) maps:to_list( PeerKv =peer_snapshot_process_exited(SnapshotPid, #{cluster := Peers} = State) -> State#{cluster => maps:put(PeerId, Peer, Peers)}. Peer = maps:put(status, Status, maps:get(PeerId, Peers)),update_peer_status(PeerId, Status, #{cluster := Peers} = State) -> ra_server_state()) -> ra_server_state().-spec update_peer_status(ra_server_id(), ra_peer_status(), State#{persisted_last_applied => LastApplied}. ok = ra_log_meta:store(UId, last_applied, LastApplied),persist_last_applied(#{last_applied := LastApplied, uid := UId} = State) -> State; % if last applied is less than PL for some reason do nothing last_applied := LA} = State) when LA =< PLA ->persist_last_applied(#{persisted_last_applied := PLA,-spec persist_last_applied(ra_server_state()) -> ra_server_state().% This is done on a schedule% takes over and this% follower that has seen an entry but not the commit_index% always guarantee that side effects won't be re-issued when a% Persist last_applied - as there is an inherent race we cannot {State#{log => Log}, Effects}. MacState, Log0), {Log, Effects} = ra_log:update_release_cursor(Index, Cluster, % simply pass on release cursor index to log State = #{log := Log0, cluster := Cluster}) ->update_release_cursor(Index, MacState, {ra_server_state(), ra_effects()}.-spec update_release_cursor(ra_index(), term(), ra_server_state()) ->% at this index.% stores the cluster config at an index such that we can later snapshot State#{log => Log}}. leader_commit = CommitIndex}}, prev_log_term = PrevTerm, prev_log_index = PrevIdx, leader_id = Id, term = Term, {send_rpc, PeerId, #append_entries_rpc{entries = Entries, {NextIndex, end, LastIdx + 1 {Next, _, _} = hd(Entries), %% assertion {LastIdx, _, _} = lists:last(Entries), _ -> [] -> Next; NextIndex = case Entries of {Entries, Log} = ra_log:take(Next, Num, Log0), %% read at fixed cost %% ra_log:take should be able to return the actual number of entries %% TODO: refactor to avoid lists:last call later Next = PrevIdx + 1, commit_index := CommitIndex} = State) -> #{log := Log0, current_term := Term, id := Id,make_append_entries_rpc(PeerId, PrevIdx, PrevTerm, Num, end. end State#{log => Log}}ad"WzZ}p,  \ 3 2 1 a 8 0  q L $ e Q B  k b a   I*yBbNR*h7b$uJ [{next_event, cast, VoteForSelf}, {send_vote_requests, Reqs}]}. pre_vote_token => Token}, {pre_vote, State#{leader_id => undefined, votes => 0, State = update_meta(Term, Id, State0), vote_granted = true}, VoteForSelf = #pre_vote_result{term = Term, token = Token, % vote for self || PeerId <- PeerIds], last_log_term = LastTerm}} last_log_index = LastIdx, candidate_id = Id, token = Token, Reqs = [{PeerId, #pre_vote_rpc{term = Term, {LastIdx, LastTerm} = last_idx_term(State0), PeerIds = peer_ids(State0), Token = make_ref(), ?DEBUG("~w: pre_vote election called for in term ~b~n", [Id, Term]), current_term := Term} = State0) ->call_for_election(pre_vote, #{id := Id, [{next_event, cast, VoteForSelf}, {send_vote_requests, Reqs}]}; {candidate, State#{leader_id => undefined, votes => 0}, State = update_meta(NewTerm, Id, State0), VoteForSelf = #request_vote_result{term = NewTerm, vote_granted = true}, % vote for self || PeerId <- PeerIds], last_log_term = LastTerm}} last_log_index = LastIdx, candidate_id = Id, Reqs = [{PeerId, #request_vote_rpc{term = NewTerm, {LastIdx, LastTerm} = last_idx_term(State0), % increment current term PeerIds = peer_ids(State0), ?DEBUG("~w: election called for in term ~b~n", [Id, NewTerm]), NewTerm = CurrentTerm + 1, current_term := CurrentTerm} = State0) ->call_for_election(candidate, #{id := Id,%%%===================================================================%%% Internal functions%%%=================================================================== end. {error, Reason, RaState#{log => Log1}} {error, Reason, Log1} -> {ok, State1, RaState#{log => Log1}}; {ok, {State1, Log1}} -> case fold_log_from(Idx, Fun, {State, Log}) of end, 1 undefined -> PrevIdx; {PrevIdx, _PrevTerm} -> Idx = case ra_log:snapshot_index_term(Log) oflog_fold(#{log := Log} = RaState, Fun, State) -> {error, term(), ra_server_state()}. {ok, State, ra_server_state()} |-spec log_fold(ra_server_state(), fun((term(), State) -> State), State) -> ok. catch ra_log:close(Log), #{log := Log} = persist_last_applied(State),terminate(State, _Reason) -> ok; catch ra_log:delete_everything(Log), ?NOTICE("~w: terminating and deleting all data~n", [id(_State)]),terminate(#{log := Log} = _State, {shutdown, delete}) ->-spec terminate(ra_server_state(), Reason :: {shutdown, delete} | term()) -> ok. {RaftState, State#{log => Log}, []}. Log = ra_log:set_snapshot_state(SnapState, Log0), SnapState = ra_snapshot:handle_down(Pid, Info, SnapState0), SnapState0 = ra_log:snapshot_state(Log0), end, [Id, Pid, Info]) ?WARN("~w: Snapshot write process ~w exited with ~w~n", _ -> normal -> ok; noproc -> ok; case Info of #{id := Id, log := Log0} = State) ->handle_down(RaftState, snapshot_writer, Pid, Info, {leader, peer_snapshot_process_exited(Pid, State), []}; [Id, Pid, Info, 10]), ?DEBUG("~w: Snapshot sender process ~w exited with ~W~n",handle_down(leader, snapshot_sender, Pid, Info, #{id := Id} = State) ->ad!XyAnG ` * s c  D  ~ a E J .  t<f7!xJI;$ po3 ]&nDQ3 maps:keys(peers(State)).peer_ids(State) -> end, peers(State)). false (_, _Peer) -> true; % the commit index has been updated when CI < CommitIndex -> (_, #{commit_index_sent := CI}) true; % there are unconfirmed items when MI < NI - 1 -> match_index := MI}) (_, #{next_index := NI, false; maps:filter(fun (_, #{status := {sending_snapshot, _}}) ->stale_peers(#{commit_index := CommitIndex} = State) ->% peers that could need an update end, peers(State)). (_, _) -> true maps:filter(fun (_, #{status := {sending_snapshot, _}}) -> false;peers_not_sending_snapshots(State) ->%% remove any peers that are currently receiving a snapshot maps:remove(Id, Peers).peers(#{id := Id, cluster := Peers}) -> maps:merge(new_peer(), Map).new_peer_with(Map) -> commit_index_sent => 0}. match_index => 0, #{next_index => 1,new_peer() -> vote_granted = Success}. token = Token, #pre_vote_result{term = Term,pre_vote_result(Term, Token, Success) -> [{reply, pre_vote_result(CurTerm, Token, false)}]}. {FsmState, State, [id(State), _Cand, Term, CurTerm]), ?DEBUG("~w declining pre-vote to ~w for term ~b, current term ~b~n", when Term < CurTerm -> #{current_term := CurTerm} = State) candidate_id = _Cand}, token = Token,process_pre_vote(FsmState, #pre_vote_rpc{term = Term, end; end [{reply, pre_vote_result(Term, Token, false)}]} {FsmState, State, pre_vote -> call_for_election(pre_vote, State); %% stale pre voter %% likely to win but could be held back by a persistent %% immediately enter pre_vote election as this node is more follower -> case FsmState of [id(State0), Cand, Term, {LLIdx, LLTerm}, LastIdxTerm]), "Last log entry idxterm seen was: ~w~n", " candidate last log index term was: ~w~n" ?DEBUG("~w: declining pre-vote for ~w for term ~b," false -> {FsmState, State, [{reply, pre_vote_result(Term, Token, false)}]}; [id(State0), Cand, Version]), ?DEBUG("~w: declining pre-vote for ~w for protocol version ~b~n", true -> [{reply, pre_vote_result(Term, Token, true)}]}; {FsmState, State#{voted_for => Cand}, [id(State0), Cand, {LLIdx, LLTerm}, Term, CurTerm]), " for term ~b previous term ~b~n", ?DEBUG("~w: granting pre-vote for ~w with last indexterm ~w" true when Version =< ?RA_PROTO_VERSION -> case is_candidate_log_up_to_date(LLIdx, LLTerm, LastIdxTerm) of LastIdxTerm = last_idx_term(State), State = update_term(Term, State0), when Term >= CurTerm -> #{current_term := CurTerm}= State0) last_log_term = LLTerm}, last_log_index = LLIdx, token = Token, version = Version,process_pre_vote(FsmState, #pre_vote_rpc{term = Term, candidate_id = Cand,ad!exIH{F  \ /  z - ~ 4 % | r E : 9 a - ~mV*R}T2ulk+ 7yJ*)b1 To = min(From + 1024, ApplyTo), From = LastApplied + 1, when ApplyTo > LastApplied -> machine_state := MacState0} = State0) #{last_applied := LastApplied,apply_to(ApplyTo, ApplyFun, NumApplied, Notifys0, Effects0, apply_to(ApplyTo, ApplyFun, 0, #{}, Effs, State).apply_to(ApplyTo, ApplyFun, State, Effs) -> end, State, Effs). apply_with(Machine, E, S) fun(E, S) -> apply_to(ApplyTo,apply_to(ApplyTo, #{machine := Machine} = State, Effs) -> State#{cluster => Cluster}. end, Cluster0, PeerIds), new_peer_with(#{next_index => NextIdx})} Acc#{PeerId => Cluster = lists:foldl(fun(PeerId, Acc) -> NextIdx = ra_log:next_index(Log), PeerIds = peer_ids(State),initialise_peers(State = #{log := Log, cluster := Cluster0}) -> end. Cluster#{Self => new_peer()} % add current server to cluster Cluster -> Cluster; % current server is already in cluster - do nothing #{Self := _} = Cluster -> end, #{}, Nodes) of Acc#{N => new_peer()} case lists:foldl(fun(N, Acc) ->make_cluster(Self, Nodes) -> {Entries, State#{log => Log}}. {Entries, Log} = ra_log:take(From, To - From + 1, Log0),fetch_entries(From, To, #{log := Log0} = State) -> ra_log:fetch_term(Idx, Log).fetch_term(Idx, #{log := Log}) -> end. {term_mismatch, OtherTerm, State#{log => Log}} {OtherTerm, Log} -> {entry_ok, State#{log => Log}}; {Term, Log} -> end; {missing, State#{log => Log}} _ -> {term_mismatch, OtherTerm, State#{log => Log}}; {Idx, OtherTerm} -> {entry_ok, State#{log => Log}}; {Idx, Term} -> case ra_log:snapshot_index_term(Log) of {undefined, Log} -> case ra_log:fetch_term(Idx, Log0) ofhas_log_entry_or_snapshot(Idx, Term, #{log := Log0} = State) -> false.is_candidate_log_up_to_date(_, _, {_, _}) -> true; when Idx >= LastIdx ->is_candidate_log_up_to_date(Idx, Term, {LastIdx, Term}) true; when Term > LastTerm ->is_candidate_log_up_to_date(_, Term, {_, LastTerm}) boolean().-spec is_candidate_log_up_to_date(ra_index(), ra_term(), ra_idxterm()) ->%% longer is more up-to-dat%% up-to-date. If the logs end with the same term, then whichever log is%% entries with different terms, then the log with the later term is more%% the index and term of the last entries in the logs. If the logs have last%% § 5.4.1 Raft determines which of two logs is more up-to-date by comparing ra_log:last_index_term(Log).last_idx_term(#{log := Log}) -> State.update_term(_, State) -> update_meta(Term, undefined, State); when Term =/= undefined andalso Term > CurTerm ->update_term(Term, State = #{current_term := CurTerm}) voted_for => VotedFor}. State#{current_term => Term, ok = ra_log_meta:store_sync(UId, voted_for, VotedFor), ok = ra_log_meta:store(UId, current_term, Term),update_meta(Term, VotedFor, #{uid := UId} = State) -> State; %% no update needed voted_for := VotedFor} = State) ->update_meta(Term, VotedFor, #{current_term := Term, State#{cluster => Nodes#{PeerId => Peer}}.update_peer(PeerId, Peer, #{cluster := Nodes} = State) -> maps:get(PeerId, Nodes, undefined).peer(PeerId, #{cluster := Nodes}) ->adLJvEm0 m ( O F  W 3 2  o $ b  y4EiTW?=f;' % cluster deletion {_, State0, MacSt, Effects0, Notifys0}) -> {Idx, _, {'$ra_cluster', #{from := From}, delete, ReplyType}},apply_with(Machine, {Idx, State, MacSt, Effects, Notifys}; %% don't log these as unhandled {_, State, MacSt, Effects, Notifys}) ->apply_with(_, {Idx, _, noop}, {Idx, State, MacSt, Effects, Notifys}; State = State0#{cluster_change_permitted => true}, ?DEBUG("~w: enabling ra cluster changes in ~b~n", [id(State0), Term]), {_, State0 = #{current_term := Term}, MacSt, Effects, Notifys}) -> {Idx, Term, noop}, % Idxapply_with(_, % Machine {Idx, State1, MacSt, Effects1, Notifys}; {Effects1, State1} = add_next_cluster_change(Effects, State), % add pending cluster change as next event State = State0#{cluster_change_permitted => true}, Effects0, Notifys0), {Effects, Notifys} = add_reply(From, ok, ReplyType, [id(State0), maps:keys(_New)]), ?DEBUG("~w: applying ra cluster change to ~w~n", {_, State0, MacSt, Effects0, Notifys0}) -> ReplyType}}, {Idx, _, {'$ra_cluster_change', #{from := From}, _New,apply_with(_Machine, {Idx, State, MacSt, Effects, Notifys0}; Effects = [{reply, From, {machine_reply, Result}} | Effects0], Result = ra_machine:query(MacMod, QueryFun, MacSt), {_, State, MacSt, Effects0, Notifys0}) -> {Idx, _, {'$ra_query', #{from := From}, QueryFun, _}},apply_with({machine, MacMod, _}, % Machine end; {Idx, State, NextMacSt, ReplyEffs, Notifys} Effects, Notifys0), {ReplyEffs, Notifys} = add_reply(From, Reply, ReplyType, {NextMacSt, Reply} -> {Idx, State, NextMacSt, [AppEffs | ReplyEffs], Notifys}; Effects, Notifys0), {ReplyEffs, Notifys} = add_reply(From, Reply, ReplyType, {NextMacSt, Reply, AppEffs} -> case ra_machine:apply(Machine, Meta, Cmd, MacSt) of Meta = Meta0#{index => Idx, term => Term}, {_, State, MacSt, Effects, Notifys0}) -> {Idx, Term, {'$usr', #{from := From} = Meta0, Cmd, ReplyType}},apply_with(Machine, end, Prior, Nots). [{notify, Pid, lists:reverse(Corrs)} | Acc] maps:fold(fun (Pid, Corrs, Acc) ->make_notify_effects(Nots, Prior) -> {State, FinalEffs, NumApplied}. FinalEffs = make_notify_effects(Notifys, lists:reverse(Effects)), %% notifications are processed first %% reverse list before consing the notifications to ensureapply_to(_, _, NumApplied, Notifys, Effects, State) -> % ApplyTo end; machine_state => MacState}) Notifys, Effects, State#{last_applied => AppliedTo, apply_to(ApplyTo, ApplyFun, NumApplied + length(Entries), % ?INFO("applied: ~p ~nAfter: ~p", [Entries, MacState]), % {AppliedTo,_, _} = lists:last(Entries), Effects0, Notifys0}, Entries), lists:foldl(ApplyFun, {LastApplied, State1, MacState0, {AppliedTo, State, MacState, Effects, Notifys} = {[{From, _, _} | _] = Entries, State1} -> %% assert first item read is from {State, FinalEffs, NumApplied}; FinalEffs = make_notify_effects(Notifys0, lists:reverse(Effects0)), %% notifications are processed first %% reverse list before consing the notifications to ensure {[], State} -> case fetch_entries(From, To, State0) ofad~Tg)8 ` C B ! t E    F ' h T =  v ^ ] 2 o"t=&j]]F'rPGvuCG~ {Idx, State#{cluster => Cluster, {'$ra_cluster_change', _, Cluster, _} -> case Cmd of % cluster % change (can this even happen?) we should revert back to the last known % it has been overwritten by a new leader. Unless it is another cluster % the index for the cluster config entry has a different term, i.e. when Term /= CITTerm -> {_, State = #{cluster_index_term := {Idx, CITTerm}}})pre_append_log_follower({Idx, Term, Cmd} = Entry, {ok, NextIdx, Term, State#{log => Log}}. Log = ra_log:append({NextIdx, Term, Cmd}, Log0), NextIdx = ra_log:next_index(Log0),append_log_leader(Cmd, State = #{log := Log0, current_term := Term}) -> end; {not_appended, State} % not a member - do nothing _ -> append_cluster_change(Cluster, From, ReplyMode, State); Cluster = maps:remove(LeavingNode, OldCluster), #{LeavingNode := _} -> case OldCluster of State = #{cluster := OldCluster}) ->append_log_leader({'$ra_leave', From, LeavingNode, ReplyMode}, end; append_cluster_change(Cluster, From, ReplyMode, State) Cluster = OldCluster#{JoiningNode => new_peer()}, _ -> {not_appended, State}; % TODO: reply? If we don't reply the caller may block until timeout % already a member do nothing #{JoiningNode := _} -> case OldCluster of State = #{cluster := OldCluster}) ->append_log_leader({'$ra_join', From, JoiningNode, ReplyMode}, {not_appended, State#{pending_cluster_changes => Pending ++ [Cmd]}}; % in this term - stash the request % cluster change is in progress or leader has not yet committed anything CmdTag == '$ra_leave' -> when CmdTag == '$ra_join' orelse pending_cluster_changes := Pending}) State = #{cluster_change_permitted := false,append_log_leader({CmdTag, _, _, _} = Cmd, {Effects, Notifys}. Effects, Notifys) ->add_reply(_, _, _, % From, Reply, Mode end; {Effects, Notifys#{Pid => [CorrData]}} _ -> {Effects, Notifys#{Pid => [CorrData | T]}}; #{Pid := T} -> case Notifys of CorrData = {Corr, Reply}, % own bookkeeping % reply with the supplied correlation so that the sending can do their % notify are casts and thus have to include their own pid() Effects, Notifys) ->add_reply(undefined, Reply, {notify_on_consensus, Corr, Pid}, {[{reply, From, {machine_reply, Reply}} | Effects], Notifys};add_reply(From, Reply, await_consensus, Effects, Notifys) -> {Effects, State}.add_next_cluster_change(Effects, State) -> State#{pending_cluster_changes => Rest}}; {[{next_event, {call, From}, {command, C}} | Effects], {_, #{from := From} , _, _} = C, #{pending_cluster_changes := [C | Rest]} = State) ->add_next_cluster_change(Effects, setelement(1, Acc, Idx). [id(element(2, Acc)), Cmd, 10]), ?WARN("~p: apply_with: unhandled command: ~W~n", % TODO: remove to make more strics, ideally we should not need a catch allapply_with(_, {Idx, _, _} = Cmd, Acc) -> throw({delete_and_terminate, State, EOLEffects ++ NotEffs ++ Effects1}); State = State0#{last_applied => Idx, machine_state => MacSt}, % need to update the state before throw % non-local return to be caught by ra_server_proc EOLEffects = ra_machine:state_enter(Machine, eol, MacSt), %% virtual "eol" state NotEffs = make_notify_effects(Notifys, []), {Effects1, Notifys} = add_reply(From, ok, ReplyType, Effects0, Notifys0),adU=G> d 0   7 N '  \ D , h g  i0r@[2gfBWW9wLagreed_commit(Indexes) ->-spec agreed_commit(list()) -> ra_index(). end, [LWIdx], peers(State)). [Idx | Acc] maps:fold(fun(_K, #{match_index := Idx}, Acc) -> {LWIdx, _} = ra_log:last_written(Log),match_indexes(#{log := Log} = State) -> end. State#{log => Log} {_, Log} -> log => Log}; State#{commit_index => PotentialNewCommitIndex, {CurrentTerm, Log} -> case fetch_term(PotentialNewCommitIndex, State) of % log entry term matches the current term. See (§5.4.2) % leaders can only increment their commit index if the corresponding PotentialNewCommitIndex = agreed_commit(match_indexes(State)),increment_commit_index(State = #{current_term := CurrentTerm}) -> apply_to(CI, State, Effects). State = #{commit_index := CI} = increment_commit_index(State0), % TODO: shortcut function if commit index was not incrementedevaluate_quorum(State0, Effects) -> last_term = LWTerm}. last_index = LWIdx, next_index = LastIdx + 1, #append_entries_reply{term = Term, success = Success, {LastIdx, _} = last_idx_term(State), {LWIdx, LWTerm} = ra_log:last_written(Log), % be written. % the follower may resent items that are currently waiting to % also we can't use the last writted Idx as then % as it may not have been persisted yet % ah - we can't use the the last received idxappend_entries_reply(Term, Success, State = #{log := Log}) -> State#{log => Log}}. last_term = CITerm}, last_index = CommitIndex, next_index = CommitIndex + 1, {#append_entries_reply{term = Term, success = false, false = CITerm =:= undefined, % assert CITerm is found {CITerm, Log} = ra_log:fetch_term(CommitIndex, Log0),mismatch_append_entries_reply(Term, CommitIndex, State = #{log := Log0}) -> previous_cluster => {PrevCITIdx, PrevCITTerm, PrevCluster}}}. cluster_index_term => IdxTerm, cluster_change_permitted => false, cluster => Cluster, State#{log => Log, {ok, NextIdx, Term, Log = ra_log:append({NextIdx, Term, Command}, Log0), % what happens if the write fails? % TODO: is it safe to do change the cluster config with an async write? IdxTerm = {NextIdx, Term}, NextIdx = ra_log:next_index(Log0), Command = {'$ra_cluster_change', From, Cluster, ReplyMode}, % that include the new cluster configuration % turn join command into a generic cluster change command current_term := Term}) -> cluster_index_term := {PrevCITIdx, PrevCITTerm}, cluster := PrevCluster, State = #{log := Log0,append_cluster_change(Cluster, From, ReplyMode, {Idx, State}.pre_append_log_follower({Idx, _, _}, {_, State}) -> cluster_index_term => {Idx, Term}}}; {{Idx, Term}, State#{cluster => Cluster, {_, State}) ->pre_append_log_follower({Idx, Term, {'$ra_cluster_change', _, Cluster, _}}, end; pre_append_log_follower(Entry, {Idx, State1}) cluster_index_term => {PrevIdx, PrevTerm}}, State1 = State#{cluster => PrevCluster, State), {PrevIdx, PrevTerm, PrevCluster} = maps:get(previous_cluster, % revert back to previous cluster _ -> cluster_index_term => {Idx, Term}}};adprU9zgH( m K * Q P 5   b 7 g E # v I  p&ba/.`!|{Q[4b@? wvaG1 {reply, term(), ra_reply_body()} | {reply, ra_reply_body()} | ra_log:effect() | ra_machine:effect() |-type ra_effect() :: #pre_vote_result{}. #install_snapshot_result{} | #request_vote_result{} |-type ra_reply_body() :: #append_entries_reply{} | ra_log:event(). {commands, [command()]} | {command, command()} | await_condition_timeout | election_timeout | #install_snapshot_rpc{} | #pre_vote_result{} | #pre_vote_rpc{} | #request_vote_result{} | #request_vote_rpc{} | {ra_server_id(), #install_snapshot_result{}} | {ra_server_id(), #append_entries_reply{}} |-type ra_msg() :: #append_entries_rpc{} | UserCommand :: term(), command_reply_mode()} | noop.-type command() :: {command_type(), command_meta(), noreply. command_correlation(), pid()} | {notify_on_consensus, await_consensus |-type command_reply_mode() :: after_log_append |-type command_correlation() :: integer() | reference().-type command_meta() :: #{from := maybe(from())}. '$ra_cluster_change' | '$ra_cluster'.-type command_type() :: '$usr' | '$ra_query' | '$ra_join' | '$ra_leave' | | recovered | stop | receive_snapshot. | terminating_leader | terminating_follower | recover | pre_vote | await_condition | delete_and_terminate-type ra_state() :: leader | follower | candidate }. pre_vote_token => reference() condition_timeout_effects => [ra_effect()], condition => ra_await_condition_fun(), aux_state => term(), machine_state := term(), machine := ra_machine:machine(), stop_after => ra_index(), persisted_last_applied => ra_index(), last_applied := ra_index(), commit_index := ra_index(), votes => non_neg_integer(), voted_for => maybe(ra_server_id()), % persistent log := term(), current_term := ra_term(), previous_cluster => {ra_index(), ra_term(), ra_cluster()}, pending_cluster_changes := [term()], cluster_index_term := ra_idxterm(), cluster_change_permitted := boolean(), cluster := ra_cluster(), leader_id => maybe(ra_server_id()), uid := ra_uid(), #{id := ra_server_id(),-type ra_server_state() :: fun((ra_msg(), ra_server_state()) -> boolean()).-type ra_await_condition_fun() :: ]). recover/1 log_fold/3, terminate/2, handle_down/5, update_peer_status/3, persist_last_applied/1, update_release_cursor/3, make_rpcs/1, % TODO: hide behind a handle_leader current_term/1, machine/1, leader_id/1, uid/1, id/1, % properties is_fully_replicated/1, is_fully_persisted/1, is_new/1, overview/1, tick/1, handle_state_enter/2, handle_aux/4, handle_await_condition/2, handle_receive_snapshot/2, handle_follower/2, handle_pre_vote/2, handle_candidate/2, handle_leader/2, init/1, name/2,-export([-compile(inline_list_funcs).-include("ra.hrl").-module(ra_server).adUj2gD  ?  e  Y * j ' Y Z 54jL'qJIZI*yG%lC Log0 = ra_log:init(LogInitArgs#{snapshot_module => SnapModule}), SnapModule = ra_machine:snapshot_module(Machine), end, {machine, Mod, Args} {module, Mod, Args} -> initial_state => S}}; {machine, ra_machine_simple, #{simple_fun => Fun, {simple, Fun, S} -> Machine = case MachineConf of Name = ra_lib:ra_server_id_to_local_name(Id), machine := MachineConf} = Config) -> log_init_args := LogInitArgs, initial_members := InitialNodes, cluster_name := _ClusterName, uid := UId, friendly_name := LogId,init(#{id := Id,-spec init(ra_server_config()) -> ra_server_state(). list_to_atom("ra_" ++ ClusterName ++ "_server_" ++ UniqueSuffix).name(ClusterName, UniqueSuffix) ->-spec name(ClusterName :: ra_cluster_name(), UniqueSuffix::string()) -> atom().-define(MAX_PIPELINE_DISTANCE, 10000).% TODO: make configurable% TODO: test what is a good defult here-define(FOLD_LOG_BATCH_SIZE, 25).-define(AER_CHUNK_SIZE, 25). ]). command_reply_mode/0 command_correlation/0, command_meta/0, command_type/0, command/0, machine_conf/0, ra_msg/0, ra_server_config/0, ra_state/0,-export_type([ra_server_state/0, await_condition_timeout => non_neg_integer()}. tick_timeout => non_neg_integer(), % ms % and persisting last_applied index % for periodic actions such as sending stale rpcs broadcast_time => non_neg_integer(), % ms % setting election timeouts % TODO: review - only really used for machine := machine_conf(), initial_members := [ra_server_id()], log_init_args := ra_log:ra_log_init_args(), cluster_name := ra_cluster_name(), friendly_name := unicode:chardata(), %% with `~w' %% server - will default to the id formatted %% a friendly name to refer to a particular uid := ra_uid(),-type ra_server_config() :: #{id := ra_server_id(),%% machines that cannot access any of the advanced features.%% The simple machine config is version that can only be used for simple state%% and the config to pass to the {@link ra_machine:init/1} implementation.%% This is how ra knows which module to use to invoke the ra_machine callbacks%% The machine configuration. InitialState :: term()}. {simple, simple_apply_fun(term()),-type machine_conf() :: {module, module(), InitConfig :: map()} |-type simple_apply_fun(State) :: fun((term(), State) -> State).-type ra_effects() :: [ra_effect()]. [{Pos :: non_neg_integer(), Incr :: integer()}]}. {incr_metrics, Table :: atom(), {notify, pid(), reference()} | {next_event, cast, ra_msg()} | {next_event, ra_msg()} | LeaderId :: ra_server_id(), Term :: ra_term()}} | {Module :: module(), Ref :: term(), {send_snapshot, To :: ra_server_id(), {send_rpc, ra_server_id(), #append_entries_rpc{}} | #request_vote_rpc{} | #pre_vote_rpc{}}]} | {send_vote_requests, [{ra_server_id(), {cast, ra_server_id(), term()} |ad;^! Y% E   _ " { D  W +  x \ 9  X MhP0[ K \F({1l State0 = #{current_term := Term, id := Id}) -> last_index = LastIdx}}, next_index = NextIdx,handle_leader({PeerId, #append_entries_reply{term = Term, success = true, {ra_state(), ra_server_state(), ra_effects()}.-spec handle_leader(ra_msg(), ra_server_state()) -> State#{log => Log}. Log = ra_log:release_resources(1, Log0), State0, []), end, setelement(4, apply_with(Machine, E, S), []) %% on server startup (queue recovery) %% we throw away %% up a long list of effects than then %% Clear out the effects to avoid building fun(E, S) -> apply_to(CommitIndex, {#{log := Log0} = State, _, _} = [LogId, _LastApplied, CommitIndex]), ?DEBUG("~s: recovering state machine from ~b to ~b~n", machine := Machine} = State0) -> last_applied := _LastApplied, commit_inde log_id := LogId,rrecover(#{log_id := LogId, cluster_index_term => ClusterIndexTerm}. cluster => Cluster, State0#{log => Log, % TODO: do we need to set previous cluster here? end, {{SnapshotIndexTerm, Cluster0}, Log0}), Acc (_, Acc) -> {{Idx, Term}, Cluster}; fun({Idx, Term, {'$ra_cluster_change', _, Cluster, _}}, _) -> fold_log_from(CommitIndex, {ok, {{ClusterIndexTerm, Cluster}, Log}} = % peers and become a leader % This is required as otherwise a server could restart without any known % Find last cluster change and idxterm and use as initial cluster condition_timeout_effects => []}, aux_state => ra_machine:init_aux(Machine, Name), machine_state => MacState, machine => Machine, log => Log0, persisted_last_applied => LastApplied, last_applied => FirstIndex, commit_index => CommitIndex, voted_for => VotedFor, current_term => CurrentTerm, pending_cluster_changes => [], cluster_index_term => {0, 0}, cluster_change_permitted => false, % to be applied in the current term? % there is no `noop` command % TODO: what if we have snapshotted and % commands that can't be applied. % yet re-applied its noop command that we may receive other join % starts up but hasn't % There may be scenarios when a single server cluster => Cluster0, log_id => LogId, uid => UId, State0 = #{id => Id, CommitIndex = max(LastApplied, FirstIndex), end, {Idx, Clu, MacSt, {Idx, Term}} %% the snapshot is the last index before the first index Clu = make_cluster(Id, ClusterNodes), {{Idx, Term, ClusterNodes}, MacSt} -> InitialMachineState, {0, 0}}; {0, make_cluster(Id, InitialNodes), InitialMachineState = ra_machine:init(Machine, Name), undefined -> case ra_log:recover_snapshot(Log0) of {FirstIndex, Cluster0, MacState, SnapshotIndexTerm} = VotedFor = ra_log_meta:fetch(UId, voted_for, undefined), LastApplied = ra_log_meta:fetch(UId, last_applied, 0), CurrentTerm = ra_log_meta:fetch(UId, current_term, 0), ok = ra_log:write_config(Config, Log0),ad Kv6  q S &  \  State0 = #{current_term := Term, id := Id}) -> last_index = LastIdx}}, next_index = NextIdx,handle_leader({PeerId, #append_entries_reply{term = Term, success = true, {ra_state(), ra_server_state(), ra_effects()}.-spec handle_leader(ra_msg(), ra_server_state()) -> State#{log => Log}. Log = ra_log:release_resources(1, Log0), State0, []), end, setelement(4, apply_with(Machine, E, S), []) %% on server startup (queue recovery) %% we throw away %% up a long list of effects than then %% Clear out the effects to avoid building fun(E, S) -> apply_to(CommitIndex, {#{log := Log0} = State, _, _} = [LogId, _LastApplied, CommitIndex]), ?DEBUG("~s: recovering state machine from ~b to ~b~n", machine := Machine} = State0) -> last_applied := _LastApplied, commit_index := CommitIndex,ad|H,`6yi`$ ;  u H  E P v + p2H " next_index ~b for ~w", ?DEBUG("~s: setting last index to ~b, " {LastTerm, L} when LastIdx >= MI -> % entry exists we can forward next_index => NextIdx}, L}; {Peer0#{match_index => LastIdx, [LogId, PeerId, NextIdx]), ?DEBUG("~s: setting next index for ~w ~b", % entry was not found - simply set next index to {undefined, L} -> {Peer, Log} = case ra_log:fetch_term(LastIdx, Log0) of % match_index and update next_index directly % if the last_index exists and has a matching term we can forward next_index := NI}} = Nodes, #{PeerId := Peer0 = #{match_index := MI, State0 = #{log_id := LogId, cluster := Nodes, log := Log0}) -> last_term = LastTerm}}, last_index = LastIdx, next_index = NextIdx,handle_leader({PeerId, #append_entries_reply{success = false, end; {follower, update_term(Term, State0), []} [LogId, Term, CurTerm]), "abdicates term: ~b!~n", ?NOTICE("~s leader saw append_entries_reply for term ~b " _ -> {leader, State0, []}; [LogId, PeerId]), ?WARN("~s saw append_entries_reply from unknown peer ~w~n", undefined -> case peer(PeerId, State0) of log_id := LogId} = State0) when Term > CurTerm -> #{current_term := CurTerm,handle_leader({PeerId, #append_entries_reply{term = Term}}, end; end {leader, State, Effects} _ -> {stop, State, Effects}; ?INFO("~s: leader not in new cluster - goodbye", [LogId]), % time to say goodbye % config has been committed % leader is not in the cluster and the new cluster when CI >= CITIndex ->ad%):Ds^ \  b a 1 j N M a F , { J q'`s) #{current_term := CurTerm} = State0) when Term > CurTerm ->handle_leader(#append_entries_rpc{term = Term} = Msg, end; {follower, update_term(Term, State0), [{next_event, Evt}]} Term, CurTerm]), [id(State0), Evt#install_snapshot_rpc.leader_id, "abdicates term: ~b!~n", ?INFO("~w leader saw install_snapshot_rpc from ~w for term ~b " _ -> {leader, State0, []}; [Id, Leader]), ?WARN("~w saw install_snapshot_rpc from unknown leader ~w~n", undefined -> case peer(Leader, State0) of id := Id} = State0) when Term > CurTerm -> #{current_term := CurTerm, leader_id = Leader} = Evt,handle_leader(#install_snapshot_rpc{term = Term, {leader, State, Effects}; end, Effects0 false -> [{next_event, info, pipeline_rpcs} | Effects0]; true -> Effects = case More of {State, More, Effects0} = make_pipelined_rpc_effects(State0, []),handle_leader(pipeline_rpcs, State0) -> end; {leader, State, Effects} {State, _, Effects} = make_pipelined_rpc_effects(State1, Effects0), end, _ -> [] [{demonitor, process, Pid}]; #{status := {sending_snapshot, Pid}} -> Effects0 = case Peer0 of %% we can now demonitor the process State0), next_index => LastIndex + 1}, commit_index_sent => LastIndex, match_index => LastIndex, Peer0#{status => normal, State1 = update_peer(PeerId, Peer0 -> {leader, State0, []}; [Id, PeerId]), ?WARN("~w saw install_snapshot_result from unknown peer ~w~n", undefined -> case peer(PeerId, State0) of #{id := Id} = State0) ->handle_leader({PeerId, #install_snapshot_result{last_index = LastIndex}}, end; {follower, update_term(Term, State0), []} " abdicates term: ~b!~n", [LogId, Term, CurTerm]), ?DEBUG("~s: leader saw install_snapshot_result for term ~b" _ -> {leader, State0, []};