Packages
ra
0.9.2
3.1.9
3.1.8
3.1.7
3.1.6
3.1.5
3.1.4
3.1.3
3.1.2
3.1.1
3.1.0
3.0.2
3.0.1
3.0.0
3.0.0-beta.1
2.17.3
2.17.2
2.17.1
2.17.0
2.16.13
2.16.12
2.16.11
2.16.10
2.16.9
2.16.8
2.16.7
2.16.6
2.16.5
2.16.4
2.16.3
2.16.2
2.16.1
2.16.0
2.16.0-pre.12
2.16.0-pre.11
2.16.0-pre.10
2.16.0-pre.9
2.16.0-pre.8
2.16.0-pre.7
2.16.0-pre.6
2.16.0-pre.5
2.16.0-pre.4
2.16.0-pre.3
2.16.0-pre.2
2.16.0-pre.1
2.15.4
2.15.3
2.15.2
2.15.1
2.15.0
2.14.0
2.13.6
2.13.5
2.13.4
2.13.3
2.13.2
2.13.1
2.13.0
2.13.0-pre.1
2.12.0
2.11.0
2.11.0-pre.1
2.10.2-pre.2
2.10.2-pre.1
2.10.1
2.10.0
2.10.0-pre.3
2.10.0-pre.2
2.10.0-pre.1
2.9.10-pre.1
2.9.1
2.9.1-pre.2
2.9.1-pre.1
2.9.0
2.8.0
retired
2.7.3
2.7.2
2.7.1
2.7.0
2.7.0-pre.3
2.7.0-pre.2
2.7.0-pre.1
2.6.3
2.6.2
2.6.1
2.6.0-pre.1
2.5.1
2.5.1-pre.1
2.5.0
2.4.9
2.4.8
2.4.7
2.4.6
2.4.5
2.4.4
2.4.3
2.4.2
retired
2.4.1
2.4.0
2.3.0
2.2.0
2.1.0
2.0.13
2.0.12
2.0.11
2.0.10
2.0.9
2.0.8
2.0.7
2.0.6
2.0.5
2.0.4
2.0.3
2.0.2
2.0.1
2.0.0
1.1.9
1.1.8
1.1.7
1.1.6
1.1.5
1.1.4
1.1.3
1.1.2
1.1.1
1.1.0
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.9.6
0.9.5
0.9.4
0.9.2
0.3.3
retired
0.3.2
retired
0.3.1
retired
Raft library
Current section
Files
Jump to
Current section
Files
src/ra_server.erl
%% @hidden
-module(ra_server).
-include("ra.hrl").
-compile(inline_list_funcs).
-export([
name/2,
init/1,
handle_leader/2,
handle_candidate/2,
handle_pre_vote/2,
handle_follower/2,
handle_await_condition/2,
handle_aux/4,
handle_state_enter/2,
tick/1,
overview/1,
is_new/1,
is_fully_persisted/1,
is_fully_replicated/1,
% properties
id/1,
uid/1,
leader_id/1,
machine/1,
current_term/1,
% TODO: hide behind a handle_leader
make_rpcs/1,
update_release_cursor/3,
persist_last_applied/1,
terminate/2,
log_fold/3,
recover/1
]).
-type ra_await_condition_fun() :: fun((ra_msg(), ra_server_state()) -> boolean()).
-type ra_server_state() ::
#{id => ra_server_id(),
uid => ra_uid(),
leader_id => maybe(ra_server_id()),
cluster => ra_cluster(),
cluster_change_permitted => boolean(),
cluster_index_term => ra_idxterm(),
pending_cluster_changes => [term()],
previous_cluster => {ra_index(), ra_term(), ra_cluster()},
current_term => ra_term(),
log => term(),
voted_for => maybe(ra_server_id()), % persistent
votes => non_neg_integer(),
commit_index => ra_index(),
last_applied => ra_index(),
persisted_last_applied => ra_index(),
stop_after => ra_index(),
machine => ra_machine:machine(),
machine_state => term(),
aux_state => term(),
condition => ra_await_condition_fun(),
condition_timeout_effects => [ra_effect()],
pre_vote_token => reference()
}.
-type ra_state() :: leader | follower | candidate
| pre_vote | await_condition | delete_and_terminate
| terminating_leader | terminating_follower | recover
| recovered | stop.
-type command_type() :: '$usr' | '$ra_query' | '$ra_join' | '$ra_leave' |
'$ra_cluster_change' | '$ra_cluster'.
-type command_meta() :: #{from := maybe(from())}.
-type command_correlation() :: integer() | reference().
-type command_reply_mode() :: after_log_append |
await_consensus |
{notify_on_consensus,
command_correlation(), pid()} |
noreply.
-type command() :: {command_type(), command_meta(),
UserCommand :: term(), command_reply_mode()} | noop.
-type ra_msg() :: #append_entries_rpc{} |
{ra_server_id(), #append_entries_reply{}} |
{ra_server_id(), #install_snapshot_result{}} |
#request_vote_rpc{} |
#request_vote_result{} |
#pre_vote_rpc{} |
#pre_vote_result{} |
#install_snapshot_rpc{} |
election_timeout |
await_condition_timeout |
{command, command()} |
{commands, [command()]} |
ra_log:event().
-type ra_reply_body() :: #append_entries_reply{} |
#request_vote_result{} |
#pre_vote_result{}.
-type ra_effect() ::
ra_machine:effect() |
{reply, ra_reply_body()} |
{reply, term(), ra_reply_body()} |
{cast, ra_server_id(), term()} |
{send_vote_requests, [{ra_server_id(),
#request_vote_rpc{} | #pre_vote_rpc{}}]} |
{send_rpcs, [{ra_server_id(), #append_entries_rpc{}}]} |
{next_event, ra_msg()} |
{next_event, cast, ra_msg()} |
{notify, pid(), reference()} |
{incr_metrics, Table :: atom(),
[{Pos :: non_neg_integer(), Incr :: integer()}]}.
-type ra_effects() :: [ra_effect()].
-type simple_apply_fun(State) :: fun((term(), State) -> State).
-type machine_conf() :: {module, module(), InitConfig :: map()} |
{simple, simple_apply_fun(term()),
InitialState :: term()}.
%% The machine configuration.
%% This is how ra knows which module to use to invoke the ra_machine callbacks
%% and the config to pass to the {@link ra_machine:init/1} implementation.
%% The simple machine config is version that can only be used for simple state
%% machines that cannot access any of the advanced features.
-type ra_server_config() :: #{id := ra_server_id(),
uid := ra_uid(),
cluster_name := ra_cluster_name(),
log_init_args := ra_log:ra_log_init_args(),
initial_members := [ra_server_id()],
machine := machine_conf(),
% TODO: review - only really used for
% setting election timeouts
broadcast_time => non_neg_integer(), % ms
% for periodic actions such as sending stale rpcs
% and persisting last_applied index
tick_timeout => non_neg_integer(), % ms
await_condition_timeout => non_neg_integer()}.
-export_type([ra_server_state/0,
ra_state/0,
ra_server_config/0,
ra_msg/0,
machine_conf/0,
command/0,
command_type/0,
command_meta/0,
command_correlation/0,
command_reply_mode/0
]).
-define(AER_CHUNK_SIZE, 25).
-define(FOLD_LOG_BATCH_SIZE, 25).
% TODO: test what is a good defult here
% TODO: make configurable
-define(MAX_PIPELINE_DISTANCE, 10000).
-spec name(ClusterName :: ra_cluster_name(), UniqueSuffix::string()) -> atom().
name(ClusterName, UniqueSuffix) ->
list_to_atom("ra_" ++ ClusterName ++ "_server_" ++ UniqueSuffix).
-spec init(ra_server_config()) -> ra_server_state().
init(#{id := Id,
uid := UId,
cluster_name := _ClusterName,
initial_members := InitialNodes,
log_init_args := LogInitArgs,
machine := MachineConf} = Config) ->
Name = ra_lib:ra_server_id_to_local_name(Id),
Log0 = ra_log:init(LogInitArgs),
ok = ra_log:write_config(Config, Log0),
CurrentTerm = ra_log:read_meta(current_term, Log0, 0),
LastApplied = ra_log:read_meta(last_applied, Log0, 0),
VotedFor = ra_log:read_meta(voted_for, Log0, undefined),
Machine = case MachineConf of
{simple, Fun, S} ->
{machine, ra_machine_simple, #{simple_fun => Fun,
initial_state => S}};
{module, Mod, Args} ->
{machine, Mod, Args}
end,
InitialMachineState = ra_machine:init(Machine, Name),
{FirstIndex, Cluster0, MacState, SnapshotIndexTerm} =
case ra_log:read_snapshot(Log0) of
undefined ->
{0, make_cluster(Id, InitialNodes),
InitialMachineState, {0, 0}};
{Idx, Term, ClusterNodes, MacSt} ->
Clu = make_cluster(Id, ClusterNodes),
%% the snapshot is the last index before the first index
{Idx, Clu, MacSt, {Idx, Term}}
end,
CommitIndex = max(LastApplied, FirstIndex),
State0 = #{id => Id,
uid => UId,
cluster => Cluster0,
% There may be scenarios when a single server
% starts up but hasn't
% yet re-applied its noop command that we may receive other join
% commands that can't be applied.
% TODO: what if we have snapshotted and
% there is no `noop` command
% to be applied in the current term?
cluster_change_permitted => false,
cluster_index_term => {0, 0},
pending_cluster_changes => [],
current_term => CurrentTerm,
voted_for => VotedFor,
commit_index => CommitIndex,
last_applied => FirstIndex,
persisted_last_applied => LastApplied,
log => Log0,
machine => Machine,
machine_state => MacState,
aux_state => ra_machine:init_aux(Machine, Name),
condition_timeout_effects => []},
% Find last cluster change and idxterm and use as initial cluster
% This is required as otherwise a server could restart without any known
% peers and become a leader
{ok, {{ClusterIndexTerm, Cluster}, Log}} =
fold_log_from(CommitIndex,
fun({Idx, Term, {'$ra_cluster_change', _, Cluster, _}}, _) ->
{{Idx, Term}, Cluster};
(_, Acc) ->
Acc
end, {{SnapshotIndexTerm, Cluster0}, Log0}),
% TODO: do we need to set previous cluster here?
State0#{log => Log,
cluster => Cluster,
cluster_index_term => ClusterIndexTerm}.
recover(#{id := Id,
commit_index := CommitIndex,
last_applied := _LastApplied,
log := Log0,
machine := Machine} = State0) ->
?INFO("~w: recovering state machine from ~b to ~b~n",
[Id, _LastApplied, CommitIndex]),
{State, _, _} = apply_to(CommitIndex,
fun(E, S) ->
%% Clear out the effects to avoid building
%% up a long list of effects than then
%% we throw away
%% on server startup (queue recovery)
setelement(4,
apply_with(Machine, E, S), [])
end,
State0),
% close and re-open log to ensure segments aren't unnecessarily kept
% open
Log = ra_log:release_resources(1, Log0),
State#{log => Log}.
% the peer id in the append_entries_reply message is an artifact of
% the "fake" rpc call in ra_proxy as when using reply the unique reference
% is joined with the msg itself. In this instance it is treated as an info
% message.
-spec handle_leader(ra_msg(), ra_server_state()) ->
{ra_state(), ra_server_state(), ra_effects()}.
handle_leader({PeerId, #append_entries_reply{term = Term, success = true,
next_index = NextIdx,
last_index = LastIdx}},
State0 = #{current_term := Term, id := Id}) ->
case peer(PeerId, State0) of
undefined ->
?WARN("~w saw append_entries_reply from unknown peer ~w~n",
[Id, PeerId]),
{leader, State0, []};
Peer0 = #{match_index := MI, next_index := NI} ->
Peer = Peer0#{match_index => max(MI, LastIdx),
next_index => max(NI, NextIdx)},
State1 = update_peer(PeerId, Peer, State0),
{State2, Effects0, Applied} = evaluate_quorum(State1),
{State, Rpcs} = make_pipelined_rpcs(State2),
% rpcs need to be issued _AFTER_ machine effects or there is
% a chance that effects will never be issued if the leader crashes
% after sending rpcs but before actioning the machine effects
Effects = Effects0 ++ [{send_rpcs, Rpcs},
{incr_metrics, ra_metrics, [{3, Applied}]}],
case State of
#{id := Id, cluster := #{Id := _}} ->
% leader is in the cluster
{leader, State, Effects};
#{commit_index := CI, cluster_index_term := {CITIndex, _}}
when CI >= CITIndex ->
% leader is not in the cluster and the new cluster
% config has been committed
% time to say goodbye
?INFO("~w leader not in new cluster - goodbye", [Id]),
{stop, State, Effects};
_ ->
{leader, State, Effects}
end
end;
handle_leader({PeerId, #append_entries_reply{term = Term}},
#{current_term := CurTerm,
id := Id} = State0) when Term > CurTerm ->
case peer(PeerId, State0) of
undefined ->
?WARN("~w saw append_entries_reply from unknown peer ~w~n",
[Id, PeerId]),
{leader, State0, []};
_ ->
?INFO("~w leader saw append_entries_reply for term ~b "
"abdicates term: ~b!~n",
[Id, Term, CurTerm]),
{follower, update_term(Term, State0), []}
end;
handle_leader({PeerId, #append_entries_reply{success = false,
next_index = NextIdx,
last_index = LastIdx,
last_term = LastTerm}},
State0 = #{id := Id, cluster := Nodes, log := Log0}) ->
#{PeerId := Peer0 = #{match_index := MI,
next_index := NI}} = Nodes,
% if the last_index exists and has a matching term we can forward
% match_index and update next_index directly
{Peer, Log} = case ra_log:fetch_term(LastIdx, Log0) of
{undefined, L} ->
% entry was not found - simply set next index to
?INFO("~w: setting next index for ~w ~b",
[Id, PeerId, NextIdx]),
{Peer0#{match_index => LastIdx,
next_index => NextIdx}, L};
% entry exists we can forward
{LastTerm, L} when LastIdx >= MI ->
?INFO("~w: setting last index to ~b, next_index ~b"
" for ~w", [Id, LastIdx, NextIdx, PeerId]),
{Peer0#{match_index => LastIdx,
next_index => NextIdx}, L};
{_Term, L} when LastIdx < MI ->
% TODO: this can only really happen when peers are
% non-persistent.
% should they turn-into non-voters when this sitution
% is detected
?ERR("~w: leader saw peer with last_index [~b in ~b]"
" lower than recorded match index [~b]."
"Resetting peer's state to last_index.~n",
[Id, LastIdx, LastTerm, MI]),
{Peer0#{match_index => LastIdx,
next_index => LastIdx + 1}, L};
{_EntryTerm, L} ->
?INFO("~w: leader received last_index ~b"
" from ~w with "
"term ~b different term ~b~n",
[Id, LastIdx, PeerId, LastTerm, _EntryTerm]),
% last_index has a different term or entry does not
% exist
% The peer must have received an entry from a previous
% leader and the current leader wrote a different
% entry at the same index in a different term.
% decrement next_index but don't go lower than
% match index.
{Peer0#{next_index => max(min(NI-1, LastIdx), MI)}, L}
end,
State1 = State0#{cluster => Nodes#{PeerId => Peer}, log => Log},
{State, Rpcs} = make_pipelined_rpcs(State1),
{leader, State, [{send_rpcs, Rpcs}]};
handle_leader({command, Cmd}, State00 = #{id := Id}) ->
case append_log_leader(Cmd, State00) of
{not_appended, State = #{cluster_change_permitted := CCP}} ->
?WARN("~w command ~W NOT appended to log, "
"cluster_change_permitted ~w~n", [Id, Cmd, 5, CCP]),
{leader, State, []};
{ok, Idx, Term, State0} ->
% ?INFO("~p ~p command appended to log at ~p term ~p~n",
% [Id, Cmd, Idx, Term]),
% Only "pipeline" in response to a command
{State, Rpcs} = make_pipelined_rpcs(State0),
Effects1 = [{send_rpcs, Rpcs},
{incr_metrics, ra_metrics, [{2, 1}]}],
% check if a reply is required.
% TODO: refactor - can this be made a bit nicer/more explicit?
Effects = case Cmd of
{_, _, _, await_consensus} ->
Effects1;
{_, #{from := undefined}, _, _} ->
Effects1;
{_, #{from := From}, _, _} ->
[{reply, From, {Idx, Term}} | Effects1];
_ ->
Effects1
end,
{leader, State, Effects}
end;
handle_leader({commands, Cmds}, State00 = #{id := _Id}) ->
{State0, Effects0} =
lists:foldl( fun(C, {S0, E}) ->
{ok, I, T, S} = append_log_leader(C, S0),
case C of
{_, #{from := From}, _, after_log_append} ->
{S, [{reply, From , {I, T}} | E]};
_ ->
{S, E}
end
end, {State00, []}, Cmds),
{State, Rpcs} = make_pipelined_rpcs(length(Cmds), State0),
%% TOOD: ra_metrics
{leader, State, [{send_rpcs, Rpcs} | Effects0]};
handle_leader({ra_log_event, {written, _} = Evt}, State0 = #{log := Log0}) ->
Log = ra_log:handle_event(Evt, Log0),
{State1, Effects, Applied} = evaluate_quorum(State0#{log => Log}),
{State, Rpcs} = make_pipelined_rpcs(State1),
{leader, State, [{send_rpcs, Rpcs},
{incr_metrics, ra_metrics, [{3, Applied}]} | Effects]};
handle_leader({ra_log_event, Evt}, State = #{log := Log0}) ->
% simply forward all other events to ra_log
{leader, State#{log => ra_log:handle_event(Evt, Log0)}, []};
handle_leader({aux_command, Type, Cmd}, State0) ->
handle_aux(leader, Type, Cmd, State0);
handle_leader({PeerId, #install_snapshot_result{term = Term}},
#{id := Id, current_term := CurTerm} = State0)
when Term > CurTerm ->
case peer(PeerId, State0) of
undefined ->
?WARN("~w: saw install_snapshot_result from unknown peer ~w~n",
[Id, PeerId]),
{leader, State0, []};
_ ->
?INFO("~w: leader saw install_snapshot_result for term ~b"
" abdicates term: ~b!~n", [Id, Term, CurTerm]),
{follower, update_term(Term, State0), []}
end;
handle_leader({PeerId, #install_snapshot_result{last_index = LastIndex}},
#{id := Id} = State0) ->
case peer(PeerId, State0) of
undefined ->
?WARN("~w saw install_snapshot_result from unknown peer ~w~n",
[Id, PeerId]),
{leader, State0, []};
Peer0 = #{next_index := NI} ->
State1 = update_peer(PeerId,
Peer0#{match_index => LastIndex,
commit_index_sent => LastIndex,
% leader might have pipelined
% append entries
% since snapshot was sent
% need to ensure next index is at least
% LastIndex + 1 though
next_index => max(NI, LastIndex+1) },
State0),
{State, Rpcs} = make_pipelined_rpcs(State1),
Effects = [{send_rpcs, Rpcs}],
{leader, State, Effects}
end;
handle_leader(#install_snapshot_rpc{term = Term,
leader_id = Leader} = Evt,
#{current_term := CurTerm,
id := Id} = State0) when Term > CurTerm ->
case peer(Leader, State0) of
undefined ->
?WARN("~w saw install_snapshot_rpc from unknown leader ~w~n",
[Id, Leader]),
{leader, State0, []};
_ ->
?INFO("~w leader saw install_snapshot_rpc for term ~b "
"abdicates term: ~b!~n",
[Id, Term, CurTerm]),
{follower, update_term(Term, State0), [{next_event, Evt}]}
end;
handle_leader(#append_entries_rpc{term = Term} = Msg,
#{current_term := CurTerm} = State0) when Term > CurTerm ->
?INFO("~w leader saw append_entries_rpc for term ~b abdicates term: ~b!~n",
[id(State0), Term, CurTerm]),
{follower, update_term(Term, State0), [{next_event, Msg}]};
handle_leader(#append_entries_rpc{term = Term}, #{current_term := Term,
id := Id}) ->
?ERR("~w leader saw append_entries_rpc for same term ~b"
" this should not happen!~n", [Id, Term]),
exit(leader_saw_append_entries_rpc_in_same_term);
% TODO: reply to append_entries_rpcs that have lower term?
handle_leader(#request_vote_rpc{term = Term, candidate_id = Cand} = Msg,
#{current_term := CurTerm,
id := Id} = State0) when Term > CurTerm ->
case peer(Cand, State0) of
undefined ->
?WARN("~w leader saw request_vote_rpc for unknown peer ~w~n",
[Id, Cand]),
{leader, State0, []};
_ ->
?INFO("~w leader saw request_vote_rpc for term ~b"
" abdicates term: ~b!~n", [Id, Term, CurTerm]),
{follower, update_term(Term, State0), [{next_event, Msg}]}
end;
handle_leader(#request_vote_rpc{}, State = #{current_term := Term}) ->
Reply = #request_vote_result{term = Term, vote_granted = false},
{leader, State, [{reply, Reply}]};
handle_leader(#pre_vote_rpc{term = Term, candidate_id = Cand} = Msg,
#{current_term := CurTerm,
id := Id} = State0) when Term > CurTerm ->
case peer(Cand, State0) of
undefined ->
?WARN("~w leader saw pre_vote_rpc for unknown peer ~w~n",
[Id, Cand]),
{leader, State0, []};
_ ->
?INFO("~w leader saw pre_vote_rpc for term ~b"
" abdicates term: ~b!~n", [Id, Term, CurTerm]),
{follower, update_term(Term, State0), [{next_event, Msg}]}
end;
handle_leader(#pre_vote_rpc{term = Term},
#{current_term := CurTerm} = State0)
when Term =< CurTerm ->
% enforce leadership
{State, Rpcs} = make_all_rpcs(State0),
Effects = [{send_rpcs, Rpcs}],
{leader, State, Effects};
handle_leader(Msg, State) ->
log_unhandled_msg(leader, Msg, State),
{leader, State, []}.
-spec handle_candidate(ra_msg() | election_timeout, ra_server_state()) ->
{ra_state(), ra_server_state(), ra_effects()}.
handle_candidate(#request_vote_result{term = Term, vote_granted = true},
#{current_term := Term, votes := Votes,
cluster := Nodes} = State0) ->
NewVotes = Votes + 1,
case trunc(maps:size(Nodes) / 2) + 1 of
NewVotes ->
{State, Rpcs} = make_all_rpcs(initialise_peers(State0)),
% Effects = ra_machine:leader_effects(Machine, MacState),
{leader, maps:without([votes, leader_id], State),
[{send_rpcs, Rpcs},
{next_event, cast, {command, noop}}]};
_ ->
{candidate, State0#{votes => NewVotes}, []}
end;
handle_candidate(#request_vote_result{term = Term},
#{current_term := CurTerm} = State0)
when Term > CurTerm ->
?INFO("~w: candidate request_vote_result with higher term"
" received ~b -> ~b", [id(State0), CurTerm, Term]),
State = update_meta(Term, undefined, State0),
{follower, State, []};
handle_candidate(#request_vote_result{vote_granted = false}, State) ->
{candidate, State, []};
handle_candidate(#append_entries_rpc{term = Term} = Msg,
#{current_term := CurTerm} = State0) when Term >= CurTerm ->
State = update_meta(Term, undefined, State0),
{follower, State, [{next_event, Msg}]};
handle_candidate(#append_entries_rpc{leader_id = LeaderId},
#{current_term := CurTerm} = State) ->
% term must be older return success=false
Reply = append_entries_reply(CurTerm, false, State),
{candidate, State, [{cast, LeaderId, {id(State), Reply}}]};
handle_candidate({_PeerId, #append_entries_reply{term = Term}},
#{current_term := CurTerm} = State0)
when Term > CurTerm ->
?INFO("~w: candidate append_entries_reply with higher"
" term received ~b -> ~b~n",
[id(State0), CurTerm, Term]),
State = update_meta(Term, undefined, State0),
{follower, State, []};
handle_candidate(#request_vote_rpc{term = Term} = Msg,
#{current_term := CurTerm} = State0)
when Term > CurTerm ->
?INFO("~w: candidate request_vote_rpc with higher term received ~b -> ~b~n",
[id(State0), CurTerm, Term]),
State = update_meta(Term, undefined, State0),
{follower, State, [{next_event, Msg}]};
handle_candidate(#request_vote_rpc{}, State = #{current_term := Term}) ->
Reply = #request_vote_result{term = Term, vote_granted = false},
{candidate, State, [{reply, Reply}]};
handle_candidate(election_timeout, State) ->
call_for_election(candidate, State);
handle_candidate(Msg, State) ->
log_unhandled_msg(candidate, Msg, State),
{candidate, State, []}.
-spec handle_pre_vote(ra_msg(), ra_server_state()) ->
{ra_state(), ra_server_state(), ra_effects()}.
handle_pre_vote(#append_entries_rpc{term = Term} = Msg,
#{current_term := CurTerm} = State0)
when Term >= CurTerm ->
State = update_term(Term, State0),
% revert to follower state
{follower, State#{votes => 0}, [{next_event, Msg}]};
handle_pre_vote(#request_vote_rpc{term = Term} = Msg,
#{current_term := CurTerm} = State0)
when Term > CurTerm ->
State = update_term(Term, State0),
% revert to follower state
{follower, State#{votes => 0}, [{next_event, Msg}]};
handle_pre_vote(#pre_vote_result{term = Term},
#{current_term := CurTerm} = State0)
when Term > CurTerm ->
% higher term always reverts?
State = update_term(Term, State0),
{follower, State#{votes => 0}, []};
handle_pre_vote(#install_snapshot_rpc{term = Term} = ISR,
#{current_term := CurTerm} = State0)
when Term >= CurTerm ->
{follower, State0#{votes => 0}, [{next_event, ISR}]};
handle_pre_vote(#pre_vote_result{term = Term, vote_granted = true,
token = Token},
#{current_term := Term, votes := Votes,
pre_vote_token := Token,
cluster := Nodes} = State0) ->
NewVotes = Votes + 1,
State = update_term(Term, State0),
case trunc(maps:size(Nodes) / 2) + 1 of
NewVotes ->
call_for_election(candidate, State);
_ ->
{pre_vote, State#{votes => NewVotes}, []}
end;
handle_pre_vote(#pre_vote_rpc{} = PreVote, State) ->
process_pre_vote(pre_vote, PreVote, State);
handle_pre_vote(election_timeout, State) ->
call_for_election(pre_vote, State);
handle_pre_vote(Msg, State) ->
log_unhandled_msg(pre_vote, Msg, State),
{pre_vote, State, []}.
-spec handle_follower(ra_msg(), ra_server_state()) ->
{ra_state(), ra_server_state(), ra_effects()}.
handle_follower(#append_entries_rpc{term = Term,
leader_id = LeaderId,
leader_commit = LeaderCommit,
prev_log_index = PLIdx,
prev_log_term = PLTerm,
entries = Entries0},
State000 = #{id := Id, log := Log0, current_term := CurTerm})
when Term >= CurTerm ->
State00 = update_term(Term, State000),
case has_log_entry_or_snapshot(PLIdx, PLTerm, State00) of
{entry_ok, State0} ->
% filter entries already seen
{Log1, Entries} = drop_existing({Log0, Entries0}),
case Entries of
[] ->
LastIdx = ra_log:last_index_term(Log1),
Log2 = case Entries0 of
[] when element(1, LastIdx) > PLIdx ->
%% if no entries were sent we need to reset
%% last index to match the leader
?INFO("ra: resetting last index to ~b~n",
[PLIdx]),
{ok, L} = ra_log:set_last_index(PLIdx, Log1),
L;
_ ->
Log1
end,
% update commit index to be the min of the last
% entry seen (but not necessarily written)
% and the leader commit
{Idx, _} = ra_log:last_index_term(Log2),
State1 = State0#{commit_index => min(Idx, LeaderCommit),
log => Log2,
leader_id => LeaderId},
% evaluate commit index as we may have received an updated
% commit index for previously written entries
evaluate_commit_index_follower(State1);
[{FirstIdx, _, _} | _] -> % FirstTerm
{LastIdx, State1} = lists:foldl(
fun pre_append_log_follower/2,
{FirstIdx, State0}, Entries),
% Increment only commit_index here as we are not applying
% anything at this point.
% last_applied will be incremented when the written event is
% processed
State = State1#{commit_index => min(LeaderCommit, LastIdx),
leader_id => LeaderId},
case ra_log:write(Entries, Log1) of
{ok, Log} ->
{follower, State#{log => Log}, []};
{error, wal_down} ->
{await_condition,
State#{condition => fun wal_down_condition/2}, []};
{error, _} = Err ->
exit(Err)
end
end;
{missing, State0} ->
Reply = append_entries_reply(Term, false, State0),
?INFO("~w: follower did not have entry at ~b in ~b."
" Requesting from ~b~n",
[Id, PLIdx, PLTerm, Reply#append_entries_reply.next_index]),
Effects = [cast_reply(Id, LeaderId, Reply)],
{await_condition,
State0#{leader_id => LeaderId,
condition => fun follower_catchup_cond/2,
% repeat reply effect on condition timeout
condition_timeout_effects => Effects}, Effects};
{term_mismatch, OtherTerm, State0} ->
CommitIndex = maps:get(commit_index, State0),
?INFO("~w: term mismatch - follower had entry at ~b with term ~b "
"but not with term ~b~n"
"Asking leader to resend from ~b~n",
[Id, PLIdx, OtherTerm, PLTerm, CommitIndex + 1]),
% This situation arises when a minority leader replicates entries
% that it cannot commit then gets replaced by a majority leader
% that also has made progress
% As the follower is responsible for telling the leader
% which their next expected entry is the best we can do here
% is rewind back and use the commit index as the last index
% and commit_index + 1 as the next expected.
% This _may_ overwrite some valid entries but is probably the
% simplest way to proceed
{Reply, State} = mismatch_append_entries_reply(Term, CommitIndex,
State0),
Effects = [cast_reply(Id, LeaderId, Reply)],
{await_condition,
State#{leader_id => LeaderId,
condition => fun follower_catchup_cond/2,
% repeat reply effect on condition timeout
condition_timeout_effects => Effects}, Effects}
end;
handle_follower(#append_entries_rpc{term = _Term, leader_id = LeaderId},
#{id := Id, current_term := CurTerm} = State) ->
% the term is lower than current term
Reply = append_entries_reply(CurTerm, false, State),
?INFO("~w: follower request_vote_rpc in ~b but current term ~b~n",
[Id, _Term, CurTerm]),
{follower, State, [cast_reply(Id, LeaderId, Reply)]};
handle_follower({ra_log_event, {written, _} = Evt},
State00 = #{log := Log0}) ->
State0 = State00#{log => ra_log:handle_event(Evt, Log0)},
evaluate_commit_index_follower(State0);
handle_follower({ra_log_event, Evt}, State = #{log := Log0}) ->
% simply forward all other events to ra_log
{follower, State#{log => ra_log:handle_event(Evt, Log0)}, []};
handle_follower(#pre_vote_rpc{} = PreVote, State) ->
process_pre_vote(follower, PreVote, State);
handle_follower(#request_vote_rpc{candidate_id = Cand, term = Term},
#{current_term := Term, voted_for := VotedFor} = State)
when VotedFor /= undefined andalso VotedFor /= Cand ->
% already voted for another in this term
?INFO("~w: follower request_vote_rpc for ~w already voted for ~w in ~b",
[id(State), Cand, VotedFor, Term]),
Reply = #request_vote_result{term = Term, vote_granted = false},
{follower, State, [{reply, Reply}]};
handle_follower(#request_vote_rpc{term = Term, candidate_id = Cand,
last_log_index = LLIdx,
last_log_term = LLTerm},
#{current_term := CurTerm} = State0)
when Term >= CurTerm ->
State = update_term(Term, State0),
LastIdxTerm = last_idx_term(State),
case is_candidate_log_up_to_date(LLIdx, LLTerm, LastIdxTerm) of
true ->
?INFO("~w: granting vote for ~w with last indexterm ~w"
"for term ~b previous term was ~b~n",
[id(State0), Cand, {LLIdx, LLTerm}, Term, CurTerm]),
Reply = #request_vote_result{term = Term, vote_granted = true},
{follower, State#{voted_for => Cand, current_term => Term},
[{reply, Reply}]};
false ->
?INFO("~w: declining vote for ~w for term ~b,"
" candidate last log index term was: ~w~n"
" last log entry idxterm seen was: ~w~n",
[id(State0), Cand, Term, {LLIdx, LLTerm}, {LastIdxTerm}]),
Reply = #request_vote_result{term = Term, vote_granted = false},
{follower, State#{current_term => Term}, [{reply, Reply}]}
end;
handle_follower(#request_vote_rpc{term = Term, candidate_id = _Cand},
State = #{current_term := CurTerm})
when Term < CurTerm ->
?INFO("~w declining vote to ~w for term ~b, current term ~b~n",
[id(State), _Cand, Term, CurTerm]),
Reply = #request_vote_result{term = CurTerm, vote_granted = false},
{follower, State, [{reply, Reply}]};
handle_follower({_PeerId, #append_entries_reply{term = Term}},
State = #{current_term := CurTerm}) when Term > CurTerm ->
{follower, update_term(Term, State), []};
handle_follower(#install_snapshot_rpc{term = Term,
leader_id = LeaderId,
last_index = LastIndex,
last_term = LastTerm},
State = #{id := Id, current_term := CurTerm})
when Term < CurTerm ->
?INFO("~w: install_snapshot old term ~b in ~b~n",
[Id, LastIndex, LastTerm]),
% follower receives a snapshot from an old term
Reply = #install_snapshot_result{term = CurTerm,
last_term = LastTerm,
last_index = LastIndex},
{follower, State, [cast_reply(Id, LeaderId, Reply)]};
handle_follower(#install_snapshot_rpc{term = Term,
leader_id = LeaderId,
last_term = LastTerm,
last_index = LastIndex,
last_config = ClusterNodes,
data = Data},
State0 = #{id := Id, log := Log0,
current_term := CurTerm})
when Term >= CurTerm ->
?INFO("~w: installing snapshot at index ~b in term ~b~n",
[Id, LastIndex, LastTerm]),
% follower receives a snapshot to be installed
Log = ra_log:install_snapshot({LastIndex, LastTerm, ClusterNodes, Data}, Log0),
State1 = State0#{log => Log,
current_term => Term,
commit_index => LastIndex,
last_applied => LastIndex,
cluster => make_cluster(Id, ClusterNodes),
machine_state => Data,
leader_id => LeaderId},
State = persist_last_applied(State1),
Reply = #install_snapshot_result{term = CurTerm,
last_term = LastTerm,
last_index = LastIndex},
{follower, State, [cast_reply(Id, LeaderId, Reply)]};
handle_follower(election_timeout, State) ->
call_for_election(pre_vote, State);
handle_follower(Msg, State) ->
log_unhandled_msg(follower, Msg, State),
{follower, State, []}.
-spec handle_await_condition(ra_msg(), ra_server_state()) ->
{ra_state(), ra_server_state(), ra_effects()}.
handle_await_condition(#request_vote_rpc{} = Msg, State) ->
{follower, State, [{next_event, cast, Msg}]};
handle_await_condition(election_timeout, State) ->
call_for_election(pre_vote, State);
handle_await_condition(await_condition_timeout,
#{condition_timeout_effects := Effects} = State) ->
{follower, State#{condition_timeout_effects => []}, Effects};
handle_await_condition(Msg, #{condition := Cond} = State) ->
case Cond(Msg, State) of
true ->
{follower, State, [{next_event, cast, Msg}]};
false ->
% log_unhandled_msg(await_condition, Msg, State),
{await_condition, State, []}
end.
-spec tick(ra_server_state()) -> ra_effects().
tick(#{machine := Machine, machine_state := MacState}) ->
Now = os:system_time(millisecond),
ra_machine:tick(Machine, Now, MacState).
-spec handle_state_enter(ra_state() | eol, ra_server_state()) ->
{ra_server_state() | eol, ra_effects()}.
handle_state_enter(RaftState, #{machine := Machine,
machine_state := MacState} = State) ->
{become(RaftState, State),
ra_machine:state_enter(Machine, RaftState, MacState)}.
-spec overview(ra_server_state()) -> map().
overview(#{log := Log, machine := Machine,
machine_state := MacState} = State) ->
O = maps:with([uid, current_term, commit_index, last_applied,
cluster, leader_id, voted_for], State),
LogOverview = ra_log:overview(Log),
MacOverview = ra_machine:overview(Machine, MacState),
O#{log => LogOverview,
machine => MacOverview}.
-spec is_new(ra_server_state()) -> boolean().
is_new(#{log := Log}) ->
ra_log:next_index(Log) =:= 1.
-spec is_fully_persisted(ra_server_state()) -> boolean().
is_fully_persisted(#{log := Log}) ->
LastWritten = ra_log:last_written(Log),
LastIdxTerm = ra_log:last_index_term(Log),
LastWritten =:= LastIdxTerm.
-spec is_fully_replicated(ra_server_state()) -> boolean().
is_fully_replicated(#{commit_index := CI} = State) ->
case maps:values(peers(State)) of
[] -> true; % there is only one server
Peers ->
MinMI = lists:min([M || #{match_index := M} <- Peers]),
MinMI >= CI
end.
handle_aux(RaftState, Type, Cmd, #{aux_state := Aux0, log := Log0,
machine := Machine,
machine_state := MacState0} = State0) ->
case ra_machine:handle_aux(Machine, RaftState, Type, Cmd, Aux0,
Log0, MacState0) of
{reply, Reply, Aux, Log} ->
{RaftState, State0#{log => Log, aux_state => Aux},
[{reply, Reply}]};
{no_reply, Aux, Log} ->
{RaftState, State0#{log => Log, aux_state => Aux}, []};
undefined ->
{RaftState, State0, []}
end.
% property helpers
-spec id(ra_server_state()) -> ra_server_id().
id(#{id := Id}) -> Id.
-spec uid(ra_server_state()) -> ra_uid().
uid(#{uid := UId}) -> UId.
-spec machine(ra_server_state()) -> module().
machine(#{machine := Machine}) ->
ra_machine:module(Machine).
-spec leader_id(ra_server_state()) -> maybe(ra_server_id()).
leader_id(State) ->
maps:get(leader_id, State, undefined).
-spec current_term(ra_server_state()) -> maybe(ra_term()).
current_term(State) ->
maps:get(current_term, State).
% Internal
become(leader, #{cluster := Cluster, log := Log0} = State) ->
Log = ra_log:release_resources(maps:size(Cluster), Log0),
State#{log => Log};
become(follower, #{log := Log0} = State) ->
%% followers should only ever need a single segment open at any one
%% time
State#{log => ra_log:release_resources(1, Log0)};
become(_RaftState, State) ->
State.
follower_catchup_cond(#append_entries_rpc{term = Term,
prev_log_index = PLIdx,
prev_log_term = PLTerm},
State0 = #{current_term := CurTerm})
when Term >= CurTerm ->
case has_log_entry_or_snapshot(PLIdx, PLTerm, State0) of
{entry_ok, _State} ->
true;
_ ->
false
end;
follower_catchup_cond(#install_snapshot_rpc{term = Term,
last_index = PLIdx},
#{current_term := CurTerm,
log := Log})
when Term >= CurTerm ->
% term is ok - check if the snapshot index is greater than the last
% index seen
PLIdx >= ra_log:next_index(Log);
follower_catchup_cond(_Msg, _State) ->
false.
wal_down_condition(_Msg, #{log := Log}) ->
ra_log:can_write(Log).
evaluate_commit_index_follower(#{commit_index := CommitIndex,
id := Id, leader_id := LeaderId,
current_term := Term,
log := Log} = State0)
when LeaderId =/= undefined ->
% as writes are async we can't use the index of the last available entry
% in the log as they may not have been fully persisted yet
% Take the smaller of the two values as commit index may be higher
% than the last entry received
{Idx, _} = ra_log:last_written(Log),
EffectiveCommitIndex = min(Idx, CommitIndex),
% neet catch termination throw
case catch apply_to(EffectiveCommitIndex, State0) of
{delete_and_terminate, State1, Effects} ->
Reply = append_entries_reply(Term, true, State1),
{delete_and_terminate, State1,
[cast_reply(Id, LeaderId, Reply) |
filter_follower_effects(Effects)]};
{State, Effects0, Applied} ->
% filter the effects that should be applied on a follower
Effects = filter_follower_effects(Effects0),
Reply = append_entries_reply(Term, true, State),
{follower, State, [cast_reply(Id, LeaderId, Reply),
{incr_metrics, ra_metrics, [{3, Applied}]}
| Effects]}
end;
evaluate_commit_index_follower(State) ->
%% when no leader is known
{follower, State, []}.
filter_follower_effects(Effects) ->
lists:filter(fun ({release_cursor, _, _}) -> true;
({incr_metrics, _, _}) -> true;
({aux, _}) -> true;
(garbage_collection) -> true;
(_) -> false
end, Effects).
make_pipelined_rpcs(State) ->
make_pipelined_rpcs(?AER_CHUNK_SIZE, State).
make_pipelined_rpcs(MaxBatchSize, #{commit_index := CommitIndex} = State0) ->
maps:fold(fun(PeerId, Peer0 = #{next_index := Next}, {S0, Entries}) ->
{LastIdx, Entry, S} =
append_entries_or_snapshot(PeerId, Next,
MaxBatchSize, S0),
Peer = Peer0#{next_index => LastIdx+1,
commit_index_sent => CommitIndex},
{update_peer(PeerId, Peer, S), [Entry | Entries]}
end, {State0, []}, pipelineable_peers(State0)).
make_rpcs(State) ->
make_rpcs_for(stale_peers(State), State).
% makes empty append entries for peers that aren't pipelineable
make_all_rpcs(State) ->
make_rpcs_for(peers(State), State).
make_rpcs_for(Peers, State) ->
maps:fold(fun(PeerId, #{next_index := Next}, {S0, Entries}) ->
{_, Entry, S} =
append_entries_or_snapshot(PeerId, Next,
?AER_CHUNK_SIZE,
S0),
{S, [Entry | Entries]}
end, {State, []}, Peers).
append_entries_or_snapshot(PeerId, Next, MaxBatchSize,
#{id := Id, log := Log0,
current_term := Term} = State) ->
PrevIdx = Next - 1,
case ra_log:fetch_term(PrevIdx, Log0) of
{PrevTerm, Log} when PrevTerm =/= undefined ->
make_aer_chunk(PeerId, PrevIdx, PrevTerm, MaxBatchSize,
State#{log => Log});
{undefined, Log} ->
% The assumption here is that a missing entry means we need
% to send a snapshot.
case ra_log:snapshot_index_term(Log) of
{PrevIdx, PrevTerm} ->
% Previous index is the same as snapshot index
make_aer_chunk(PeerId, PrevIdx, PrevTerm, MaxBatchSize,
State#{log => Log});
_ ->
{LastIndex, LastTerm, Config, MacState} =
ra_log:read_snapshot(Log),
{LastIndex,
{PeerId, #install_snapshot_rpc{term = Term,
leader_id = Id,
last_index = LastIndex,
last_term = LastTerm,
last_config = Config,
data = MacState}},
State#{log => Log}}
end
end.
make_aer_chunk(PeerId, PrevIdx, PrevTerm, Num,
#{log := Log0, current_term := Term, id := Id,
commit_index := CommitIndex} = State) ->
Next = PrevIdx + 1,
%% TODO: refactor to avoid lists:last call later
%% ra_log:take should be able to return the actual number of entries
%% read at fixed cost
{Entries, Log} = ra_log:take(Next, Num, Log0),
LastIndex = case Entries of
[] -> PrevIdx;
_ ->
{LastIdx, _, _} = lists:last(Entries),
%% assertion
{Next, _, _} = hd(Entries),
LastIdx
end,
{LastIndex,
{PeerId, #append_entries_rpc{entries = Entries,
term = Term,
leader_id = Id,
prev_log_index = PrevIdx,
prev_log_term = PrevTerm,
leader_commit = CommitIndex}},
State#{log => Log}}.
% stores the cluster config at an index such that we can later snapshot
% at this index.
-spec update_release_cursor(ra_index(), term(), ra_server_state()) ->
ra_server_state().
update_release_cursor(Index, MacState,
State = #{log := Log0, cluster := Cluster}) ->
% simply pass on release cursor index to log
Log = ra_log:update_release_cursor(Index, Cluster, MacState, Log0),
State#{log => Log}.
% Persist last_applied - as there is an inherent race we cannot
% always guarantee that side effects won't be re-issued when a
% follower that has seen an entry but not the commit_index
% takes over and this
% This is done on a schedule
-spec persist_last_applied(ra_server_state()) -> ra_server_state().
persist_last_applied(#{persisted_last_applied := L,
last_applied := L} = State) ->
% do nothing
State;
persist_last_applied(#{last_applied := L, log := Log0} = State) ->
ok = ra_log:write_meta(last_applied, L, Log0, false),
State#{persisted_last_applied => L}.
-spec terminate(ra_server_state(), Reason :: {shutdown, delete} | term()) -> ok.
terminate(#{log := Log} = _State, {shutdown, delete}) ->
?INFO("~w: terminating and deleting all data~n", [id(_State)]),
catch ra_log:delete_everything(Log),
ok;
terminate(State, _Reason) ->
#{log := Log} = persist_last_applied(State),
catch ra_log:close(Log),
ok.
-spec log_fold(ra_server_state(), fun((term(), State) -> State), State) ->
{ok, State, ra_server_state()} |
{error, term(), ra_server_state()}.
log_fold(#{log := Log} = RaState, Fun, State) ->
Idx = case ra_log:snapshot_index_term(Log) of
{PrevIdx, _PrevTerm} ->
PrevIdx;
undefined ->
1
end,
case fold_log_from(Idx, Fun, {State, Log}) of
{ok, {State1, Log1}} ->
{ok, State1, RaState#{log => Log1}};
{error, Reason, Log1} ->
{error, Reason, RaState#{log => Log1}}
end.
%%%===================================================================
%%% Internal functions
%%%===================================================================
call_for_election(candidate, #{id := Id,
current_term := CurrentTerm} = State0) ->
NewTerm = CurrentTerm + 1,
?INFO("~w: election called for in term ~b~n", [Id, NewTerm]),
PeerIds = peer_ids(State0),
% increment current term
{LastIdx, LastTerm} = last_idx_term(State0),
Reqs = [{PeerId, #request_vote_rpc{term = NewTerm,
candidate_id = Id,
last_log_index = LastIdx,
last_log_term = LastTerm}}
|| PeerId <- PeerIds],
% vote for self
VoteForSelf = #request_vote_result{term = NewTerm, vote_granted = true},
State = update_meta(NewTerm, Id, State0),
{candidate, State#{leader_id => undefined, votes => 0},
[{next_event, cast, VoteForSelf}, {send_vote_requests, Reqs}]};
call_for_election(pre_vote, #{id := Id,
current_term := Term} = State0) ->
?INFO("~w: pre_vote election called for in term ~b~n", [Id, Term]),
Token = make_ref(),
PeerIds = peer_ids(State0),
{LastIdx, LastTerm} = last_idx_term(State0),
Reqs = [{PeerId, #pre_vote_rpc{term = Term,
token = Token,
candidate_id = Id,
last_log_index = LastIdx,
last_log_term = LastTerm}}
|| PeerId <- PeerIds],
% vote for self
VoteForSelf = #pre_vote_result{term = Term, token = Token,
vote_granted = true},
State = update_meta(Term, Id, State0),
{pre_vote, State#{leader_id => undefined, votes => 0,
pre_vote_token => Token},
[{next_event, cast, VoteForSelf}, {send_vote_requests, Reqs}]}.
process_pre_vote(FsmState, #pre_vote_rpc{term = Term, candidate_id = Cand,
version = Version,
token = Token,
last_log_index = LLIdx,
last_log_term = LLTerm},
#{current_term := CurTerm}= State0)
when Term >= CurTerm ->
State = update_term(Term, State0),
LastIdxTerm = last_idx_term(State),
case is_candidate_log_up_to_date(LLIdx, LLTerm, LastIdxTerm) of
true when Version =< ?RA_PROTO_VERSION ->
?INFO("~w: granting pre-vote for ~w with last indexterm ~w"
"for term ~b previous term was ~b~n",
[id(State0), Cand, {LLIdx, LLTerm}, Term, CurTerm]),
{FsmState, State#{voted_for => Cand},
[{reply, pre_vote_result(Term, Token, true)}]};
true ->
?INFO("~w: declining pre-vote for ~w for protocol version ~b~n",
[id(State0), Cand, Version]),
{FsmState, State, [{reply, pre_vote_result(Term, Token, false)}]};
false ->
?INFO("~w: declining pre-vote for ~w for term ~b,"
" candidate last log index term was: ~w~n"
"Last log entry idxterm seen was: ~w~n",
[id(State0), Cand, Term, {LLIdx, LLTerm}, LastIdxTerm]),
{FsmState, State,
[{reply, pre_vote_result(Term, Token, false)}]}
end;
process_pre_vote(FsmState, #pre_vote_rpc{term = Term,
token = Token,
candidate_id = _Cand},
#{current_term := CurTerm} = State)
when Term < CurTerm ->
?INFO("~w declining pre-vote to ~w for term ~b, current term ~b~n",
[id(State), _Cand, Term, CurTerm]),
{FsmState, State,
[{reply, pre_vote_result(CurTerm, Token, false)}]}.
pre_vote_result(Term, Token, Success) ->
#pre_vote_result{term = Term,
token = Token,
vote_granted = Success}.
new_peer() ->
#{next_index => 1,
match_index => 0,
commit_index_sent => 0}.
new_peer_with(Map) ->
maps:merge(new_peer(), Map).
peers(#{id := Id, cluster := Nodes}) ->
maps:remove(Id, Nodes).
% returns the peers that should receive piplined entries
pipelineable_peers(#{commit_index := CommitIndex,
log := Log} = State) ->
NextIdx = ra_log:next_index(Log),
maps:filter(fun (_, #{next_index := NI,
match_index := MI}) when NI < NextIdx ->
% there are unsent items
NI - MI < ?MAX_PIPELINE_DISTANCE;
(_, #{commit_index_sent := CI,
next_index := NI,
match_index := MI}) when CI < CommitIndex ->
% the commit index has been updated
NI - MI < ?MAX_PIPELINE_DISTANCE;
(_, _) ->
false
end, peers(State)).
% peers that could need an update
stale_peers(#{commit_index := CommitIndex} = State) ->
maps:filter(fun (_Id, #{next_index := NI,
match_index := MI}) when MI < NI - 1 ->
% there are unconfirmed items
true;
(_Id, #{commit_index_sent := CI}) when CI < CommitIndex ->
% the commit index has been updated
true;
(_Id, _Peer) ->
false
end, peers(State)).
peer_ids(State) ->
maps:keys(peers(State)).
peer(PeerId, #{cluster := Nodes}) ->
maps:get(PeerId, Nodes, undefined).
update_peer(PeerId, Peer, #{cluster := Nodes} = State) ->
State#{cluster => Nodes#{PeerId => Peer}}.
update_meta(Term, VotedFor, #{log := Log} = State) ->
ok = ra_log:write_meta(current_term, Term, Log, false),
ok = ra_log:write_meta(voted_for, VotedFor, Log, true),
State#{current_term => Term,
voted_for => VotedFor}.
update_term(Term, State = #{current_term := CurTerm})
when Term =/= undefined andalso Term > CurTerm ->
update_meta(Term, undefined, State);
update_term(_, State) ->
State.
last_idx_term(#{log := Log}) ->
ra_log:last_index_term(Log).
%% § 5.4.1 Raft determines which of two logs is more up-to-date by comparing
%% the index and term of the last entries in the logs. If the logs have last
%% entries with different terms, then the log with the later term is more
%% up-to-date. If the logs end with the same term, then whichever log is
%% longer is more up-to-dat
-spec is_candidate_log_up_to_date(ra_index(), ra_term(), ra_idxterm()) ->
boolean().
is_candidate_log_up_to_date(_, Term, {_, LastTerm})
when Term > LastTerm ->
true;
is_candidate_log_up_to_date(Idx, Term, {LastIdx, Term})
when Idx >= LastIdx ->
true;
is_candidate_log_up_to_date(_, _, {_, _}) ->
false.
has_log_entry_or_snapshot(Idx, Term, #{log := Log0} = State) ->
case ra_log:fetch_term(Idx, Log0) of
{undefined, Log} ->
case ra_log:snapshot_index_term(Log) of
{Idx, Term} ->
{entry_ok, State#{log => Log}};
{Idx, OtherTerm} ->
{term_mismatch, OtherTerm, State#{log => Log}};
_ ->
{missing, State#{log => Log}}
end;
{Term, Log} ->
{entry_ok, State#{log => Log}};
{OtherTerm, Log} ->
{term_mismatch, OtherTerm, State#{log => Log}}
end.
fetch_term(Idx, #{log := Log}) ->
ra_log:fetch_term(Idx, Log).
fetch_entries(From, To, #{log := Log0} = State) ->
{Entries, Log} = ra_log:take(From, To - From + 1, Log0),
{Entries, State#{log => Log}}.
make_cluster(Self, Nodes) ->
case lists:foldl(fun(N, Acc) ->
Acc#{N => new_peer()}
end, #{}, Nodes) of
#{Self := _} = Cluster ->
% current server is already in cluster - do nothing
Cluster;
Cluster ->
% add current server to cluster
Cluster#{Self => new_peer()}
end.
initialise_peers(State = #{log := Log, cluster := Cluster0}) ->
PeerIds = peer_ids(State),
NextIdx = ra_log:next_index(Log),
Cluster = lists:foldl(fun(PeerId, Acc) ->
Acc#{PeerId =>
new_peer_with(#{next_index => NextIdx})}
end, Cluster0, PeerIds),
State#{cluster => Cluster}.
apply_to(ApplyTo, #{machine := Machine} = State) ->
apply_to(ApplyTo, fun(E, S) -> apply_with(Machine, E, S) end, State).
apply_to(ApplyTo, ApplyFun, State) ->
apply_to(ApplyTo, ApplyFun, 0, #{}, [], State).
apply_to(ApplyTo, ApplyFun, NumApplied, Notifys0, Effects0,
#{last_applied := LastApplied,
machine_state := MacState0} = State0)
when ApplyTo > LastApplied ->
From = LastApplied + 1,
To = min(From + 1024, ApplyTo),
case fetch_entries(From, To, State0) of
{[], State} ->
%% reverse list before consing the notifications to ensure
%% notifications are processed first
Effects = make_notify_effects(Notifys0, lists:reverse(Effects0)),
{State, Effects, NumApplied};
%% assert first item read is from
{[{From, _, _} | _] = Entries, State1} ->
{AppliedTo, State, MacState, Effects, Notifys} =
lists:foldl(ApplyFun, {LastApplied, State1, MacState0,
Effects0, Notifys0}, Entries),
% {AppliedTo,_, _} = lists:last(Entries),
% ?INFO("applied: ~p ~nAfter: ~p", [Entries, MacState]),
apply_to(ApplyTo, ApplyFun, NumApplied + length(Entries),
Notifys, Effects, State#{last_applied => AppliedTo,
machine_state => MacState})
end;
apply_to(_, _, NumApplied, Notifys, Effects0, State) -> % ApplyTo
%% reverse list before consing the notifications to ensure
%% notifications are processed first
Effects = make_notify_effects(Notifys, lists:reverse(Effects0)),
{State, Effects, NumApplied}.
make_notify_effects(Nots, Effects) ->
maps:fold(fun (Pid, Corrs, Acc) ->
[{notify, Pid, lists:reverse(Corrs)} | Acc]
end, Effects, Nots).
apply_with(Machine,
{Idx, Term, {'$usr', #{from := From} = Meta0, Cmd, ReplyType}},
{_, State, MacSt, Effects0, Notifys0}) ->
Meta = Meta0#{index => Idx, term => Term},
{NextMacSt, Effects1, Reply} = ra_machine:apply(Machine, Meta, Cmd,
Effects0, MacSt),
{Effects, Notifys} = add_reply(From, Reply, ReplyType,
Effects1, Notifys0),
{Idx, State, NextMacSt, Effects, Notifys};
apply_with({machine, MacMod, _}, % Machine
{Idx, _, {'$ra_query', #{from := From}, QueryFun, _}},
{_, State, MacSt, Effects0, Notifys0}) ->
Result = ra_machine:query(MacMod, QueryFun, MacSt),
Effects = [{reply, From, {machine_reply, Result}} | Effects0],
{Idx, State, MacSt, Effects, Notifys0};
apply_with(_Machine,
{Idx, _, {'$ra_cluster_change', #{from := From}, _New,
ReplyType}},
{_, State0, MacSt, Effects0, Notifys0}) ->
?INFO("~w: applying ra cluster change to ~w~n",
[id(State0), maps:keys(_New)]),
{Effects, Notifys} = add_reply(From, ok, ReplyType,
Effects0, Notifys0),
State = State0#{cluster_change_permitted => true},
% add pending cluster change as next event
{Effects1, State1} = add_next_cluster_change(Effects, State),
{Idx, State1, MacSt, Effects1, Notifys};
apply_with(_, % Machine
{Idx, Term, noop}, % Idx
{_, State0 = #{current_term := Term}, MacSt, Effects, Notifys}) ->
?INFO("~w: enabling ra cluster changes in ~b~n", [id(State0), Term]),
State = State0#{cluster_change_permitted => true},
{Idx, State, MacSt, Effects, Notifys};
apply_with(Machine,
{Idx, _, {'$ra_cluster', #{from := From}, delete, ReplyType}},
{_, State0, MacSt, Effects0, Notifys0}) ->
% cluster deletion
{Effects1, Notifys} = add_reply(From, ok, ReplyType, Effects0, Notifys0),
Effects = make_notify_effects(Notifys, Effects1),
%% virtual "eol" state
EOLEffects = ra_machine:state_enter(Machine, eol, MacSt),
% non-local return to be caught by ra_server_proc
% need to update the state before throw
State = State0#{last_applied => Idx, machine_state => MacSt},
throw({delete_and_terminate, State, EOLEffects ++ Effects});
apply_with(_, {Idx, _, _} = Cmd, Acc) ->
% TODO: remove to make more strics, ideally we should not need a catch all
?WARN("~p: apply_with: unhandled command: ~W~n",
[id(element(2, Acc)), Cmd, 10]),
setelement(1, Acc, Idx).
add_next_cluster_change(Effects,
#{pending_cluster_changes := [C | Rest]} = State) ->
{_, #{from := From} , _, _} = C,
{[{next_event, {call, From}, {command, C}} | Effects],
State#{pending_cluster_changes => Rest}};
add_next_cluster_change(Effects, State) ->
{Effects, State}.
add_reply(From, Reply, await_consensus, Effects, Notifys) ->
{[{reply, From, {machine_reply, Reply}} | Effects], Notifys};
add_reply(undefined, Reply, {notify_on_consensus, Corr, Pid},
Effects, Notifys0) ->
% notify are casts and thus have to include their own pid()
% reply with the supplied correlation so that the sending can do their
% own bookkeeping
CorrData = {Corr, Reply},
Notifys = maps:update_with(Pid, fun (T) -> [CorrData | T] end,
[CorrData], Notifys0),
{Effects, Notifys};
add_reply(_, _, _, % From, Reply, Mode
Effects, Notifys) ->
{Effects, Notifys}.
append_log_leader({CmdTag, _, _, _} = Cmd,
State = #{cluster_change_permitted := false,
pending_cluster_changes := Pending})
when CmdTag == '$ra_join' orelse
CmdTag == '$ra_leave' ->
% cluster change is in progress or leader has not yet committed anything
% in this term - stash the request
{not_appended, State#{pending_cluster_changes => Pending ++ [Cmd]}};
append_log_leader({'$ra_join', From, JoiningNode, ReplyMode},
State = #{cluster := OldCluster}) ->
case OldCluster of
#{JoiningNode := _} ->
% already a member do nothing
% TODO: reply? If we don't reply the caller may block until timeout
{not_appended, State};
_ ->
Cluster = OldCluster#{JoiningNode => new_peer()},
append_cluster_change(Cluster, From, ReplyMode, State)
end;
append_log_leader({'$ra_leave', From, LeavingNode, ReplyMode},
State = #{cluster := OldCluster}) ->
case OldCluster of
#{LeavingNode := _} ->
Cluster = maps:remove(LeavingNode, OldCluster),
append_cluster_change(Cluster, From, ReplyMode, State);
_ ->
% not a member - do nothing
{not_appended, State}
end;
append_log_leader(Cmd, State = #{log := Log0, current_term := Term}) ->
NextIdx = ra_log:next_index(Log0),
Log = ra_log:append({NextIdx, Term, Cmd}, Log0),
{ok, NextIdx, Term, State#{log => Log}}.
pre_append_log_follower({Idx, Term, Cmd} = Entry,
{_, State = #{cluster_index_term := {Idx, CITTerm}}})
when Term /= CITTerm ->
% the index for the cluster config entry has a different term, i.e.
% it has been overwritten by a new leader. Unless it is another cluster
% change (can this even happen?) we should revert back to the last known
% cluster
case Cmd of
{'$ra_cluster_change', _, Cluster, _} ->
{Idx, State#{cluster => Cluster,
cluster_index_term => {Idx, Term}}};
_ ->
% revert back to previous cluster
{PrevIdx, PrevTerm, PrevCluster} = maps:get(previous_cluster,
State),
State1 = State#{cluster => PrevCluster,
cluster_index_term => {PrevIdx, PrevTerm}},
pre_append_log_follower(Entry, {Idx, State1})
end;
pre_append_log_follower({Idx, Term, {'$ra_cluster_change', _, Cluster, _}},
{_, State}) ->
{{Idx, Term}, State#{cluster => Cluster,
cluster_index_term => {Idx, Term}}};
pre_append_log_follower({Idx, _, _}, {_, State}) ->
{Idx, State}.
append_cluster_change(Cluster, From, ReplyMode,
State = #{log := Log0,
cluster := PrevCluster,
cluster_index_term := {PrevCITIdx, PrevCITTerm},
current_term := Term}) ->
% turn join command into a generic cluster change command
% that include the new cluster configuration
Command = {'$ra_cluster_change', From, Cluster, ReplyMode},
NextIdx = ra_log:next_index(Log0),
IdxTerm = {NextIdx, Term},
% TODO: is it safe to do change the cluster config with an async write?
% what happens if the write fails?
Log = ra_log:append({NextIdx, Term, Command}, Log0),
{ok, NextIdx, Term,
State#{log => Log,
cluster => Cluster,
cluster_change_permitted => false,
cluster_index_term => IdxTerm,
previous_cluster => {PrevCITIdx, PrevCITTerm, PrevCluster}}}.
mismatch_append_entries_reply(Term, CommitIndex, State = #{log := Log0}) ->
{CITerm, Log} = ra_log:fetch_term(CommitIndex, Log0),
% assert CITerm is found
false = CITerm =:= undefined,
{#append_entries_reply{term = Term, success = false,
next_index = CommitIndex + 1,
last_index = CommitIndex,
last_term = CITerm},
State#{log => Log}}.
append_entries_reply(Term, Success, State = #{log := Log}) ->
% ah - we can't use the the last received idx
% as it may not have been persisted yet
% also we can't use the last writted Idx as then
% the follower may resent items that are currently waiting to
% be written.
{LWIdx, LWTerm} = ra_log:last_written(Log),
{LastIdx, _} = last_idx_term(State),
#append_entries_reply{term = Term, success = Success,
next_index = LastIdx + 1,
last_index = LWIdx,
last_term = LWTerm}.
evaluate_quorum(State0) ->
State = #{commit_index := CI} = increment_commit_index(State0),
apply_to(CI, State).
increment_commit_index(State = #{current_term := CurrentTerm}) ->
PotentialNewCommitIndex = agreed_commit(match_indexes(State)),
% leaders can only increment their commit index if the corresponding
% log entry term matches the current term. See (§5.4.2)
case fetch_term(PotentialNewCommitIndex, State) of
{CurrentTerm, Log} ->
State#{commit_index => PotentialNewCommitIndex,
log => Log};
{_, Log} ->
State#{log => Log}
end.
match_indexes(#{log := Log} = State) ->
{LWIdx, _} = ra_log:last_written(Log),
maps:fold(fun(_K, #{match_index := Idx}, Acc) ->
[Idx | Acc]
end, [LWIdx], peers(State)).
-spec agreed_commit(list()) -> ra_index().
agreed_commit(Indexes) ->
SortedIdxs = lists:sort(fun erlang:'>'/2, Indexes),
Nth = trunc(length(SortedIdxs) / 2) + 1,
lists:nth(Nth, SortedIdxs).
log_unhandled_msg(RaState, Msg, #{id := Id}) ->
?WARN("~w ~w received unhandled msg: ~W~n", [Id, RaState, Msg, 6]).
fold_log_from(From, Folder, {St, Log0}) ->
case ra_log:take(From, ?FOLD_LOG_BATCH_SIZE, Log0) of
{[], Log} ->
{ok, {St, Log}};
{Entries, Log} ->
try
St1 = lists:foldl(Folder, St, Entries),
fold_log_from(From + ?FOLD_LOG_BATCH_SIZE, Folder, {St1, Log})
catch
_:Reason ->
{error, Reason, Log}
end
end.
drop_existing({Log0, []}) ->
{Log0, []};
drop_existing({Log0, [{Idx, Trm, _} | Tail] = Entries}) ->
case ra_log:exists({Idx, Trm}, Log0) of
{true, Log} ->
drop_existing({Log, Tail});
{false, Log} ->
{Log, Entries}
end.
cast_reply(From, To, Msg) ->
{cast, To, {From, Msg}}.
%%% ===================
%%% Internal unit tests
%%% ===================
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
agreed_commit_test() ->
% one server
4 = agreed_commit([4]),
% 2 servers - only leader has seen new commit
3 = agreed_commit([4, 3]),
% 2 servers - all servers have seen new commit
4 = agreed_commit([4, 4, 4]),
% 3 servers - leader + 1 server has seen new commit
4 = agreed_commit([4, 4, 3]),
% only other servers have seen new commit
4 = agreed_commit([3, 4, 4]),
% 3 servers - only leader has seen new commit
3 = agreed_commit([4, 2, 3]),
ok.
-endif.