%% License, v. 2.0. If a copy of the MPL was not distributed with this %% file, You can obtain one at https://mozilla.org/MPL/2.0/. %% %% Copyright (c) 2017-2025 Broadcom. All Rights Reserved. The term Broadcom refers to Broadcom Inc. and/or its subsidiaries. %% @hidden -module(ra_server). -include_lib("stdlib/include/assert.hrl"). -include("ra.hrl"). -include("ra_server.hrl"). -compile(inline_list_funcs). -elvis([{elvis_style, dont_repeat_yourself, disable}]). -elvis([{elvis_style, god_modules, disable}]). -export([ name/2, init/1, process_new_leader_queries/1, handle_leader/2, handle_candidate/2, handle_pre_vote/2, handle_follower/2, handle_receive_snapshot/2, handle_await_condition/2, handle_aux/4, handle_state_enter/3, tick/1, log_tick/1, overview/1, is_new/1, is_fully_persisted/1, is_fully_replicated/1, % properties id/1, uid/1, log_id/1, system_config/1, leader_id/1, clear_leader_id/1, current_term/1, machine_version/1, machine/1, machine_query/2, % TODO: hide behind a handle_leader make_rpcs/1, update_release_cursor/4, promote_checkpoint/2, checkpoint/3, persist_last_applied/1, peers/1, peer_status/2, update_peer/3, update_disconnected_peers/3, handle_down/5, handle_node_status/6, terminate/2, log_fold/3, log_read/2, log_partial_read/2, get_membership/1, get_condition_timeout/2, recover/1, state_query/2, fetch_term/2, transform_for_partial_read/3 ]). -type ra_await_condition_fun() :: fun((ra_msg(), ra_server_state()) -> {boolean(), ra_server_state()}). -type ra_server_state() :: #{cfg := #cfg{}, leader_id => option(ra_server_id()), cluster := ra_cluster(), cluster_change_permitted := boolean(), cluster_index_term := ra_idxterm(), previous_cluster => {ra_index(), ra_term(), ra_cluster()}, current_term := ra_term(), log := term(), voted_for => option(ra_server_id()), % persistent votes => non_neg_integer(), membership => ra_membership(), commit_index := ra_index(), last_applied := ra_index(), persisted_last_applied => ra_index(), machine_state := term(), aux_state => term(), condition => #{predicate_fun := ra_await_condition_fun(), transition_to => ra_state(), timeout => #{duration => integer(), transition_to => ra_state(), effects => [effect()]}}, pre_vote_token => reference(), query_index := non_neg_integer(), queries_waiting_heartbeats := queue:queue({non_neg_integer(), consistent_query_ref()}), pending_consistent_queries := [consistent_query_ref()], commit_latency => option(non_neg_integer()), %% snapshot_phase tracks the current phase of snapshot reception: %% - chunk_flag() for normal phases (init, pre, next, last) %% - {awaiting_pending, EventType, Rpc} when waiting for pending WAL %% writes to complete before finalizing snapshot installation snapshot_phase => chunk_flag() | {awaiting_pending, term(), #install_snapshot_rpc{}}, pending_release_cursor => {ra_index(), term(), [ra_machine:release_cursor_condition()]}, %% temporary state for handle_receive_snapshot current_event_type => term(), snapshot_has_live_indexes => boolean() }. -type state() :: ra_server_state(). -type ra_state() :: leader | follower | candidate | pre_vote | await_condition | delete_and_terminate | terminating_leader | terminating_follower | recover | recovered | stop | receive_snapshot. -type command_type() :: '$usr' | '$ra_join' | '$ra_leave' | '$ra_cluster_change' | '$ra_cluster'. -type command_meta() :: #{from => from(), ts := integer()}. -type command_correlation() :: integer() | reference(). -type command_priority() :: normal | low. -type command_reply_options() :: #{reply_from => ra_reply_from()}. -type command_reply_mode() :: after_log_append | await_consensus | {await_consensus, command_reply_options()} | {notify, command_correlation(), pid()} | noreply. -type command() :: {command_type(), command_meta(), UserCommand :: term(), command_reply_mode()} | {noop, command_meta(), CurrentMachineVersion :: ra_machine:version()}. -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() | {consistent_query, from(), ra:query_fun()} | {consistent_aux, from(), AuxCmd :: term()} | #heartbeat_rpc{} | #info_rpc{} | #info_reply{} | {ra_server_id, #heartbeat_reply{}} | pipeline_rpcs | {snapshot_retry_timeout, ra_server_id()}. -type ra_reply_body() :: #append_entries_reply{} | #request_vote_result{} | #install_snapshot_result{} | #pre_vote_result{}. -type effect() :: ra_machine:effect() | ra_log:effect() | %% this is used for replies for immedate requests {reply, ra_reply_body()} | %% this is used by the leader only {reply, from(), ra_reply_body()} | {reply, from(), ra_reply_body(), Replier :: leader | local | {member, ra_server_id()}} | {cast, ra_server_id(), term()} | {send_vote_requests, [{ra_server_id(), #request_vote_rpc{} | #pre_vote_rpc{}}]} | {send_rpc, ra_server_id(), #append_entries_rpc{} | #heartbeat_rpc{} | #info_rpc{}} | {send_snapshot, To :: ra_server_id(), {Module :: module(), Ref :: term(), LeaderId :: ra_server_id(), Term :: ra_term()}} | {next_event, ra_msg()} | {next_event, cast, ra_msg()} | {notify, #{pid() => [term()]}} | %% used for tracking valid leader messages {record_leader_msg, ra_server_id()} | start_election_timeout | {start_snapshot_retry_timer, ra_server_id(), non_neg_integer()} | {cancel_snapshot_retry_timer, ra_server_id()} | {bg_work, fun(() -> ok) | mfargs(), fun()}. -type effects() :: [effect()]. -type simple_apply_fun(State) :: fun((term(), State) -> State). -type ra_event_formatter_fun() :: fun((ra_server_id(), Evt :: term()) -> term()). -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 machine_upgrade_strategy() :: all | quorum. %% When a new Ra machine version should be upgraded to. %% %% -type ra_server_config() :: #{id := ra_server_id(), uid := ra_uid(), %% a friendly name to refer to a particular %% server - will default to the id formatted %% with `~tw' cluster_name := ra_cluster_name(), log_init_args := ra_log:ra_log_init_args(), initial_members := [ra_server_id()], machine := machine_conf(), initial_machine_version => ra_machine:version(), friendly_name => unicode:chardata(), metrics_key => term(), metrics_labels => seshat:labels_map(), % 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 install_snap_rpc_timeout => non_neg_integer(), % ms await_condition_timeout => non_neg_integer(), max_pipeline_count => non_neg_integer(), ra_event_formatter => {module(), atom(), [term()]}, counter => counters:counters_ref(), membership => ra_membership(), system_config => ra_system:config(), has_changed => boolean(), parent => term(), %% the supervisor %% minimum number of log entries since last %% snapshot/checkpoint before writing a recovery %% checkpoint on shutdown. 0 disables (default). min_recovery_checkpoint_interval => non_neg_integer() }. -type ra_server_info_key() :: machine_version | atom(). %% Key one can get in `ra_server_info()'. %% %% This is used in the `#info_rpc{}' to ask for a specific list of info keys. %% %% Key meanings: %% %% %% Any atom is supported because a future version of Ra could ask for %% something this version does not know about. -type ra_server_info() :: #{machine_version => ra_machine:version(), atom() => any()}. %% Info for a Ra server, got from `#info_reply{}'. %% %% In addition, the map may contain keys unknown to this version of Ra but %% emitted by a future version. -type mutable_config() :: #{cluster_name => ra_cluster_name(), metrics_key => term(), metrics_labels => seshat:labels_map(), broadcast_time => non_neg_integer(), % ms tick_timeout => non_neg_integer(), % ms install_snap_rpc_timeout => non_neg_integer(), % ms await_condition_timeout => non_neg_integer(), max_pipeline_count => non_neg_integer(), ra_event_formatter => {module(), atom(), [term()]}, %% minimum number of log entries since last %% snapshot/checkpoint before writing a recovery %% checkpoint on shutdown. 0 disables (default). min_recovery_checkpoint_interval => non_neg_integer()}. -type config() :: ra_server_config(). -export_type([state/0, config/0, ra_server_state/0, ra_state/0, ra_server_config/0, ra_server_info_key/0, ra_server_info/0, mutable_config/0, ra_msg/0, machine_conf/0, command/0, command_type/0, command_meta/0, command_correlation/0, command_priority/0, command_reply_mode/0, ra_event_formatter_fun/0, effect/0, effects/0, machine_upgrade_strategy/0 ]). -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) -> SystemConfig = maps:get(system_config, Config, ra_system:default_config()), LogId = maps:get(friendly_name, Config, lists:flatten(io_lib:format("~tw", [Id]))), DefaultMaxPipelineCount = maps:get(default_max_pipeline_count, SystemConfig, ?DEFAULT_MAX_PIPELINE_COUNT), MaxPipelineCount = maps:get(max_pipeline_count, Config, DefaultMaxPipelineCount), DefaultMaxAERBatchSize = maps:get(default_max_append_entries_rpc_batch_size, SystemConfig, ?AER_CHUNK_SIZE), MaxAERBatchSize = maps:get(max_append_entries_rpc_batch_size, Config, DefaultMaxAERBatchSize), MetricKey = case Config of #{metrics_key := K} -> K; _ -> ra_lib:ra_server_id_to_local_name(Id) end, Name = ra_lib:ra_server_id_to_local_name(Id), Machine = case MachineConf of {simple, Fun, S} -> {machine, ra_machine_simple, #{simple_fun => Fun, initial_state => S}}; {module, Mod, Args} -> {machine, Mod, Args} end, SnapModule = ra_machine:snapshot_module(Machine), Counter = maps:get(counter, Config, undefined), Log0 = ra_log:init(LogInitArgs#{snapshot_module => SnapModule, machine => Machine, system_config => SystemConfig, uid => UId, counter => Counter, log_id => LogId, %% use sequential access pattern during %% recovery initial_access_pattern => sequential}), %% only write config if it is different from what is already on disk case Config of #{has_changed := false} -> ok; _ -> ok = ra_log:write_config(Config, Log0) end, MetaName = meta_name(SystemConfig), CurrentTerm = ra_log_meta:fetch(MetaName, UId, current_term, 0), LastApplied = ra_log_meta:fetch(MetaName, UId, last_applied, 0), VotedFor = ra_log_meta:fetch(MetaName, UId, voted_for, undefined), LatestMacVer = ra_machine:version(Machine), InitialMachineVersion = min(LatestMacVer, maps:get(initial_machine_version, Config, 0)), {Cluster0, EffectiveMacVer, MacState, {SnapshotIdx, _} = SnapshotIndexTerm} = case ra_log:recover_snapshot(Log0) of undefined -> InitialMachineState = ra_machine:init(Machine, Name, InitialMachineVersion), {make_cluster(Id, InitialNodes), InitialMachineVersion, InitialMachineState, {0, 0}}; {#{index := Idx, term := Term, cluster := ClusterNodes, machine_version := MacVersion}, MacSt} -> Clu = make_cluster(Id, ClusterNodes), %% the snapshot is the last index before the first index {Clu, MacVersion, MacSt, {Idx, Term}} end, MacMod = ra_machine:which_module(Machine, EffectiveMacVer), CommitIndex = max(LastApplied, SnapshotIdx), MinRecoveryCheckpointInterval = maps:get(min_recovery_checkpoint_interval, Config, 0), Cfg = #cfg{id = Id, uid = UId, log_id = LogId, metrics_key = MetricKey, metrics_labels = #{}, machine = Machine, machine_version = LatestMacVer, machine_versions = [{SnapshotIdx, EffectiveMacVer}], effective_machine_version = EffectiveMacVer, effective_machine_module = MacMod, effective_handle_aux_fun = ra_machine:which_aux_fun(MacMod), max_pipeline_count = MaxPipelineCount, max_append_entries_rpc_batch_size = MaxAERBatchSize, counter = maps:get(counter, Config, undefined), system_config = SystemConfig, min_recovery_checkpoint_interval = MinRecoveryCheckpointInterval}, put_counter(Cfg, ?C_RA_SVR_METRIC_COMMIT_INDEX, CommitIndex), put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_APPLIED, SnapshotIdx), put_counter(Cfg, ?C_RA_SVR_METRIC_TERM, CurrentTerm), put_counter(Cfg, ?C_RA_SVR_METRIC_EFFECTIVE_MACHINE_VERSION, EffectiveMacVer), NonVoter = get_membership(Cluster0, Id, UId, maps:get(membership, Config, voter)), #{cfg => Cfg, leader_id => undefined, current_term => CurrentTerm, 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. cluster_change_permitted => false, cluster_index_term => SnapshotIndexTerm, voted_for => VotedFor, membership => NonVoter, commit_index => CommitIndex, %% set this to the first index so that we can apply all entries %% up to the commit index during recovery last_applied => SnapshotIdx, persisted_last_applied => LastApplied, log => Log0, machine_state => MacState, %% aux state is transient and needs to be initialized every time aux_state => ra_machine:init_aux(MacMod, Name), query_index => 0, queries_waiting_heartbeats => queue:new(), pending_consistent_queries => []}. recover(#{cfg := #cfg{log_id = LogId, machine_version = MacVer}, commit_index := CommitIndex, last_applied := LastApplied, log := Log} = State0) -> %% Check if we can skip some log replay using a recovery checkpoint SnapState = ra_log:snapshot_state(Log), {LastApplied1, #{cfg := Cfg} = State1} = maybe_recover_from_recovery_checkpoint(LastApplied, CommitIndex, SnapState, State0), put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_APPLIED, LastApplied1), ?DEBUG("~ts: recovering state machine version ~b:~b from index ~b to ~b", [LogId, Cfg#cfg.effective_machine_version, MacVer, LastApplied1, CommitIndex]), Before = erlang:system_time(millisecond), {#{log := Log0, cfg := #cfg{effective_machine_version = EffMacVerAfter}} = State2, _} = apply_to(CommitIndex, fun({_Idx, _, _} = E, S0) -> %% Clear out the effects and notifies map %% to avoid memory explosion {Mod, LastAppl, S, MacSt, _E, _N, LastTs} = apply_with(E, S0), put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_APPLIED, LastAppl), {Mod, LastAppl, S, MacSt, [], #{}, LastTs} end, State1, []), After = erlang:system_time(millisecond), ?DEBUG("~ts: recovery of state machine version ~b:~b " "from index ~b to ~b took ~bms", [LogId, EffMacVerAfter, MacVer, LastApplied1, CommitIndex, After - Before]), %% scan from CommitIndex + 1 until NextIndex - 1 to see if there are %% any further cluster changes FromScan = CommitIndex + 1, {ToScan, _} = ra_log:last_index_term(Log0), ?DEBUG("~ts: scanning for cluster changes ~b:~b ", [LogId, FromScan, ToScan]), %% if we're recovering after a partial sparse write phase this will fail {{LastScannedIdx, State3}, Log1} = ra_log:fold(FromScan, ToScan, fun cluster_scan_fun/2, {CommitIndex, State2}, Log0, return), State = case LastScannedIdx < ToScan of true -> ?DEBUG("~ts: scan detected sparse log last scanned ~b:~b " "resetting log to last contiguous index ~b", [LogId, LastScannedIdx, ToScan, LastScannedIdx]), %% the end of the log is sparse and needs to be reset {ok, Log2} = ra_log:set_last_index(LastScannedIdx, Log1), State3#{log => Log2}; false -> State3 end, put_counter(Cfg, ?C_RA_SVR_METRIC_COMMIT_LATENCY, 0), State#{ %% reset commit latency as recovery may calculate a very old value commit_latency => 0}. -spec handle_leader(ra_msg(), ra_server_state()) -> {ra_state(), ra_server_state(), effects()}. handle_leader({PeerId, #append_entries_reply{term = Term, success = true, next_index = NextIdx, last_index = LastIdx}}, #{current_term := Term, cfg := #cfg{id = Id, log_id = LogId} = Cfg} = State0) -> ok = incr_counter(Cfg, ?C_RA_SRV_AER_REPLIES_SUCCESS, 1), case peer(PeerId, State0) of undefined -> ?WARN("~ts: saw append_entries_reply from unknown peer ~tw", [LogId, PeerId]), {leader, State0, []}; #{match_index := MI, next_index := NI} = Peer0 -> Peer = Peer0#{match_index => max(MI, LastIdx), next_index => max(NI, NextIdx)}, State1 = put_peer(PeerId, Peer, State0), Effects00 = maybe_promote_peer(PeerId, State1, []), {State2, Effects0} = evaluate_quorum(State1, Effects00), {State3, Effects1} = process_pending_consistent_queries(State2, Effects0), Effects2 = [{next_event, info, pipeline_rpcs} | Effects1], case State3 of #{cluster := #{Id := _}} -> % leader is in the cluster {leader, State3, Effects2}; #{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("~ts: leader not in new cluster - goodbye", [LogId]), %% need to make pipelined rpcs here as cannot use next event {State, _, Effects} = make_pipelined_rpc_effects(State3, Effects2), {stop, State, Effects}; _ -> {leader, State3, Effects2} end end; handle_leader({PeerId, #append_entries_reply{term = Term}}, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when Term > CurTerm -> case peer(PeerId, State0) of undefined -> ?WARN("~ts: saw append_entries_reply from unknown peer ~tw", [LogId, PeerId]), {leader, State0, []}; _ -> ?NOTICE("~ts: leader saw append_entries_reply from ~tw for term ~b " "abdicates term: ~b!", [LogId, PeerId, Term, CurTerm]), {follower, update_term(Term, State0#{leader_id => undefined}), []} end; handle_leader({PeerId, #append_entries_reply{success = false}}, State0 = #{cfg := #cfg{log_id = LogId}, cluster := Nodes}) when not is_map_key(PeerId, Nodes) -> ?WARN("~ts: saw append_entries_reply from unknown peer ~tw", [LogId, PeerId]), {leader, State0, []}; handle_leader({PeerId, #append_entries_reply{success = false, next_index = PeerNextIdx, last_index = PeerLastIdx, last_term = PEER_LAST_TERM}}, #{cfg := #cfg{log_id = LogId} = Cfg, cluster := Nodes, log := Log0} = State0) -> ok = incr_counter(Cfg, ?C_RA_SRV_AER_REPLIES_FAILED, 1), #{PeerId := #{match_index := MI, next_index := NI} = Peer0} = 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(PeerLastIdx, Log0) of {undefined, Log1} -> % entry was not found ?DEBUG("~ts: peer ~tw last index ~b not found " "setting next index to ~b", [LogId, PeerId, PeerLastIdx, PeerNextIdx]), {Peer0#{next_index => PeerNextIdx}, Log1}; % entry exists we can forward {PEER_LAST_TERM = Term, Log1} when is_integer(Term) andalso PeerLastIdx >= MI -> ?DEBUG("~ts: setting last index to ~b, " " next_index ~b for ~tw", [LogId, PeerLastIdx, PeerNextIdx, PeerId]), {Peer0#{match_index => PeerLastIdx, next_index => PeerNextIdx}, Log1}; {_OtherTerm, Log1} when PeerLastIdx < MI -> % TODO: this can only really happen when peers are % non-persistent. % Should they turn-into non-voters when this situation % is detected?? ?WARN("~ts: leader saw peer with last_index [~b in ~b]" " lower than recorded match index [~b]." "Resetting peer's match index to their last index.", [LogId, PeerLastIdx, PEER_LAST_TERM, MI]), {Peer0#{match_index => PeerLastIdx, next_index => PeerLastIdx + 1}, Log1}; {EntryTerm, Log1} -> % last_index has a different term or entry does not % exist % The peer must have uncommitted entries from a prior % term that have now been overwritten by the current % leader. % Decrement next_index but don't go lower than % match index. NextIndex = max(min(NI-1, PeerNextIdx), MI), ?DEBUG("~ts: leader received last_index ~b" " from ~tw with term ~b " "- expected term ~b. Setting " "next_index to ~b", [LogId, PeerLastIdx, PeerId, PEER_LAST_TERM, EntryTerm, NextIndex]), {Peer0#{next_index => NextIndex}, Log1} end, State1 = State0#{cluster => Nodes#{PeerId => Peer}, log => Log}, {State, _, Effects} = make_pipelined_rpc_effects(State1, []), {leader, State, Effects}; handle_leader({command, Cmd}, #{cfg := #cfg{id = Self, log_id = LogId} = Cfg, cluster := Cluster} = State00) -> ok = incr_counter(Cfg, ?C_RA_SRV_COMMANDS, 1), case append_log_leader(Cmd, State00, []) of {not_appended, wal_down, State0, Effects0} -> %% TODO: pick peer with highest match_index and on condition timeout %% may not be worth the effort as this almost _never_ happens CondEffs = case maps:to_list(maps:remove(Self, Cluster)) of [] -> []; [{PeerId, _} | _] -> [{next_event, cast, {transfer_leadership, PeerId}}] end, State = State0#{condition => #{predicate_fun => fun wal_down_condition/2, transition_to => leader, timeout => #{duration => 5000, effects => CondEffs, transition_to => leader}}}, Effects = append_error_reply(Cmd, wal_down, Effects0), {await_condition, State, Effects}; {not_appended, Reason, State, Effects0} -> ?WARN("~ts command ~W NOT appended to log. Reason ~w", [LogId, Cmd, 10, Reason]), Effects = append_error_reply(Cmd, Reason, Effects0), {leader, State, Effects}; {ok, Idx, Term, State0, Effects00} -> %% if the command is a noop command we should force it to %% be pipelined to all followers Force = case Cmd of {noop, _, _} -> true; _ -> false end, {State, _, Effects0} = make_pipelined_rpc_effects(State0, Effects00, Force), % check if a reply is required. Effects = after_log_append_reply(Cmd, Idx, Term, Effects0), {leader, State, Effects} end; handle_leader({commands, Cmds}, #{cfg := #cfg{id = Self, log_id = LogId} = Cfg, cluster := Cluster, log := Log0} = State000) -> Num = length(Cmds), State00 = State000#{log => ra_log:begin_tx(Log0)}, %% only user commands are appended in a {commands, Cmds} batch {State0, Effects0} = lists:foldl( fun(C, {S0, E0}) -> {ok, I, T, S, E} = append_log_leader(C, S0, E0), {S, after_log_append_reply(C, I, T, E)} end, {State00, []}, Cmds), ok = incr_counter(Cfg, ?C_RA_SRV_COMMAND_FLUSHES, 1), ok = incr_counter(Cfg, ?C_RA_SRV_COMMANDS, Num), #{log := Log1} = State0, case ra_log:commit_tx(Log1) of {ok, Log} -> {State, _, Effects} = make_pipelined_rpc_effects(State0#{log => Log}, Effects0), {leader, State, Effects}; {error, wal_down, Log} -> ?WARN("~ts ~b commands NOT appended to Raft log. Reason: wal_down", [LogId, length(Cmds)]), % at this point the entries are committed to the mem table % but didn't make it fully or at all to the the wal %% TODO: select the peer with the highest match index? CondEffs = case maps:to_list(maps:remove(Self, Cluster)) of [] -> []; [{PeerId, _} | _] -> [{next_event, cast, {transfer_leadership, PeerId}}] end, State = State0#{condition => #{predicate_fun => fun wal_down_condition/2, transition_to => leader, %% TODO: make duration configurable? timeout => #{duration => 5000, effects => CondEffs, transition_to => leader}}, log => Log}, {await_condition, State, Effects0} end; handle_leader({ra_log_event, {written, _, _} = Evt}, #{log := Log0} = State0) -> {Log, Effects0} = ra_log:handle_event(Evt, Log0), {State1, Effects1} = evaluate_quorum(State0#{log => Log}, Effects0), {State, Effects} = process_pending_consistent_queries(State1, Effects1), {leader, State, [{next_event, info, pipeline_rpcs} | Effects]}; handle_leader({ra_log_event, Evt}, State = #{log := Log0}) -> {Log1, Effects} = ra_log:handle_event(Evt, Log0), {leader, State#{log => Log1}, Effects}; handle_leader({aux_command, Type, Cmd}, State0) -> handle_aux(leader, Type, Cmd, State0); handle_leader({PeerId, #install_snapshot_result{term = Term}}, #{cfg := #cfg{log_id = LogId}, current_term := CurTerm} = State0) when Term > CurTerm -> case peer(PeerId, State0) of undefined -> ?WARN("~ts: saw install_snapshot_result from unknown peer ~tw", [LogId, PeerId]), {leader, State0, []}; _ -> ?DEBUG("~ts: leader saw install_snapshot_result from ~tw for term ~b" " abdicates term: ~b!", [LogId, PeerId, Term, CurTerm]), {follower, update_term(Term, State0#{leader_id => undefined}), []} end; handle_leader({PeerId, #install_snapshot_result{last_index = LastIndex}}, #{cfg := #cfg{log_id = LogId}} = State0) -> case peer(PeerId, State0) of undefined -> ?WARN("~ts: saw install_snapshot_result from unknown peer ~tw", [LogId, PeerId]), {leader, State0, []}; Peer0 -> State1 = put_peer(PeerId, Peer0#{status => normal, match_index => LastIndex, commit_index_sent => LastIndex, next_index => LastIndex + 1}, State0), %% we can now demonitor the process Effects0 = case Peer0 of #{status := {sending_snapshot, Pid, _}} -> [{demonitor, process, Pid}]; _ -> [] end, %% Check if pending release cursor can now proceed %% (no_snapshot_sends condition may now be satisfied) {State2, Effects1} = maybe_emit_pending_release_cursor(State1, Effects0), Effects2 = maybe_promote_peer(PeerId, State2, Effects1), {State, _, Effects} = make_pipelined_rpc_effects(State2, Effects2), {leader, State, Effects} end; handle_leader(pipeline_rpcs, State0) -> {State, More, Effects0} = make_pipelined_rpc_effects(State0, []), Effects = case More of true -> [{next_event, info, pipeline_rpcs} | Effects0]; false -> Effects0 end, {leader, State, Effects}; handle_leader({snapshot_retry_timeout, PeerId}, #{cfg := #cfg{id = Id}, current_term := Term, log := Log, cluster := Peers} = State0) -> case maps:get(PeerId, Peers, undefined) of #{status := {snapshot_backoff, _Count}} -> %% Keep status as snapshot_backoff so send_snapshot effect handler %% can read the attempt count. The effect handler will update %% status to {sending_snapshot, Pid, Count}. SnapState = ra_log:snapshot_state(Log), {leader, State0, [{send_snapshot, PeerId, {SnapState, Id, Term}}]}; _ -> %% Peer no longer in backoff (maybe removed or snapshot succeeded) {leader, State0, []} end; handle_leader(#install_snapshot_rpc{term = Term, leader_id = Leader} = Evt, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when Term > CurTerm -> case peer(Leader, State0) of undefined -> ?WARN("~ts: saw install_snapshot_rpc from unknown leader ~tw", [LogId, Leader]), {leader, State0, []}; _ -> ?INFO("~ts: leader saw install_snapshot_rpc from ~tw for term ~b " "abdicates term: ~b!", [LogId, Evt#install_snapshot_rpc.leader_id, Term, CurTerm]), {follower, update_term(Term, State0#{leader_id => undefined}), [{next_event, Evt}]} end; handle_leader(#append_entries_rpc{term = Term} = Msg, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when Term > CurTerm -> ?INFO("~ts: leader saw append_entries_rpc from ~tw for term ~b " "abdicates term: ~b!", [LogId, Msg#append_entries_rpc.leader_id, Term, CurTerm]), {follower, update_term(Term, State0#{leader_id => undefined}), [{next_event, Msg}]}; handle_leader(#append_entries_rpc{term = Term}, #{current_term := Term, cfg := #cfg{log_id = LogId}}) -> ?ERR("~ts: leader saw append_entries_rpc for same term ~b" " this should not happen!", [LogId, Term]), exit(leader_saw_append_entries_rpc_in_same_term); handle_leader(#append_entries_rpc{leader_id = LeaderId}, #{current_term := CurTerm, cfg := #cfg{id = Id}} = State0) -> Reply = append_entries_reply(CurTerm, false, State0), {leader, State0, [cast_reply(Id, LeaderId, Reply)]}; handle_leader({consistent_query, From, QueryFun}, #{commit_index := CommitIndex, cluster_change_permitted := true} = State0) -> QueryRef = {query, From, QueryFun, CommitIndex}, {State1, Effects} = make_heartbeat_rpc_effects(QueryRef, State0), {leader, State1, Effects}; handle_leader({consistent_query, From, QueryFun}, #{commit_index := CommitIndex, cluster_change_permitted := false, pending_consistent_queries := PQ} = State0) -> QueryRef = {query, From, QueryFun, CommitIndex}, {leader, State0#{pending_consistent_queries => [QueryRef | PQ]}, []}; handle_leader({consistent_aux, From, AuxCmd}, #{commit_index := CommitIndex, cluster_change_permitted := true} = State0) -> QueryRef = {aux, From, AuxCmd, CommitIndex}, {State1, Effects} = make_heartbeat_rpc_effects(QueryRef, State0), {leader, State1, Effects}; handle_leader({consistent_aux, From, AuxCmd}, #{commit_index := CommitIndex, cluster_change_permitted := false, pending_consistent_queries := PQ} = State0) -> QueryRef = {aux, From, AuxCmd, CommitIndex}, {leader, State0#{pending_consistent_queries => [QueryRef | PQ]}, []}; %% Lihtweight version of append_entries_rpc handle_leader(#heartbeat_rpc{term = Term} = Msg, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when CurTerm < Term -> ?INFO("~ts: leader saw heartbeat_rpc from ~tw for term ~b " "abdicates term: ~b!", [LogId, Msg#heartbeat_rpc.leader_id, Term, CurTerm]), {follower, update_term(Term, State0#{leader_id => undefined}), [{next_event, Msg}]}; handle_leader(#heartbeat_rpc{term = Term, query_index = QueryIndex, leader_id = LeaderId}, #{current_term := CurTerm, cfg := #cfg{id = Id}} = State) when CurTerm > Term -> Reply = heartbeat_reply(CurTerm, QueryIndex), {leader, State, [cast_reply(Id, LeaderId, Reply)]}; handle_leader(#heartbeat_rpc{term = Term}, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}}) when CurTerm == Term -> ?ERR("~ts: leader saw heartbeat_rpc for same term ~b" " this should not happen!", [LogId, Term]), exit(leader_saw_heartbeat_rpc_in_same_term); handle_leader({PeerId, #heartbeat_reply{query_index = ReplyQueryIndex, term = Term}}, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) -> case {CurTerm, Term} of {Same, Same} -> %% Heartbeat confirmed case heartbeat_rpc_quorum(ReplyQueryIndex, PeerId, State0) of {[], State} -> {leader, State, []}; {QueryRefs, State} -> Effects = apply_consistent_queries_effects(QueryRefs, State), {leader, State, Effects} end; {CurHigher, TermLower} when CurHigher > TermLower -> %% Heartbeat reply for lower term. Ignoring {leader, State0, []}; {CurLower, TermHigher} when CurLower < TermHigher -> %% A node with higher term confirmed heartbeat. ?NOTICE("~ts leader saw heartbeat_reply from ~tw for term ~b " "abdicates term: ~b!", [LogId, PeerId, Term, CurTerm]), {follower, update_term(Term, State0#{leader_id => undefined}), []} end; handle_leader(#request_vote_rpc{term = Term, candidate_id = Cand} = Msg, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when Term > CurTerm -> case peer(Cand, State0) of undefined -> ?WARN("~ts: leader saw request_vote_rpc for unknown peer ~tw", [LogId, Cand]), {leader, State0, []}; _ -> ?INFO("~ts: leader saw request_vote_rpc from ~tw for term ~b " "abdicates term: ~b!", [LogId, Msg#request_vote_rpc.candidate_id, Term, CurTerm]), {follower, update_term(Term, State0#{leader_id => undefined}), [{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, cfg := #cfg{log_id = LogId}} = State0) when Term > CurTerm -> case peer(Cand, State0) of undefined -> ?WARN("~ts: leader saw pre_vote_rpc for unknown peer ~tw", [LogId, Cand]), {leader, State0, []}; _ -> ?INFO("~ts: leader saw pre_vote_rpc from ~tw for term ~b" " abdicates term: ~b!", [LogId, Msg#pre_vote_rpc.candidate_id, Term, CurTerm]), {follower, update_term(Term, State0#{leader_id => undefined}), [{next_event, Msg}]} end; handle_leader(#pre_vote_rpc{term = Term}, #{current_term := CurTerm} = State0) when Term =< CurTerm -> % enforce leadership {State, Effects} = make_all_rpcs(State0), {leader, State, Effects}; handle_leader(#request_vote_result{}, State) -> %% handle to avoid logging as unhandled {leader, State, []}; handle_leader(#pre_vote_result{}, State) -> %% handle to avoid logging as unhandled {leader, State, []}; handle_leader(#info_rpc{term = Term} = Msg, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when CurTerm < Term -> ?INFO("~ts: leader saw info_rpc from ~tw for term ~b, abdicates term: ~b!", [LogId, Msg#info_rpc.from, Term, CurTerm]), {follower, update_term(Term, State0#{leader_id => undefined}), [{next_event, Msg}]}; handle_leader(#info_rpc{} = InfoRpc, State) -> InfoReplyEffect = info_reply_effect(State, InfoRpc), {leader, State, [InfoReplyEffect]}; handle_leader(#info_reply{term = Term} = Msg, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when CurTerm < Term -> ?INFO("~ts: leader saw info_reply from ~tw for term ~b, abdicates " "term: ~b!", [LogId, Msg#info_reply.from, Term, CurTerm]), {follower, update_term(Term, State0#{leader_id => undefined}), [{next_event, Msg}]}; handle_leader(#info_reply{} = InfoReply, State) -> {State1, Effects} = handle_info_reply(State, InfoReply), {leader, State1, Effects}; handle_leader({transfer_leadership, Leader}, #{cfg := #cfg{id = Leader, log_id = LogId}} = State) -> ?DEBUG("~ts: transfer leadership requested but already leader", [LogId]), {leader, State, [{reply, already_leader}]}; handle_leader({transfer_leadership, Member}, #{cfg := #cfg{log_id = LogId}, cluster := Members} = State) when not is_map_key(Member, Members) -> ?DEBUG("~ts: transfer leadership requested but unknown member ~tw", [LogId, Member]), {leader, State, [{reply, {error, unknown_member}}]}; handle_leader({transfer_leadership, ServerId}, #{cfg := #cfg{log_id = LogId}, cluster := Cluster} = State) -> case Cluster of #{ServerId := #{voter_status := #{membership := Membership}}} when Membership =/= voter -> ?INFO("~ts: transfer leadership requested but non-voter member ~tw", [LogId, ServerId]), {leader, State, [{reply, {error, non_voter}}]}; _ -> ?DEBUG("~ts: transfer leadership to ~tw requested", [LogId, ServerId]), %% TODO find a timeout {await_condition, State#{condition => #{predicate_fun => fun transfer_leadership_condition/2, timeout => #{effects => [], transition_to => leader}}}, [{reply, ok}, {send_msg, ServerId, election_timeout, cast}]} end; handle_leader(force_member_change, State0) -> {follower, State0#{votes => 0}, [{next_event, force_member_change}]}; handle_leader(Msg, State) -> log_unhandled_msg(leader, Msg, State), {leader, State, [{reply, {error, {unsupported_call, Msg}}}]}. -spec handle_candidate(ra_msg() | election_timeout, ra_server_state()) -> {ra_state(), ra_server_state(), effects()}. handle_candidate(#request_vote_result{term = Term, vote_granted = true}, #{cfg := #cfg{id = Id, log_id = LogId}, current_term := Term, votes := Votes, cluster := Nodes} = State0) -> NewVotes = Votes + 1, ?DEBUG("~ts: vote granted for term ~b votes ~b", [LogId, Term, NewVotes]), case required_quorum(Nodes) of NewVotes -> State = initialise_peers(State0#{leader_id => Id}), Effects = post_election_effects(State), {leader, maps:without([votes], State), Effects}; _ -> {candidate, State0#{votes => NewVotes}, []} end; handle_candidate(#request_vote_result{term = Term}, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when Term > CurTerm -> ?INFO("~ts: candidate request_vote_result with higher term" " received ~b -> ~b", [LogId, CurTerm, Term]), State = update_term_and_voted_for(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_term_and_voted_for(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(#heartbeat_rpc{term = Term} = Msg, #{current_term := CurTerm} = State0) when Term >= CurTerm -> State = update_term_and_voted_for(Term, undefined, State0), {follower, State, [{next_event, Msg}]}; handle_candidate(#heartbeat_rpc{leader_id = LeaderId, query_index = QueryIndex}, #{current_term := CurTerm} = State) -> % term must be older return success=false Reply = heartbeat_reply(CurTerm, QueryIndex), {candidate, State, [cast_reply(id(State), LeaderId, Reply)]}; handle_candidate({_PeerId, #heartbeat_reply{term = Term}}, #{cfg := #cfg{log_id = LogId}, current_term := CurTerm} = State0) when Term > CurTerm -> ?INFO("~ts: candidate heartbeat_reply with higher" " term received ~b -> ~b", [LogId, CurTerm, Term]), State = update_term_and_voted_for(Term, undefined, State0), {follower, State, []}; handle_candidate({_PeerId, #append_entries_reply{term = Term}}, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when Term > CurTerm -> ?INFO("~ts: candidate append_entries_reply with higher" " term received ~b -> ~b", [LogId, CurTerm, Term]), State = update_term_and_voted_for(Term, undefined, State0), {follower, State, []}; handle_candidate(#request_vote_rpc{term = Term} = Msg, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when Term > CurTerm -> ?INFO("~ts: candidate request_vote_rpc with higher term received ~b -> ~b", [LogId, CurTerm, Term]), State = update_term_and_voted_for(Term, undefined, State0), {follower, State, [{next_event, Msg}]}; handle_candidate(#pre_vote_rpc{term = Term} = Msg, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when Term > CurTerm -> ?INFO("~ts: candidate pre_vote_rpc with higher term received ~b -> ~b", [LogId, CurTerm, Term]), State = update_term_and_voted_for(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(#pre_vote_rpc{} = PreVote, State) -> %% unlike request_vote_rpc, a candidate cannot simply reject %% a pre_vote_rpc that does not have a higher term %% (see https://github.com/rabbitmq/ra/issues/439 for the detail) process_pre_vote(candidate, PreVote, State); handle_candidate(#request_vote_result{}, State) -> %% handle to avoid logging as unhandled {candidate, State, []}; handle_candidate(#pre_vote_result{}, State) -> %% handle to avoid logging as unhandled {candidate, State, []}; handle_candidate(#install_snapshot_rpc{term = Term} = ISR, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when Term >= CurTerm -> ?INFO("~ts: candidate received install_snapshot_rpc for term ~b," " stepping down from term ~b", [LogId, Term, CurTerm]), {follower, State0#{votes => 0}, [{next_event, ISR}]}; handle_candidate(#install_snapshot_rpc{term = Term, meta = #{index := LastIndex, term := LastTerm}}, #{cfg := #cfg{log_id = LogId}, current_term := CurTerm} = State) when Term < CurTerm -> ?DEBUG("~ts: candidate install_snapshot old term ~b in ~b", [LogId, LastIndex, LastTerm]), Reply = #install_snapshot_result{term = CurTerm, last_term = LastTerm, last_index = LastIndex}, {candidate, State, [{reply, Reply}]}; handle_candidate({ra_log_event, Evt}, State = #{log := Log0}) -> % simply forward all other events to ra_log {Log, Effects} = ra_log:handle_event(Evt, Log0), {candidate, State#{log => Log}, Effects}; handle_candidate(election_timeout, State) -> call_for_election(candidate, State); handle_candidate(force_member_change, State0) -> {follower, State0#{votes => 0}, [{next_event, force_member_change}]}; handle_candidate(#info_rpc{term = Term} = Msg, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when CurTerm < Term -> ?INFO("~ts: candidate info_rpc with higher term received ~b -> ~b", [LogId, CurTerm, Term]), State = update_term_and_voted_for(Term, undefined, State0), {follower, State, [{next_event, Msg}]}; handle_candidate(#info_rpc{} = InfoRpc, State) -> InfoReplyEffect = empty_info_reply_effect(State, InfoRpc), {candidate, State, [InfoReplyEffect]}; handle_candidate(#info_reply{term = Term} = Msg, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when CurTerm < Term -> ?INFO("~ts: candidate info_reply with higher term received ~b -> ~b", [LogId, CurTerm, Term]), State = update_term_and_voted_for(Term, undefined, State0), {follower, State, [{next_event, Msg}]}; handle_candidate(#info_reply{}, State) -> {candidate, State, []}; handle_candidate(Msg, State) -> log_unhandled_msg(candidate, Msg, State), {candidate, State, [{reply, {error, {unsupported_call, Msg}}}]}. -spec handle_pre_vote(ra_msg(), ra_server_state()) -> {ra_state(), ra_server_state(), 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(#heartbeat_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(#heartbeat_rpc{leader_id = LeaderId, query_index = QueryIndex}, #{current_term := CurTerm} = State) -> Reply = heartbeat_reply(CurTerm, QueryIndex), {pre_vote, State, [cast_reply(id(State), LeaderId, Reply)]}; handle_pre_vote({_PeerId, #heartbeat_reply{term = Term}}, #{current_term := CurTerm} = State) when Term > CurTerm -> {follower, update_term(Term, State#{votes => 0}), []}; 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, cfg := #cfg{log_id = LogId}, pre_vote_token := Token, cluster := Nodes, membership := voter} = State0) -> ?DEBUG("~ts: pre_vote granted ~tw for term ~b votes ~b", [LogId, Token, Term, Votes + 1]), NewVotes = Votes + 1, State = update_term(Term, State0), case required_quorum(Nodes) of NewVotes -> call_for_election(candidate, State); _ -> {pre_vote, State#{votes => NewVotes}, []} end; handle_pre_vote(#pre_vote_result{}, State) -> %% just handle negative results to avoid printing an unhandled message log {pre_vote, State, []}; handle_pre_vote(#pre_vote_rpc{} = PreVote, State) -> process_pre_vote(pre_vote, PreVote, State); handle_pre_vote(#request_vote_result{}, State) -> %% handle to avoid logging as unhandled {pre_vote, State, []}; handle_pre_vote(election_timeout, State) -> call_for_election(pre_vote, State); handle_pre_vote({ra_log_event, Evt}, State = #{log := Log0}) -> % simply forward all other events to ra_log {Log, Effects} = ra_log:handle_event(Evt, Log0), {pre_vote, State#{log => Log}, Effects}; handle_pre_vote(force_member_change, State0) -> {follower, State0#{votes => 0}, [{next_event, force_member_change}]}; handle_pre_vote(#info_rpc{term = Term} = Msg, #{current_term := CurTerm} = State0) when CurTerm < Term -> {follower, State0#{votes => 0}, [{next_event, Msg}]}; handle_pre_vote(#info_rpc{} = InfoRpc, State) -> InfoReplyEffect = empty_info_reply_effect(State, InfoRpc), {pre_vote, State, [InfoReplyEffect]}; handle_pre_vote(#info_reply{term = Term} = Msg, #{current_term := CurTerm} = State0) when CurTerm < Term -> {follower, State0#{votes => 0}, [{next_event, Msg}]}; handle_pre_vote(#info_reply{}, State) -> {pre_vote, State, []}; handle_pre_vote(Msg, State) -> log_unhandled_msg(pre_vote, Msg, State), {pre_vote, State, [{reply, {error, {unsupported_call, Msg}}}]}. -spec handle_follower(ra_msg(), ra_server_state()) -> {ra_state(), ra_server_state(), effects()}. handle_follower(#append_entries_rpc{term = Term, leader_id = LeaderId, leader_commit = LeaderCommit, prev_log_index = PLIdx, prev_log_term = PLTerm, entries = Entries0}, #{cfg := #cfg{log_id = LogId, id = Id} = Cfg, log := Log00, last_applied := LastApplied, current_term := CurTerm} = State00) when Term >= CurTerm -> ok = incr_counter(Cfg, ?C_RA_SRV_AER_RECEIVED_FOLLOWER, 1), %% this is a valid leader, append entries message Effects0 = [{record_leader_msg, LeaderId}], State0 = update_term(Term, State00#{leader_id => LeaderId}), case has_log_entry_or_snapshot(PLIdx, PLTerm, Log00) of {entry_ok, Log0} -> % filter entries already seen {Log1, Entries, LastValidIdx} = drop_existing(Log0, Entries0, PLIdx), case Entries of [] -> %% all entries have already been written ok = incr_counter(Cfg, ?C_RA_SRV_AER_RECEIVED_FOLLOWER_EMPTY, 1), {LocalLastIdx, _} = ra_log:last_index_term(Log1), {LogIsValidated, Log2} = case Entries0 of [] when LocalLastIdx > PLIdx -> %% if no entries were sent we need to reset %% last index to match the leader ?DEBUG("~ts: resetting last index to ~b from ~b " "in term ~b", [LogId, PLIdx, LocalLastIdx, Term]), ?assertNot(PLIdx < LastApplied), {ok, L} = ra_log:set_last_index(PLIdx, Log1), %% a reset means we've validated the last index {true, L}; _ -> %% local last index is lower or equal to the %% last validated index %% in this case we can take the updated leader %% commit as the end of the local log has been %% validated {LocalLastIdx =< LastValidIdx, Log1} end, case LogIsValidated of true -> State1 = State0#{log => Log2, commit_index => LeaderCommit}, ok = put_counter(Cfg, ?C_RA_SVR_METRIC_COMMIT_INDEX, LeaderCommit), %% evaluate commit index as we may have received %% an updated commit_index for previously %% written entries {NextState, State, Effects} = evaluate_commit_index_follower(State1, Effects0), %% log is validated so send a successful reply Reply = append_entries_reply(Term, true, State), {NextState, State, [cast_reply(Id, LeaderId, Reply) | Effects]}; false -> %% We need to ensure we make progress in case the %% leader is having to resend already received %% entries in order to validate, e.g. after a %% term_mismatch, hence we reply with success but %% only up to the last index we already had LastValidatedIdx = max(LastApplied, LastValidIdx), ?DEBUG("~ts: append_entries_rpc with last index ~b " " including ~b entries did not validate local log. " "Local last index ~b", [LogId, PLIdx, length(Entries0), LocalLastIdx]), {LVTerm, State} = fetch_term(LastValidatedIdx, State0#{log => Log2}), Reply = #append_entries_reply{term = CurTerm, success = true, next_index = LastValidatedIdx + 1, last_index = LastValidatedIdx, last_term = LVTerm}, {follower, State, [cast_reply(Id, LeaderId, Reply)]} end; [{FstIdx, _, _} | _] -> State1 = State0#{commit_index => LeaderCommit}, ok = put_counter(Cfg, ?C_RA_SVR_METRIC_COMMIT_INDEX, LeaderCommit), %% assert we're not writing below the last applied index ?assertNot(FstIdx < LastApplied), State = lists:foldl(fun pre_append_log_follower/2, State1, Entries), case ra_log:write(Entries, Log1) of {ok, Log2} -> evaluate_commit_index_follower(State#{log => Log2}, Effects0); {error, wal_down} -> %% At this point we know the wal process exited %% but we dont know exactly which in flight messages %% made it to the wal before it crashed. {await_condition, State#{log => Log1, condition => #{predicate_fun => fun wal_down_condition/2}}, Effects0}; {error, _} = Err -> exit(Err) end end; {missing, Log0} -> State = State0#{log => Log0}, Reply = append_entries_reply(Term, false, State), ?INFO("~ts: follower did not have entry at ~b in ~b." " Requesting ~tw from ~b", [LogId, PLIdx, PLTerm, LeaderId, Reply#append_entries_reply.next_index]), Effects = [cast_reply(Id, LeaderId, Reply) | Effects0], {await_condition, State#{condition => #{predicate_fun => follower_catchup_cond_fun(missing), % repeat reply effect on condition timeout timeout => #{effects => Effects, transition_to => follower}}}, Effects}; {term_mismatch, OtherTerm, Log0} -> State1 = State0#{log => Log0}, LastApplied = maps:get(last_applied, State1), ?INFO("~ts: term mismatch - follower had entry at ~b with term ~b " "but not with term ~b. " "Asking leader ~tw to resend from ~b", [LogId, PLIdx, OtherTerm, PLTerm, LeaderId, LastApplied + 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 last applied as the last index % and last applied + 1 as the next expected. % This _may_ overwrite some valid entries but is probably the % simplest and most reliable way to proceed {Reply, State} = mismatch_append_entries_reply(Term, LastApplied, State1), Effects = [cast_reply(Id, LeaderId, Reply) | Effects0], {await_condition, State#{condition => #{predicate_fun => follower_catchup_cond_fun(term_mismatch), % repeat reply effect on condition timeout timeout => #{effects => Effects, transition_to => follower}}}, Effects} end; handle_follower(#append_entries_rpc{term = Term, leader_id = LeaderId}, #{cfg := #cfg{id = Id, log_id = LogId} = Cfg, current_term := CurTerm} = State) -> ok = incr_counter(Cfg, ?C_RA_SRV_AER_RECEIVED_FOLLOWER, 1), % the term is lower than current term Reply = append_entries_reply(CurTerm, false, State), ?DEBUG("~ts: follower got append_entries_rpc from ~tw in" " ~b but current term is: ~b", [LogId, LeaderId, Term, CurTerm]), {follower, State, [cast_reply(Id, LeaderId, Reply)]}; handle_follower(#heartbeat_rpc{query_index = QueryIndex, term = Term, leader_id = LeaderId}, #{current_term := CurTerm, cfg := #cfg{id = Id}} = State0) when Term >= CurTerm -> State1 = update_term(Term, State0), State2 = State1#{leader_id => LeaderId}, Reply = heartbeat_reply(Term, QueryIndex), {follower, State2, [cast_reply(Id, LeaderId, Reply)]}; handle_follower(#heartbeat_rpc{leader_id = LeaderId, query_index = QueryIndex}, #{current_term := CurTerm, cfg := #cfg{id = Id}} = State) -> Reply = heartbeat_reply(CurTerm, QueryIndex), {follower, State, [cast_reply(Id, LeaderId, Reply)]}; handle_follower({ra_log_event, Evt}, #{log := Log0, cfg := #cfg{id = Id}, current_term := Term} = State0) -> LeaderId = maps:get(leader_id, State0, undefined), % forward events to ra_log % if the last written changes then send an append entries reply LW = ra_log:last_written(Log0), {Log, Effects} = ra_log:handle_event(Evt, Log0), State = State0#{log => Log}, {State1, Effects1} = maybe_emit_pending_release_cursor(State, Effects), case LW =/= ra_log:last_written(Log) of true when LeaderId =/= undefined -> %% last written has changed so we need to send an AER reply Reply = append_entries_reply(Term, true, State1), {follower, State1, [cast_reply(Id, LeaderId, Reply) | Effects1]}; _ -> {follower, State1, Effects1} end; handle_follower(#pre_vote_rpc{}, #{cfg := #cfg{log_id = LogId}, membership := Membership} = State) when Membership =/= voter -> ?DEBUG("~ts: follower ignored pre_vote_rpc, non-voter: ~0p", [LogId, Membership]), {follower, State, []}; handle_follower(#pre_vote_rpc{} = PreVote, State) -> process_pre_vote(follower, PreVote, State); handle_follower(#request_vote_rpc{}, #{cfg := #cfg{log_id = LogId}, membership := Membership} = State) when Membership =/= voter -> ?DEBUG("~ts: follower ignored request_vote_rpc, non-voter: ~0p", [LogId, Membership]), {follower, State, []}; handle_follower(#request_vote_rpc{candidate_id = Cand, term = Term}, #{current_term := Term, voted_for := VotedFor, cfg := #cfg{log_id = LogId}} = State) when VotedFor /= undefined andalso VotedFor /= Cand -> % already voted for another in this term ?DEBUG("~ts: follower request_vote_rpc for ~tw already voted for ~tw in ~b", [LogId, 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, cfg := #cfg{log_id = LogId}} = State0) when Term >= CurTerm -> State1 = update_term(Term, State0), LastIdxTerm = last_idx_term(State1), case is_candidate_log_up_to_date(LLIdx, LLTerm, LastIdxTerm) of true -> ?INFO("~ts: granting vote for ~tw with last {index, term} ~w" " for term ~b previous term was ~b", [LogId, Cand, {LLIdx, LLTerm}, Term, CurTerm]), Reply = #request_vote_result{term = Term, vote_granted = true}, State = update_term_and_voted_for(Term, Cand, State1), {follower, State, [{reply, Reply}]}; false -> ?INFO("~ts: declining vote for ~tw for term ~b," " candidate last log {index, term} was: ~w " " last log entry {index, term} is: ~w", [LogId, Cand, Term, {LLIdx, LLTerm}, {LastIdxTerm}]), Reply = #request_vote_result{term = Term, vote_granted = false}, {follower, update_term(Term, State1), [{reply, Reply}]} end; handle_follower(#request_vote_rpc{term = Term, candidate_id = Candidate}, State = #{current_term := CurTerm, cfg := #cfg{log_id = LogId}}) when Term < CurTerm -> ?INFO("~ts: declining vote to ~tw for term ~b, current term ~b", [LogId, Candidate, Term, CurTerm]), Reply = #request_vote_result{term = CurTerm, vote_granted = false}, {follower, State, [{reply, Reply}]}; handle_follower({_PeerId, #append_entries_reply{term = TheirTerm}}, State = #{current_term := CurTerm}) -> Term = max(TheirTerm, CurTerm), {follower, update_term(Term, State), []}; handle_follower({_PeerId, #heartbeat_reply{term = TheirTerm}}, State = #{current_term := CurTerm}) -> Term = max(TheirTerm, CurTerm), {follower, update_term(Term, State), []}; handle_follower(#install_snapshot_rpc{term = Term, meta = #{index := LastIndex, term := LastTerm}}, #{cfg := #cfg{log_id = LogId}, current_term := CurTerm} = State) when Term < CurTerm -> ?DEBUG("~ts: install_snapshot old term ~b in ~b", [LogId, LastIndex, LastTerm]), % follower receives a snapshot from an old term Reply = #install_snapshot_result{term = CurTerm, last_term = LastTerm, last_index = LastIndex}, {follower, State, [{reply, Reply}]}; handle_follower(#install_snapshot_rpc{term = Term, meta = #{index := SnapIdx, machine_version := SnapMacVer} = Meta, leader_id = LeaderId, chunk_state = {Num, ChunkFlag}} = Rpc, #{cfg := #cfg{log_id = LogId, machine_version = MacVer}, log := Log0, last_applied := LastApplied, current_term := CurTerm} = State0) when Term >= CurTerm andalso SnapIdx > LastApplied andalso %% only install snapshot if the machine version is understood MacVer >= SnapMacVer andalso Num =< 1 andalso ChunkFlag /= pre -> %% only begin snapshot procedure if Idx is higher than the last_applied %% index. ?DEBUG("~ts: begin_accept snapshot at index ~b in term ~b, phase ~s", [LogId, SnapIdx, Term, ChunkFlag]), SnapState0 = ra_log:snapshot_state(Log0), {ok, SS} = ra_snapshot:begin_accept(Meta, SnapState0), Log1 = ra_log:set_snapshot_state(SS, Log0), %% TODO: just resetting to lastapplied may not be enough if all prior %% entries are not fully written {_LastWrittenIdx, _} = ra_log:last_written(Log1), %% if the snaphost includes pre entries (live entries) then we need %% to reset the log to the last applied index to avoid issues Log = case ChunkFlag of init -> {ok, L} = ra_log:set_last_index(LastApplied, Log1), L; _ -> Log1 end, {receive_snapshot, update_term(Term, State0#{log => Log, snapshot_phase => ChunkFlag, leader_id => LeaderId}), [{next_event, Rpc}, {record_leader_msg, LeaderId}]}; handle_follower(#install_snapshot_rpc{term = Term, meta = #{index := SnapIdx, machine_version := SnapMacVer, term := _LastTerm}}, #{cfg := #cfg{log_id = LogId, machine_version = MacVer}, last_applied := LastApplied} = State0) when MacVer >= SnapMacVer -> ?DEBUG("~ts: install_snapshot received with snapshot index ~b, in ~b, local last applied index ~b ", [LogId, SnapIdx, Term, LastApplied]), %% follower receives a snapshot for an index lower than its last applied %% index, just reply with append_entries_reply to make the leader skip %% ahead {Reply, State} = mismatch_append_entries_reply(Term, LastApplied, State0), {follower, update_term(Term, State), [{reply, Reply}]}; handle_follower(#request_vote_result{}, State) -> %% handle to avoid logging as unhandled {follower, State, []}; handle_follower(#pre_vote_result{}, State) -> %% handle to avoid logging as unhandled {follower, State, []}; handle_follower(#append_entries_reply{}, State) -> %% handle to avoid logging as unhandled %% could receive a lot of these shortly after standing down as leader {follower, State, []}; handle_follower(election_timeout, #{cfg := #cfg{log_id = LogId}, membership := Membership} = State) when Membership =/= voter -> ?INFO("~ts: follower ignored election_timeout, non-voter: ~0p", [LogId, Membership]), {follower, State, []}; handle_follower(election_timeout, State) -> call_for_election(pre_vote, State); handle_follower(try_become_leader, State) -> handle_follower(election_timeout, State); handle_follower(force_member_change, #{cfg := #cfg{id = Id, uid = Uid, log_id = LogId}} = State0) -> Cluster = #{Id => new_peer_with(#{voter_status => #{uid => Uid, membership => voter}})}, ?WARN("~ts: Forcing cluster change. New cluster ~tw", [LogId, Cluster]), {ok, _, _, State, Effects} = append_cluster_change(Cluster, undefined, no_reply, State0, []), call_for_election(pre_vote, State#{membership => voter}, [{reply, ok} | Effects]); handle_follower(#info_rpc{term = Term} = Msg, #{current_term := CurTerm} = State) when CurTerm < Term -> State1 = update_term(Term, State), {follower, State1, [{next_event, Msg}]}; handle_follower(#info_rpc{} = InfoRpc, State) -> InfoReplyEffect = info_reply_effect(State, InfoRpc), {follower, State, [InfoReplyEffect]}; handle_follower(#info_reply{term = Term} = Msg, #{current_term := CurTerm} = State) when CurTerm < Term -> State1 = update_term(Term, State), {follower, State1, [{next_event, Msg}]}; handle_follower(#info_reply{}, State) -> {follower, State, []}; handle_follower(Msg, State) -> log_unhandled_msg(follower, Msg, State), {follower, State, [{reply, {error, {unsupported_call, Msg}}}]}. handle_receive_snapshot(#install_snapshot_rpc{term = Term, meta = #{index := SnapIndex, machine_version := SnapMacVer, cluster := ClusterIds, term := SnapTerm} = SnapMeta, chunk_state = {Num, ChunkFlag}, data = ChunkOrEntries} = Rpc, #{cfg := #cfg{id = Id, log_id = LogId, effective_machine_version = CurEffMacVer, machine_versions = MachineVersions, machine = Machine} = Cfg0, log := Log00, cluster := Cluster, current_term := CurTerm, last_applied := LastApplied, machine_state := OldMacState, snapshot_phase := SnapPhase} = State0) when Term >= CurTerm -> Reply = #install_snapshot_result{term = CurTerm, last_term = SnapTerm, last_index = SnapIndex}, SnapState0 = ra_log:snapshot_state(Log00), %% works as an assertion also {AcceptingSnapIdx, _} = ra_snapshot:accepting(SnapState0), SnapshotHasLiveIndexes = maps:get(snapshot_has_live_indexes, State0, false), LogHasPendingIndexes = ra_log:has_pending(Log00), case ChunkFlag of init when SnapPhase == init andalso SnapIndex == AcceptingSnapIdx -> %% this is ok, just reply, %% if we receive an init phase we can assume the snapshot contains %% live indexes {receive_snapshot, State0#{snapshot_has_live_indexes => true}, [{reply, Reply}]}; init -> ?DEBUG("~ts: receiving snapshot saw unexpected init phase at snapshot " "index term {~b, ~b}, current phase ~s restarting " "snapshot receive phase", [LogId, SnapIndex, SnapTerm, SnapPhase]), %% the snapshot sending must have been interrupted and restarted %% during the init or pre-phase %% abort the snapshot, and revert to follower State = abort_receive(State0), {follower, State, [{next_event, Rpc}]}; pre when is_list(ChunkOrEntries) -> ?assert(SnapIndex == AcceptingSnapIdx), % ?DEBUG("~ts: receiving snapshot chunk pre first index ~b snap index ~b, term ~b", % [LogId, FstIdx, SnapIndex, SnapTerm]), %% reset last index to last applied %% as we dont know for sure indexes after last applied %% are of the right term {LastIdx, _} = ra_log:last_index_term(Log00), {Log, _} = lists:foldl( fun ({I, _, _} = E, {L0, LstIdx}) when I > LastApplied -> {ok, L} = ra_log:write_sparse(E, LstIdx, L0), {L, I}; (_, Acc) -> %% drop any entries that are lower than last applied Acc end, {Log00, LastIdx}, ChunkOrEntries), State = update_term(Term, State0#{log => Log, snapshot_phase => pre}), {receive_snapshot, State, [{reply, Reply}]}; next -> ?assert(SnapIndex == AcceptingSnapIdx), ?DEBUG("~ts: receiving snapshot chunk: ~b / ~w, index ~b, term ~b", [LogId, Num, ChunkFlag, SnapIndex, SnapTerm]), SnapState = ra_snapshot:accept_chunk(ChunkOrEntries, Num, SnapState0), Log0 = ra_log:set_snapshot_state(SnapState, Log00), State = update_term(Term, State0#{log => Log0, snapshot_phase => next}), {receive_snapshot, State, [{reply, Reply}]}; last when SnapshotHasLiveIndexes andalso LogHasPendingIndexes -> %% we cannot yet complete the snapshot as there are pending %% log indexes, defer completion until they are written EventType = maps:get(current_event_type, State0), ?DEBUG("~ts: receiving snapshot chunk: ~b / ~w, index ~b, term ~b " "cannot yet complete as log has pending indexes", [LogId, Num, ChunkFlag, SnapIndex, SnapTerm]), {receive_snapshot, State0#{snapshot_phase => {awaiting_pending, EventType, Rpc}}, []}; last -> ?assert(SnapIndex == AcceptingSnapIdx), ?DEBUG("~ts: receiving snapshot chunk: ~b / ~w, index ~b, term ~b", [LogId, Num, ChunkFlag, SnapIndex, SnapTerm]), {SnapState, MacState, LiveIndexes, Effs0} = ra_snapshot:complete_accept(ChunkOrEntries, Num, Machine, SnapState0), Log0 = ra_log:set_snapshot_state(SnapState, Log00), %% if the machine version of the snapshot is higher %% we also need to update the current effective machine configuration EffMacMod = ra_machine:which_module(Machine, SnapMacVer), Cfg = case SnapMacVer > CurEffMacVer of true -> put_counter(Cfg0, ?C_RA_SVR_METRIC_EFFECTIVE_MACHINE_VERSION, SnapMacVer), Cfg0#cfg{effective_machine_version = SnapMacVer, machine_versions = [{SnapIndex, SnapMacVer} | MachineVersions], effective_machine_module = EffMacMod, effective_handle_aux_fun = ra_machine:which_aux_fun(EffMacMod)}; false -> Cfg0 end, %% this is the last chunk so we can "install" it {ok, Log, Effs} = ra_log:install_snapshot({SnapIndex, SnapTerm}, EffMacMod, LiveIndexes, Log0), OldServerIds = maps:map(fun (_, V) -> maps:with([voter_status], V) end, Cluster), OldMeta = #{machine_version => CurEffMacVer, term => CurTerm, index => LastApplied, cluster => OldServerIds}, SnapInstalledEffs = ra_machine:snapshot_installed(EffMacMod, SnapMeta, MacState, OldMeta, OldMacState), State1 = update_term(Term, State0#{cfg => Cfg, log => Log, commit_index => SnapIndex, last_applied => SnapIndex, %% this may not be the actual %% cluster index cluster_index_term => {SnapIndex, SnapTerm}, cluster => make_cluster(Id, ClusterIds), membership => get_membership(ClusterIds, State0), machine_state => MacState}), State = maps:without([snapshot_phase, snapshot_has_live_indexes, current_event_type], State1), put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_APPLIED, SnapIndex), %% it was the last snapshot chunk so we can revert back to %% follower status {follower, persist_last_applied(State), [{reply, Reply} | Effs0 ++ Effs ++ SnapInstalledEffs]} end; handle_receive_snapshot(#append_entries_rpc{term = Term} = Msg, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when Term > CurTerm -> ?INFO("~ts: follower receiving snapshot saw append_entries_rpc from ~tw " "for term ~b abdicates term: ~b!", [LogId, Msg#append_entries_rpc.leader_id, Term, CurTerm]), State = abort_receive(State0), {follower, update_term(Term, State), [{next_event, Msg}]}; handle_receive_snapshot({ra_log_event, Evt}, #{cfg := #cfg{log_id = LogId}, snapshot_phase := Phase, log := Log0} = State0) -> %% forward log events to ra_log whilst the snapshot is being received {Log, Effects} = ra_log:handle_event(Evt, Log0), case Phase of {awaiting_pending, EventType, DeferredRpc} -> case ra_log:has_pending(Log) of false -> ?DEBUG("~ts: pending indexes cleared, completing snapshot", [LogId]), %% replay the deferred last chunk, set phase to next so %% the replayed event doesn't hit the awaiting_pending guard {receive_snapshot, State0#{log => Log, snapshot_phase => next}, [{next_event, EventType, DeferredRpc} | Effects]}; true -> {receive_snapshot, State0#{log => Log}, Effects} end; _ -> {receive_snapshot, State0#{log => Log}, Effects} end; handle_receive_snapshot(receive_snapshot_timeout, #{cfg := #cfg{log_id = LogId}} = State0) -> ?INFO("~ts: ~s receive snapshot timed out.", [LogId, ?FUNCTION_NAME]), State = abort_receive(State0), {follower, State, []}; handle_receive_snapshot(#info_rpc{term = Term} = Msg, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when CurTerm < Term -> ?INFO("~ts: follower receiving snapshot saw info_rpc from ~tw for term ~b " "current term: ~b!", [LogId, Msg#info_rpc.from, Term, CurTerm]), State = abort_receive(State0), {follower, update_term(Term, State), [{next_event, Msg}]}; handle_receive_snapshot(#info_rpc{} = InfoRpc, State) -> InfoReplyEffect = empty_info_reply_effect(State, InfoRpc), {receive_snapshot, State, [InfoReplyEffect]}; handle_receive_snapshot(#info_reply{term = Term} = Msg, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when CurTerm < Term -> ?INFO("~ts: follower receiving snapshot saw info_reply from ~tw for term " "~b abdicates term: ~b!", [LogId, Msg#info_reply.from, Term, CurTerm]), State = abort_receive(State0), {follower, update_term(Term, State), [{next_event, Msg}]}; handle_receive_snapshot(#info_reply{}, State) -> {receive_snapshot, State, []}; %% Handle request_vote_rpc with higher term - abort snapshot and participate in election handle_receive_snapshot(#request_vote_rpc{term = Term} = Msg, #{current_term := CurTerm, cfg := #cfg{log_id = LogId}} = State0) when Term > CurTerm -> ?INFO("~ts: receive_snapshot saw request_vote_rpc for term ~b, " "current term ~b, aborting snapshot receive", [LogId, Term, CurTerm]), State = abort_receive(State0), {follower, update_term(Term, State), [{next_event, Msg}]}; %% Handle request_vote_rpc with lower or equal term - reject without aborting handle_receive_snapshot(#request_vote_rpc{}, #{current_term := CurTerm} = State) -> Reply = #request_vote_result{term = CurTerm, vote_granted = false}, {receive_snapshot, State, [{reply, Reply}]}; %% Handle pre_vote_rpc without leaving receive_snapshot state handle_receive_snapshot(#pre_vote_rpc{} = PreVote, State) -> process_pre_vote(receive_snapshot, PreVote, State); handle_receive_snapshot(Msg, State) -> log_unhandled_msg(receive_snapshot, Msg, State), {receive_snapshot, State, []}. abort_receive(#{snapshot_phase := _Phase, last_applied := LastApplied, log := Log0} = State) -> SnapState0 = ra_log:snapshot_state(Log0), SnapState = ra_snapshot:abort_accept(SnapState0), Log1 = ra_log:set_snapshot_state(SnapState, Log0), Log = case State of #{snapshot_has_live_indexes := true} -> %% live indexes were written during pre phase, %% reset log index to undo them {ok, Log2} = ra_log:set_last_index(LastApplied, Log1), Log2; _ -> Log1 end, clear_leader_id(maps:without([snapshot_phase, snapshot_has_live_indexes, current_event_type], State#{log => Log})). -spec handle_await_condition(ra_msg(), ra_server_state()) -> {ra_state(), ra_server_state(), effects()}. handle_await_condition(#request_vote_rpc{} = Msg, State) -> {follower, State, [{next_event, Msg}]}; handle_await_condition(#pre_vote_rpc{} = PreVote, State) -> process_pre_vote(await_condition, PreVote, State); handle_await_condition(election_timeout, #{cfg := #cfg{log_id = LogId}, membership := Membership} = State) when Membership =/= voter -> ?DEBUG("~s: await_condition ignored election_timeout, replicate membership state: ~p", [LogId, Membership]), {await_condition, State, []}; handle_await_condition(election_timeout, State) -> call_for_election(pre_vote, State); handle_await_condition(await_condition_timeout = Msg, #{condition := #{predicate_fun := Pred} = Cond} = State0) -> case Pred(Msg, State0) of {true, State1} -> CondTransitionTo = maps:get(transition_to, Cond, follower), State = maps:remove(condition, State1), {CondTransitionTo, State, []}; {false, State1} -> Timeout = maps:get(timeout, Cond, #{}), Effects = maps:get(effects, Timeout, []), TransitionTo = maps:get(transition_to, Timeout, follower), State = maps:remove(condition, State1), {TransitionTo, State, Effects} end; handle_await_condition({ra_log_event, Evt}, State = #{log := Log0}) -> % simply forward all other events to ra_log {Log, Effects} = ra_log:handle_event(Evt, Log0), {await_condition, State#{log => Log}, Effects}; handle_await_condition(Msg, #{condition := #{predicate_fun := Pred} = Cond} = State0) -> case Pred(Msg, State0) of {true, State1} -> TransitionTo = maps:get(transition_to, Cond, follower), State = maps:remove(condition, State1), {TransitionTo, State, [{next_event, Msg}]}; {false, State} -> %% do not log unhandled messages as they are often expected {await_condition, State, []} end. -spec process_new_leader_queries(ra_server_state()) -> {ra_server_state(), [from()]}. process_new_leader_queries(#{pending_consistent_queries := Pending, queries_waiting_heartbeats := Waiting} = State0) -> From0 = lists:map(fun({_, From, _, _}) -> From end, Pending), From1 = lists:map(fun({_, {_, From, _, _}}) -> From end, queue:to_list(Waiting)), {State0#{pending_consistent_queries => [], queries_waiting_heartbeats => queue:new()}, From0 ++ From1}. -spec tick(ra_server_state()) -> effects(). tick(#{cfg := #cfg{effective_machine_module = MacMod}, machine_state := MacState} = State) -> InfoRpcEffects = info_rpc_effects(State), Now = erlang:system_time(millisecond), InfoRpcEffects ++ ra_machine:tick(MacMod, Now, MacState). -spec log_tick(ra_server_state()) -> ra_server_state(). log_tick(#{cfg := #cfg{}, log := Log0} = State) -> Now = erlang:system_time(millisecond), Log = ra_log:tick(Now, Log0), State#{log => Log}. -spec handle_state_enter(ra_state() | eol, ra_state(), ra_server_state()) -> {ra_server_state() | eol, effects()}. handle_state_enter(RaftState, OldRaftState, #{cfg := #cfg{effective_machine_module = MacMod}, machine_state := MacState} = State) -> {become(RaftState, OldRaftState, State), ra_machine:state_enter(MacMod, RaftState, MacState)}. -spec overview(ra_server_state()) -> map(). overview(#{cfg := #cfg{effective_machine_module = MacMod} = Cfg, log := Log, machine_state := MacState, aux_state := Aux, queries_waiting_heartbeats := Queries, pending_consistent_queries := PendingQueries } = State) -> NumQueries = queue:len(Queries), O0 = maps:with([current_term, commit_index, last_applied, cluster, leader_id, voted_for, membership, cluster_change_permitted, cluster_index_term, query_index ], State), O = maps:merge(O0, cfg_to_map(Cfg)), LogOverview = ra_log:overview(Log), MacOverview = ra_machine:overview(MacMod, MacState), O#{log => LogOverview, aux => Aux, machine => MacOverview, num_waiting_queries => NumQueries, num_pending_queries => length(PendingQueries)}. cfg_to_map(Cfg) -> element(2, lists:foldl( fun (F, {N, Acc}) -> {N + 1, Acc#{F => element(N, Cfg)}} end, {2, #{}}, record_info(fields, cfg))). -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]), MinCI = lists:min([M || #{commit_index_sent := M} <- Peers]), MinMI >= CI andalso MinCI >= CI end. handle_aux(RaftState, Type, _Cmd, #{cfg := #cfg{effective_handle_aux_fun = undefined}} = State0) -> %% todo reply with error if Type is a call? Effects = case Type of cast -> []; _From -> [{reply, {error, aux_handler_not_implemented}}] end, {RaftState, State0, Effects}; handle_aux(RaftState, Type, MaybeWrappedCmd, #{cfg := #cfg{effective_machine_module = MacMod, effective_handle_aux_fun = {handle_aux, 5}}, aux_state := Aux0} = State0) -> {Wrap, Cmd} = case MaybeWrappedCmd of {'$wrap_reply', C} -> {true, C}; C -> {false, C} end, %% NEW API case ra_machine:handle_aux(MacMod, RaftState, Type, Cmd, Aux0, State0) of {reply, Reply, Aux, State} -> {RaftState, State#{aux_state => Aux}, [{reply, wrap_reply(Wrap, Reply)}]}; {reply, Reply, Aux, State, Effects} -> {RaftState, State#{aux_state => Aux}, [{reply, wrap_reply(Wrap, Reply)} | Effects]}; {no_reply, Aux, State} -> {RaftState, State#{aux_state => Aux}, []}; {no_reply, Aux, State, Effects} -> {RaftState, State#{aux_state => Aux}, Effects} end; handle_aux(RaftState, Type, MaybeWrappedCmd , #{cfg := #cfg{effective_machine_module = MacMod, effective_handle_aux_fun = {handle_aux, 6}}, aux_state := Aux0, machine_state := MacState, log := Log0} = State0) -> {Wrap, Cmd} = case MaybeWrappedCmd of {'$wrap_reply', C} -> {true, C}; C -> {false, C} end, %% OLD API case ra_machine:handle_aux(MacMod, RaftState, Type, Cmd, Aux0, Log0, MacState) of {reply, Reply, Aux, Log} -> {RaftState, State0#{log => Log, aux_state => Aux}, [{reply, wrap_reply(Wrap, Reply)}]}; {reply, Reply, Aux, Log, Effects} -> {RaftState, State0#{log => Log, aux_state => Aux}, [{reply, wrap_reply(Wrap, Reply)} | Effects]}; {no_reply, Aux, Log} -> {RaftState, State0#{log => Log, aux_state => Aux}, []}; {no_reply, Aux, Log, Effects} -> {RaftState, State0#{log => Log, aux_state => Aux}, Effects}; undefined -> {RaftState, State0, []} end. % property helpers wrap_reply(true, Cmd) -> {wrap_reply, Cmd}; wrap_reply(false, Cmd) -> Cmd. -spec id(ra_server_state()) -> ra_server_id(). id(#{cfg := #cfg{id = Id}}) -> Id. -spec log_id(ra_server_state()) -> unicode:chardata(). log_id(#{cfg := #cfg{log_id = LogId}}) -> LogId. -spec uid(ra_server_state()) -> ra_uid(). uid(#{cfg := #cfg{uid = UId}}) -> UId. -spec system_config(ra_server_state()) -> ra_system:config(). system_config(#{cfg := #cfg{system_config = SC}}) -> SC. -spec leader_id(ra_server_state()) -> option(ra_server_id()). leader_id(State) -> maps:get(leader_id, State, undefined). -spec clear_leader_id(ra_server_state()) -> ra_server_state(). clear_leader_id(State) -> State#{leader_id => undefined}. -spec current_term(ra_server_state()) -> option(ra_term()). current_term(State) -> maps:get(current_term, State). -spec machine_version(ra_server_state()) -> non_neg_integer(). machine_version(#{cfg := #cfg{machine_version = MacVer}}) -> MacVer. -spec machine(ra_server_state()) -> ra_machine:machine(). machine(#{cfg := #cfg{machine = Machine}}) -> Machine. -spec machine_query(ra:query_fun(), ra_server_state()) -> {ra_idxterm(), term()}. machine_query(QueryFun, #{cfg := #cfg{effective_machine_module = MacMod, effective_machine_version = MacVer}, machine_state := MacState, last_applied := Last, current_term := Term}) -> Ctx = #{index => Last, term => Term, machine_version => MacVer}, Res = ra_machine:query(MacMod, QueryFun, MacState, Ctx), {{Last, Term}, Res}. % Internal become(leader, OldRaftState, #{cluster := Cluster, cluster_change_permitted := CCP0, log := Log0} = State) -> Log = ra_log:release_resources(maps:size(Cluster), sequential, Log0), CCP = case OldRaftState of await_condition -> CCP0; _ -> false end, State#{log => Log, cluster_change_permitted => CCP}; become(follower, _, #{cluster := Cluster, log := Log0} = State) -> %% followers should only ever need a single segment open at any one %% time State#{log => ra_log:release_resources(1, random, Log0), %% reset all peers status to normal else it may interfere %% with release cursor conditions cluster => maps:map(fun (_, P) -> P#{status => normal} end, Cluster)}; become(_RaftState, _, State) -> State. follower_catchup_cond_fun(OriginalReason) -> fun (Entry, State) -> follower_catchup_cond(OriginalReason, Entry, State) end. follower_catchup_cond(OriginalReason, #append_entries_rpc{term = Term, prev_log_index = PLIdx, prev_log_term = PLTerm}, State0 = #{current_term := CurTerm, log := Log0}) when Term >= CurTerm -> case has_log_entry_or_snapshot(PLIdx, PLTerm, Log0) of {entry_ok, Log} -> {true, State0#{log => Log}}; {term_mismatch, _, Log} -> %% if the original reason to enter catch-up was a missing entry %% the next entry _could_ result in a term_mismatch if so we %% exit await_condition temporarily to process the AppendEntriesRpc %% that resulted in the term_mismatch {OriginalReason == missing, State0#{log => Log}}; {missing, Log} -> {false, State0#{log => Log}} end; follower_catchup_cond(_, #install_snapshot_rpc{term = Term, meta = #{index := PLIdx}}, #{current_term := CurTerm, log := Log} = State) when Term >= CurTerm -> % term is ok - check if the snapshot index is greater than the last % index seen {PLIdx >= ra_log:next_index(Log), State}; follower_catchup_cond(_, _Msg, State) -> {false, State}. wal_down_condition(_Msg, #{log := Log} = State) -> {ra_log:can_write(Log), State}. transfer_leadership_condition(#append_entries_rpc{term = Term}, State = #{current_term := CurTerm}) when Term > CurTerm -> {true, State}; transfer_leadership_condition(#install_snapshot_rpc{term = Term}, State = #{current_term := CurTerm}) when Term > CurTerm -> {true, State}; transfer_leadership_condition(_Msg, State) -> {false, State}. evaluate_commit_index_follower(#{commit_index := CommitIndex, cfg := #cfg{id = Id}, leader_id := LeaderId, last_applied := LastApplied0, current_term := Term, log := Log0} = State0, Effects0) when LeaderId =/= undefined -> %% take the minimum of the last index seen and the commit index %% This may mean we apply entries that have not yet been fsynced locally. %% This is ok as the append_entries_rpc with the updated commit index would %% ensure no uncommitted entries from a previous term have been truncated %% from the log {Idx, _} = ra_log:last_index_term(Log0), ApplyTo = min(Idx, CommitIndex), % need to catch a termination throw case catch apply_to(ApplyTo, State0, Effects0) of {delete_and_terminate, State1, Effects} -> Reply = append_entries_reply(Term, true, State1), {delete_and_terminate, State1, [cast_reply(Id, LeaderId, Reply) | Effects]}; {#{last_applied := LastApplied} = State, Effects} -> {State1, Effects1} = maybe_emit_pending_release_cursor(State, Effects), case LastApplied > LastApplied0 of true -> %% entries were applied, append eval_aux effect {follower, State1, [{aux, eval} | Effects1]}; false -> %% no entries were applied {follower, State1, Effects1} end end; evaluate_commit_index_follower(State, Effects) -> %% when no leader is known {follower, State, Effects}. make_pipelined_rpc_effects(State, Effects) -> make_pipelined_rpc_effects(State, Effects, false). make_pipelined_rpc_effects(#{cfg := #cfg{id = Id, max_append_entries_rpc_batch_size = MaxBatchSize, max_pipeline_count = MaxPipelineCount}, commit_index := CommitIndex, log := Log, cluster := Cluster} = State0, Effects0, Force) -> NextLogIdx = ra_log:next_index(Log), %% TODO: refactor this please, why does make_rpc_effect need to take the %% full state maps:fold( fun (PeerId, #{next_index := NextIdx, status := normal, commit_index_sent := CI, match_index := MatchIdx} = Peer0, {S0, More0, Effs} = Acc) when PeerId =/= Id andalso (NextIdx < NextLogIdx orelse CI < CommitIndex) -> % the status is normal and % there are unsent items or a new commit index % check if the match index isn't too far behind the % next index NumInFlight = NextIdx - MatchIdx - 1, case NumInFlight < MaxPipelineCount orelse Force of true -> %% use the last list of entries as a cache %% for the next to potentially avoid additional reads %% from the log EntryCache = case Effs of [{send_rpc, _, #append_entries_rpc{entries = Es}} | _] -> Es; _ -> [] end, %% ensure we don't pass a batch size that would allow %% the peer to go over the max pipeline count %% we'd only really get here if Force=true so setting %% a single entry batch size should be fine BatchSize = max(1, min(MaxBatchSize, MaxPipelineCount - NumInFlight)), {NewNextIdx, Eff, S} = make_rpc_effect(PeerId, Peer0, BatchSize, S0, EntryCache), ?assert(NewNextIdx >= NextIdx), Peer = Peer0#{next_index => NewNextIdx, commit_index_sent => CommitIndex}, NewNumInFlight = NewNextIdx - MatchIdx - 1, %% is there more potentially pipelining More = More0 orelse (NewNextIdx < NextLogIdx andalso NewNumInFlight < MaxPipelineCount), {put_peer(PeerId, Peer, S), More, [Eff | Effs]}; false -> Acc end; (_, _, Acc) -> Acc end, {State0, false, Effects0}, Cluster). make_rpcs(State) -> {State1, EffectsHR} = update_heartbeat_rpc_effects(State), {State2, EffectsAER} = make_rpcs_for(stale_peers(State1), State1), {State2, EffectsAER ++ EffectsHR}. % makes empty append entries for peers that aren't pipelineable make_all_rpcs(State0) -> {State1, EffectsHR} = update_heartbeat_rpc_effects(State0), {CancelEffects, Peers} = maps:fold( fun(PeerId, #{status := {snapshot_backoff, _}} = Peer, {Cs, Ps}) -> {[{cancel_snapshot_retry_timer, PeerId} | Cs], Ps#{PeerId => Peer}}; (PeerId, #{status := normal} = Peer, {Cs, Ps}) -> {Cs, Ps#{PeerId => Peer}}; (_, _, Acc) -> Acc end, {[], #{}}, peers(State1)), {State2, EffectsAER} = make_rpcs_for(Peers, State1), {State2, CancelEffects ++ EffectsAER ++ EffectsHR}. make_rpcs_for(Peers, State) -> maps:fold(fun(PeerId, Peer, {S0, Effs}) -> {_, Eff, S} = %% set a very small batch size here as these are only %% used to establish leadership / periodic heartbeats etc %% normal replication would use make_pipeline_rpc make_rpc_effect(PeerId, Peer, 1, S0), {S, [Eff | Effs]} end, {State, []}, Peers). make_rpc_effect(PeerId, Peer, MaxBatchSize, State) -> make_rpc_effect(PeerId, Peer, MaxBatchSize, State, []). make_rpc_effect(PeerId, #{next_index := Next}, MaxBatchSize, #{cfg := #cfg{id = Id}, log := Log0, current_term := Term} = State, EntryCache) -> PrevIdx = Next - 1, case ra_log:fetch_term(PrevIdx, Log0) of {PrevTerm, Log} when is_integer(PrevTerm) -> make_append_entries_rpc(PeerId, PrevIdx, PrevTerm, MaxBatchSize, State#{log => Log}, EntryCache); {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_append_entries_rpc(PeerId, PrevIdx, PrevTerm, MaxBatchSize, State#{log => Log}, EntryCache); {SnapIdx, _} -> ?DEBUG("~ts: sending snapshot to ~tw as their next index ~b " "is lower than snapshot index ~b", [log_id(State), PeerId, Next, SnapIdx]), ?assert(PrevIdx < SnapIdx), SnapState = ra_log:snapshot_state(Log), %% don't increment the next index here as we will do %% that once the snapshot is fully replicated %% and we don't pipeline entries until after snapshot {SnapIdx, {send_snapshot, PeerId, {SnapState, Id, Term}}, State#{log => Log}} end end. make_append_entries_rpc(PeerId, PrevIdx, PrevTerm, Num, #{log := Log0, current_term := Term, cfg := #cfg{id = Id}, commit_index := CommitIndex} = State, EntryCache) -> {LastIndex, _} = ra_log:last_index_term(Log0), From = PrevIdx + 1, To = min(LastIndex, PrevIdx + Num), {Entries, Log} = log_read(From, To, EntryCache, Log0), {To + 1, {send_rpc, PeerId, #append_entries_rpc{entries = lists:reverse(Entries), term = Term, leader_id = Id, prev_log_index = PrevIdx, prev_log_term = PrevTerm, leader_commit = CommitIndex}}, State#{log => Log}}. log_read(From, To, [], Log0) -> ra_log:fold(From, To, fun (E, A) -> [E | A] end, [], Log0); log_read(From0, To, Cache, Log0) -> {From, Entries0} = log_fold_cache(From0, To, Cache, []), ra_log:fold(From, To, fun (E, A) -> [E | A] end, Entries0, Log0). %% this cache is a bit so and so as it will only really work when each follower %% begins with the same from index log_fold_cache(From, To, [{From, _, _} = Entry | Rem], Acc) when From =< To -> log_fold_cache(From + 1, To, Rem, [Entry | Acc]); log_fold_cache(From, _To, _Cache, Acc) -> {From, Acc}. % stores the cluster config at an index such that we can later snapshot % at this index. -spec update_release_cursor(ra_index(), term(), map(), ra_server_state()) -> {ra_server_state(), effects()}. update_release_cursor(Index, MacState, #{condition := Conds}, State) when is_list(Conds) -> case check_release_cursor_conditions(Conds, State) of true -> do_update_release_cursor(Index, MacState, State); false -> %% stash until conditions are met {State#{pending_release_cursor => {Index, MacState, Conds}}, []} end; update_release_cursor(Index, MacState, _Opts, State) -> %% no conditions specified, process immediately do_update_release_cursor(Index, MacState, State). check_release_cursor_conditions(Conds, State) -> lists:all(fun(Cond) -> check_release_cursor_condition(Cond, State) end, Conds). check_release_cursor_condition({written, WrittenIdx}, #{log := Log}) -> {LastWrittenIdx, _} = ra_log:last_written(Log), WrittenIdx =< LastWrittenIdx; check_release_cursor_condition(no_snapshot_sends, #{cluster := Cluster}) -> not maps:fold(fun(_PeerId, #{status := {sending_snapshot, _, _}}, _Acc) -> true; (_PeerId, _Peer, Acc) -> Acc end, false, Cluster). do_update_release_cursor(Index, MacState, #{cfg := #cfg{machine = Machine}, log := Log0, cluster := Cluster} = State) -> MacVersion = index_machine_version(Index, State), MacMod = ra_machine:which_module(Machine, MacVersion), % simply pass on release cursor index to log {Log, Effects} = ra_log:update_release_cursor(Index, Cluster, {MacVersion, MacMod}, MacState, Log0), {State#{log => Log}, Effects}. maybe_emit_pending_release_cursor(#{pending_release_cursor := {Index, MacState, Conds}} = State, Effects) -> case check_release_cursor_conditions(Conds, State) of true -> %% can now process the stashed release cursor case do_update_release_cursor(Index, MacState, maps:remove(pending_release_cursor, State)) of {State1, []} -> {State1, Effects}; {State1, [Eff]} -> {State1, [Eff | Effects ]}; {State1, Effects1} -> {State1, Effects1 ++ Effects} end; false -> {State, Effects} end; maybe_emit_pending_release_cursor(State, Effects) -> {State, Effects}. -spec checkpoint(ra_index(), term(), ra_server_state()) -> {ra_server_state(), effects()}. checkpoint(Index, MacState, #{cfg := #cfg{machine = Machine}, log := Log0, cluster := Cluster} = State) -> MacVersion = index_machine_version(Index, State), MacMod = ra_machine:which_module(Machine, MacVersion), {Log, Effects} = ra_log:checkpoint(Index, Cluster, {MacVersion, MacMod}, MacState, Log0), {State#{log => Log}, Effects}. -spec promote_checkpoint(ra_index(), ra_server_state()) -> {ra_server_state(), effects()}. promote_checkpoint(Index, #{log := Log0} = State) -> {Log, Effects} = ra_log:promote_checkpoint(Index, Log0), {State#{log => Log}, Effects}. % 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 := PLA, last_applied := LA} = State) when LA =< PLA -> % if last applied is less than or equal to PL do nothing State; persist_last_applied(#{last_applied := LastApplied0, log := Log, cfg := #cfg{uid = UId} = Cfg} = State) -> {LastWrittenIdx, _} = ra_log:last_written(Log), LastApplied = min(LastApplied0, LastWrittenIdx), PersistedLastApplied = maps:get(persisted_last_applied, State, 0), %% only persist the last applied if the last written has also caught up, %% else it is possible that on recovery the last applied index refers to %% indexes that never were durably written if LastApplied > PersistedLastApplied -> ok = ra_log_meta:store(meta_name(Cfg), UId, last_applied, LastApplied), State#{persisted_last_applied => LastApplied}; true -> State end. -spec update_peer(ra_server_id(), #{next_index => non_neg_integer(), query_index => non_neg_integer(), commit_index_sent => non_neg_integer(), status => ra_peer_status()}, ra_server_state()) -> ra_server_state(). update_peer(PeerId, Update, #{cluster := Peers} = State) when is_map(Update) -> Peer = maps:merge(maps:get(PeerId, Peers), Update), put_peer(PeerId, Peer, State). -spec update_disconnected_peers(node(), nodeup | nodedown, ra_server_state()) -> ra_server_state(). update_disconnected_peers(Node, nodeup, #{cluster := Peers} = State) -> State#{cluster => maps:map( fun ({_, PeerNode}, #{status := disconnected} = Peer) when PeerNode == Node -> Peer#{status => normal}; ({_, PeerNode}, #{status := {snapshot_backoff, _}} = Peer) when PeerNode == Node -> %% Node came back up - reset backoff to allow immediate retry Peer#{status => normal}; (_, Peer) -> Peer end, Peers)}; update_disconnected_peers(_Node, _Status, State) -> State. %% Helper to find peer by snapshot sender pid find_peer_with_snapshot_pid(SnapshotPid, Peers) -> case maps:to_list( maps:filter(fun(_, #{status := {sending_snapshot, Pid, _}}) when Pid =:= SnapshotPid -> true; (_, _) -> false end, Peers)) of [{PeerId, Peer}] -> {PeerId, Peer}; _ -> undefined end. peer_snapshot_process_exited(SnapshotPid, #{cluster := Peers} = State) -> case find_peer_with_snapshot_pid(SnapshotPid, Peers) of {PeerId, Peer} -> put_peer(PeerId, Peer#{status => normal}, State); undefined -> State end. %% Set backoff status preserving attempt count from sending_snapshot peer_snapshot_process_exited_with_backoff(SnapshotPid, #{cluster := Peers} = State) -> case find_peer_with_snapshot_pid(SnapshotPid, Peers) of {PeerId, #{status := {sending_snapshot, _, AttemptCount}} = Peer} -> %% Increment count and set backoff status NewCount = AttemptCount + 1, NewState = put_peer(PeerId, Peer#{status => {snapshot_backoff, NewCount}}, State), {NewState, PeerId, NewCount}; _ -> {State, undefined, 0} end. -spec handle_down(ra_state(), machine | snapshot_sender | snapshot_writer | aux, pid(), term(), ra_server_state()) -> {ra_state(), ra_server_state(), effects()}. handle_down(leader, machine, Pid, Info, State) when is_pid(Pid) -> % %% commit command to be processed by state machine Eff = {next_event, {command, low, {'$usr', {down, Pid, Info}, noreply}}}, {leader, State, [Eff]}; handle_down(RaftState, snapshot_sender, Pid, Info, #{cfg := #cfg{log_id = LogId}} = State0) when (RaftState == leader orelse RaftState == await_condition) andalso is_pid(Pid) -> %% if a rebalance is being done we also need to handle snapshot_sender %% downs here case Info of normal -> %% Success - handled by install_snapshot_result, just clean up State = peer_snapshot_process_exited(Pid, State0), %% Check if pending release cursor can now proceed %% (no_snapshot_sends condition may now be satisfied) {State1, Effects} = maybe_emit_pending_release_cursor(State, []), {RaftState, State1, Effects}; _Error -> ?DEBUG("~ts: Snapshot sender process ~w exited with ~W", [LogId, Pid, Info, 10]), {State, PeerId, AttemptCount} = peer_snapshot_process_exited_with_backoff(Pid, State0), case PeerId of undefined -> {RaftState, State, []}; _ -> %% Exponential backoff: min(5000 * 2^(count-1), 60000) ms Delay = min(5000 * (1 bsl (AttemptCount - 1)), 60000), {RaftState, State, [{start_snapshot_retry_timer, PeerId, Delay}]} end end; handle_down(RaftState, log, Pid, Info, #{log := Log0} = State) -> {Log, Effects} = ra_log:handle_event({down, Pid, Info}, Log0), {RaftState, State#{log => Log}, Effects}; handle_down(RaftState, aux, Pid, Info, State) when is_pid(Pid) -> handle_aux(RaftState, cast, {down, Pid, Info}, State); handle_down(RaftState, Type, Pid, Info, #{cfg := #cfg{log_id = LogId}} = State) -> ?DEBUG("~ts: handle_down: unexpected ~w ~w exited with ~W", [LogId, Type, Pid, Info, 10]), {RaftState, State, []}. -spec handle_node_status(ra_state(), machine | aux, node(), nodeup | nodedown, term(), ra_server_state()) -> {ra_state(), ra_server_state(), effects()}. handle_node_status(leader, machine, Node, Status, _Infos, State) when is_atom(Node) -> %% commit command to be processed by state machine %% TODO: provide an option where the machine or aux can be provided with %% the node down reason Meta = #{ts => erlang:system_time(millisecond)}, handle_leader({command, {'$usr', Meta, {Status, Node}, noreply}}, State); handle_node_status(RaftState, aux, Node, Status, _Infos, State) when is_atom(Node) -> handle_aux(RaftState, cast, {Status, Node}, State); handle_node_status(RaftState, Type, Node, Status, _Info, #{cfg := #cfg{log_id = LogId}} = State) -> ?DEBUG("~ts: handle_node_status: unexpected ~w ~w status change ~w", [LogId, Type, Node, Status]), {RaftState, State, []}. -spec terminate(ra_server_state(), Reason :: {shutdown, delete} | term()) -> ok. terminate(#{log := Log, cfg := #cfg{log_id = LogId}} = _State, {shutdown, delete}) -> ?NOTICE("~ts: terminating with reason 'delete'", [LogId]), catch ra_log:delete_everything(Log), ok; terminate(#{cfg := #cfg{log_id = LogId}} = State, Reason) -> ?DEBUG("~ts: terminating with reason '~w'", [LogId, Reason]), State1 = maybe_write_recovery_checkpoint(State), #{log := Log} = persist_last_applied(State1), catch ra_log:close(Log), ok. %% @doc Maybe write a recovery checkpoint during shutdown. %% Only writes if min_recovery_checkpoint_interval is configured (> 0) and %% enough entries have been applied since the last snapshot/checkpoint. maybe_write_recovery_checkpoint(#{cfg := #cfg{log_id = LogId, machine = Machine}, log := Log0, cluster := Cluster, last_applied := LastApplied, machine_state := MacState} = State) -> SnapState = ra_log:snapshot_state(Log0), MinInterval = get_min_recovery_checkpoint_interval(State), case MinInterval of 0 -> %% Feature disabled State; _ -> HighestIdx = ra_snapshot:highest_idx(SnapState), case LastApplied - HighestIdx >= MinInterval of true -> %% Fetch the term for the LastApplied index {Term, Log} = ra_log:fetch_term(LastApplied, Log0), case Term of undefined -> ?WARN("~ts: cannot write recovery checkpoint, " "term not found for index ~b", [LogId, LastApplied]), State#{log => Log}; _ -> %% Write recovery checkpoint SnapModule = ra_machine:snapshot_module(Machine), %% Include cluster membership for proper recovery ClusterServerIds = maps:map(fun (_, V) -> maps:with([voter_status], V) end, Cluster), Meta = #{index => LastApplied, term => Term, cluster => ClusterServerIds, machine_version => ra_machine:version(Machine)}, case ra_snapshot:write_recovery_checkpoint( Meta, MacState, SnapModule, SnapState) of {ok, SnapState1} -> ?DEBUG("~ts: wrote recovery checkpoint at index ~b", [LogId, LastApplied]), Log1 = ra_log:set_snapshot_state(SnapState1, Log), State#{log => Log1}; {error, Err} -> ?WARN("~ts: failed to write recovery checkpoint: ~p", [LogId, Err]), State#{log => Log} end end; false -> %% Not enough entries since last snapshot/checkpoint State end end. get_min_recovery_checkpoint_interval(#{cfg := #cfg{min_recovery_checkpoint_interval = Interval}}) -> Interval. %% @doc Check if we can skip some log replay using a recovery checkpoint. %% If a recovery checkpoint exists with an index higher than LastApplied, %% recover its machine state and update LastApplied to skip log replay. maybe_recover_from_recovery_checkpoint(LastApplied, CommitIndex, SnapState, #{cfg := #cfg{log_id = LogId, machine = Machine, machine_versions = MachineVersions, effective_machine_version = CurEffMacVer } = Cfg0} = State) -> case ra_snapshot:recovery_checkpoint(SnapState) of {RCIdx, _} when RCIdx > LastApplied andalso RCIdx =< CommitIndex -> %% Recovery checkpoint exists with higher index - try to use it case ra_snapshot:recover_recovery_checkpoint(SnapState) of {ok, #{index := RCIdx, term := RCTerm, cluster := ClusterNodes, machine_version := RCMacVer}, RCMacState} -> ?INFO("~ts: using recovery checkpoint at index ~b, machine version ~b " "to skip log replay from ~b", [LogId, RCIdx, RCMacVer, LastApplied]), %% Update effective machine version if recovery checkpoint %% has a higher version (we're skipping log entries that %% may have included machine version upgrades) Cfg = case RCMacVer > CurEffMacVer of true -> EffMacMod = ra_machine:which_module(Machine, RCMacVer), put_counter(Cfg0, ?C_RA_SVR_METRIC_EFFECTIVE_MACHINE_VERSION, RCMacVer), Cfg0#cfg{effective_machine_version = RCMacVer, machine_versions = [{RCIdx, RCMacVer} | MachineVersions], effective_machine_module = EffMacMod, effective_handle_aux_fun = ra_machine:which_aux_fun(EffMacMod)}; false -> Cfg0 end, %% Update state with recovery checkpoint's machine state State1 = State#{cfg => Cfg, machine_state => RCMacState, last_applied => RCIdx}, %% Update cluster if recovery checkpoint has it State2 = case ClusterNodes of Nodes when map_size(Nodes) > 0 -> #{cfg := #cfg{id = Id}} = State1, Cluster = make_cluster(Id, Nodes), State1#{cluster => Cluster, cluster_index_term => {RCIdx, RCTerm}}; _ -> State1 end, {RCIdx, State2}; {error, Reason} -> ?WARN("~ts: failed to recover from recovery checkpoint: ~p, " "falling back to log replay", [LogId, Reason]), {LastApplied, State} end; _ -> %% No recovery checkpoint or not useful {LastApplied, State} end. -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, try fold_log_from(Idx, Fun, {State, Log}) of {ok, {State1, Log1}} -> {ok, State1, RaState#{log => Log1}} catch _:Err -> {error, Err, RaState} end. %% reads user commands at the specified index -spec log_read([ra_index()], ra_server_state()) -> {ok, [term()], ra_server_state()} | {error, ra_server_state()}. log_read(Indexes, #{log := Log0} = State) -> {Entries, Log} = ra_log:sparse_read(Indexes, Log0), {ok, [Data || {_Idx, _Term, {'$usr', _, Data, _}} <- Entries], State#{log => Log}}. %% reads user commands at the specified index -spec log_partial_read([ra_index()], ra_server_state()) -> ra_log:read_plan(). log_partial_read(Indexes, #{log := Log0}) -> ra_log:partial_read(Indexes, Log0, fun transform_for_partial_read/3). %%%=================================================================== %%% Internal functions %%%=================================================================== call_for_election(TargetState, State) -> call_for_election(TargetState, State, []). call_for_election(candidate, #{cfg := #cfg{id = Id, log_id = LogId} = Cfg, current_term := CurrentTerm} = State0, Effects) -> ok = incr_counter(Cfg, ?C_RA_SRV_ELECTIONS, 1), NewTerm = CurrentTerm + 1, ?DEBUG("~ts: election called for in term ~b", [LogId, 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_term_and_voted_for(NewTerm, Id, State0), {candidate, State#{leader_id => undefined, votes => 0}, [{next_event, cast, VoteForSelf}, {send_vote_requests, Reqs} | Effects]}; call_for_election(pre_vote, #{cfg := #cfg{id = Id, log_id = LogId, machine_version = MacVer} = Cfg, current_term := Term} = State0, Effects) -> ok = incr_counter(Cfg, ?C_RA_SRV_PRE_VOTE_ELECTIONS, 1), ?DEBUG("~ts: pre_vote election called for in term ~b", [LogId, Term]), Token = make_ref(), PeerIds = peer_ids(State0), {LastIdx, LastTerm} = last_idx_term(State0), Reqs = [{PeerId, #pre_vote_rpc{term = Term, token = Token, machine_version = MacVer, 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_term_and_voted_for(Term, Id, State0), {pre_vote, State#{leader_id => undefined, votes => 0, pre_vote_token => Token}, [{next_event, cast, VoteForSelf}, {send_vote_requests, Reqs} | Effects]}. process_pre_vote(FsmState, #pre_vote_rpc{term = Term, candidate_id = Cand, version = Version, machine_version = TheirMacVer, token = Token, last_log_index = LLIdx, last_log_term = LLTerm}, #{cfg := #cfg{machine_version = OurMacVer, effective_machine_version = EffMacVer}, current_term := CurTerm} = State0) when Term >= CurTerm -> %% Pre-vote must not set voted_for (per Raft thesis Section 9.6), %% but should still update current_term to retain the highest known term. 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-> ?DEBUG("~ts: declining pre-vote for ~tw for protocol version ~b", [log_id(State0), Cand, Version]), {FsmState, State, [{reply, pre_vote_result(Term, Token, false)}]}; true when TheirMacVer == EffMacVer orelse (TheirMacVer >= EffMacVer andalso TheirMacVer =< OurMacVer) -> ?DEBUG("~ts: granting pre-vote for ~tw" " machine version (their:ours:effective) ~b:~b:~b" " with last {index, term} ~tw" " for term ~b previous term ~b", [log_id(State0), Cand, TheirMacVer, OurMacVer, EffMacVer, {LLIdx, LLTerm}, Term, CurTerm]), {FsmState, State, [{reply, pre_vote_result(Term, Token, true)}]}; true -> ?DEBUG("~ts: declining pre-vote for ~tw their machine version ~b" " ours is ~b effective ~b", [log_id(State0), Cand, TheirMacVer, OurMacVer, EffMacVer]), {FsmState, State, [{reply, pre_vote_result(Term, Token, false)}, start_election_timeout]}; false -> ?DEBUG("~ts: declining pre-vote for ~tw for term ~b," " candidate last log {index, term} was: ~tw " "last log entry {index, term} seen is: ~tw", [log_id(State0), Cand, Term, {LLIdx, LLTerm}, LastIdxTerm]), case FsmState of follower -> {FsmState, State, [start_election_timeout]}; _ -> {FsmState, State, [{reply, pre_vote_result(Term, Token, false)}]} end end; process_pre_vote(FsmState, #pre_vote_rpc{term = Term, token = Token, candidate_id = Candidate}, #{current_term := CurTerm} = State) when Term < CurTerm -> ?DEBUG("~ts declining pre-vote to ~tw for term ~b, current term ~b", [log_id(State), Candidate, 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, query_index => 0, status => normal}. new_peer_with(Map) -> maps:merge(new_peer(), Map). peers(#{cfg := #cfg{id = Id}, cluster := Peers}) -> maps:remove(Id, Peers). peer_status(PeerId, #{cluster := Peers}) -> case Peers of #{PeerId := #{status := Status}} -> Status; _ -> undefined end. % peers that could need an update stale_peers(#{commit_index := CommitIndex, cfg := #cfg{id = ThisId}, cluster := Cluster}) -> maps:filter(fun (Id , _) when Id == ThisId -> false; (_, #{status := normal, next_index := NI, match_index := MI}) when MI < NI - 1 -> % there are unconfirmed items true; (_, #{status := normal, commit_index_sent := CI}) when CI < CommitIndex -> % the commit index has been updated true; (_, _Peer) -> false end, Cluster). peer_ids(State) -> maps:keys(peers(State)). peer(PeerId, #{cluster := Nodes}) -> maps:get(PeerId, Nodes, undefined). put_peer(PeerId, Peer, #{cluster := Peers} = State) -> State#{cluster => Peers#{PeerId => Peer}}. update_term_and_voted_for(Term, VotedFor, #{cfg := #cfg{uid = UId} = Cfg, current_term := CurTerm} = State) -> CurVotedFor = maps:get(voted_for, State, undefined), case Term =:= CurTerm andalso VotedFor =:= CurVotedFor of true -> %% no update needed State; false -> MetaName = meta_name(Cfg), %% as this is a rare event it is ok to go sync here ok = ra_log_meta:store(MetaName, UId, current_term, Term), ok = ra_log_meta:store_sync(MetaName, UId, voted_for, VotedFor), incr_counter(Cfg, ?C_RA_SRV_TERM_AND_VOTED_FOR_UPDATES, 1), put_counter(Cfg, ?C_RA_SVR_METRIC_TERM, Term), %% this is probably not necessary reset_query_index(State#{current_term => Term, voted_for => VotedFor}) end. update_term(Term, State = #{current_term := CurTerm}) when Term =/= undefined andalso Term > CurTerm -> update_term_and_voted_for(Term, undefined, State); update_term(_, State) -> State. last_idx_term(#{log := Log}) -> ra_log:last_index_term(Log). state_query(all, State) -> State; state_query(overview, State) -> overview(State); state_query(machine, #{machine_state := MacState}) -> MacState; state_query(voters, #{cluster := Cluster}) -> maps:fold(fun(K, V, Acc) -> case maps:get(voter_status, V, undefined) of undefined -> [K|Acc]; S -> case maps:get(membership, S, undefined) of undefined -> [K|Acc]; voter -> [K|Acc]; _ -> Acc end end end, [], Cluster); state_query(leader, State) -> maps:get(leader_id, State, undefined); state_query(members, #{cluster := Cluster}) -> maps:keys(Cluster); state_query(members_info, #{cfg := #cfg{id = Self}, cluster := Cluster, leader_id := Self, query_index := QI, commit_index := CI, membership := Membership}) -> maps:map(fun(Id, Peer) -> case {Id, Peer} of {Self, Peer = #{voter_status := VoterStatus}} -> %% For completeness sake, preserve `target` %% of once promoted leader. #{next_index => CI+1, match_index => CI, query_index => QI, status => normal, voter_status => VoterStatus#{membership => Membership}}; {Self, _} -> #{next_index => CI+1, match_index => CI, query_index => QI, status => normal, voter_status => #{membership => Membership}}; {_, Peer = #{voter_status := _}} -> Peer; {_, Peer} -> %% Initial cluster members have no voter_status. Peer#{voter_status => #{membership => voter}} end end, Cluster); state_query(members_info, #{cfg := #cfg{id = Self}, cluster := Cluster, query_index := QI, commit_index := CI, membership := Membership}) -> %% Followers do not have sufficient information, %% bail out and send whatever we have. maps:map(fun(Id, Peer) -> case {Id, Peer} of {Self, #{voter_status := VS}} -> #{match_index => CI, query_index => QI, voter_status => VS#{membership => Membership}}; {Self, _} -> #{match_index => CI, query_index => QI, voter_status => #{membership => Membership}}; _ -> #{} end end, Cluster); state_query(initial_members, #{log := Log}) -> case ra_log:read_config(Log) of {ok, #{initial_members := InitialMembers}} -> InitialMembers; _ -> error end; state_query(last_applied, State) -> maps:get(last_applied, State, undefined); state_query(Query, _State) -> {error, {unknown_query, Query}}. %% § 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, Log0) -> case ra_log:fetch_term(Idx, Log0) of {undefined, Log} -> case ra_log:snapshot_index_term(Log) of {Idx, Term} -> {entry_ok, Log}; {Idx, OtherTerm} -> {term_mismatch, OtherTerm, Log}; _ -> {missing, Log} end; {Term, Log} -> {entry_ok, Log}; {OtherTerm, Log} -> {term_mismatch, OtherTerm, Log} end. fetch_term(Idx, #{log := Log0} = State) -> case ra_log:fetch_term(Idx, Log0) of {undefined, Log} -> case ra_log:snapshot_index_term(Log) of {Idx, Term} -> {Term, State#{log => Log}}; _ -> {undefined, State#{log => Log}} end; {Term, Log} -> {Term, State#{log => Log}} end. %% @doc Strips the Ra internal command wrapper away for user commands %% to return the original machine command written %% @end transform_for_partial_read(_Idx, _Term, {'$usr', _, Cmd, _}) -> Cmd; transform_for_partial_read(_Idx, _Term, Cmd) -> %% Other commands leave as is. %% It would be quite unusual for these to be read externally %% but you never know. Cmd. -spec make_cluster(ra_server_id(), ra_cluster_snapshot() | [ra_server_id()]) -> ra_cluster(). make_cluster(Self, Nodes0) when is_list(Nodes0) -> Nodes = lists:foldl(fun(N, Acc) -> Acc#{N => new_peer()} end, #{}, Nodes0), append_self(Self, Nodes); make_cluster(Self, Nodes0) when is_map(Nodes0) -> Nodes = maps:map(fun(_, Peer0) -> new_peer_with(Peer0) end, Nodes0), append_self(Self, Nodes). append_self(Self, Nodes) -> case 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}) -> NextIdx = ra_log:next_index(Log), Cluster = maps:map(fun (_, Peer0) -> Peer1 = maps:with([voter_status, machine_version], Peer0), Peer2 = Peer1#{next_index => NextIdx}, new_peer_with(Peer2) end, Cluster0), State#{cluster => Cluster}. apply_to(ApplyTo, State, Effs) -> apply_to(ApplyTo, fun apply_with/2, #{}, Effs, State). apply_to(ApplyTo, ApplyFun, State, Effs) -> apply_to(ApplyTo, ApplyFun, #{}, Effs, State). apply_to(ApplyTo, ApplyFun, Notifys0, Effects0, #{last_applied := LastApplied, cfg := #cfg{machine_version = MacVer, effective_machine_module = MacMod, effective_machine_version = EffMacVer} = Cfg, machine_state := MacState0, log := Log0} = State0) when ApplyTo > LastApplied andalso MacVer >= EffMacVer -> From = LastApplied + 1, {LastIdx, _LastTerm} = ra_log:last_index_term(Log0), To = min(LastIdx, ApplyTo), FoldState = {MacMod, LastApplied, State0, MacState0, Effects0, Notifys0, undefined}, {{_, AppliedTo, State, MacState, Effects, Notifys, LastTs}, Log} = ra_log:fold(From, To, ApplyFun, FoldState, Log0), CommitLatency = case LastTs of undefined -> 0; _ when is_integer(LastTs) -> erlang:system_time(millisecond) - LastTs end, %% due to machine versioning all entries may not have been applied FinalEffs = make_notify_effects(Notifys, lists:reverse(Effects)), put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_APPLIED, AppliedTo), put_counter(Cfg, ?C_RA_SVR_METRIC_COMMIT_LATENCY, CommitLatency), {State#{last_applied => AppliedTo, log => Log, commit_latency => CommitLatency, machine_state => MacState}, FinalEffs}; apply_to(_ApplyTo, _, Notifys, Effects, State) when is_list(Effects) -> FinalEffs = make_notify_effects(Notifys, lists:reverse(Effects)), {State, FinalEffs}. make_notify_effects(Nots, Prior) when map_size(Nots) > 0 -> [{notify, Nots} | Prior]; make_notify_effects(_Nots, Prior) -> Prior. append_machine_effects([], Effs) -> Effs; append_machine_effects([AppEff], Effs) -> [AppEff | Effs]; append_machine_effects(AppEffs, Effs) -> [AppEffs | Effs]. cluster_scan_fun({Idx, Term, {'$ra_cluster_change', _Meta, NewCluster, _}}, {_, State0}) -> ?DEBUG("~ts: ~ts: applying ra cluster change to ~tw", [log_id(State0), ?FUNCTION_NAME, maps:keys(NewCluster)]), %% we are recovering and should apply the cluster change {Idx, State0#{cluster => NewCluster, membership => get_membership(NewCluster, State0), cluster_change_permitted => true, cluster_index_term => {Idx, Term}}}; cluster_scan_fun({Idx, _, _}, {_, State}) -> {Idx, State}. apply_with(_Cmd, {Mod, LastAppliedIdx, #{cfg := #cfg{machine_version = MacVer, effective_machine_version = Effective}} = State, MacSt, Effects, Notifys, LastTs}) when MacVer < Effective -> %% we cannot apply any further entries {Mod, LastAppliedIdx, State, MacSt, Effects, Notifys, LastTs}; apply_with({Idx, Term, {'$usr', CmdMeta, Cmd, ReplyMode}}, {Module, _LastAppliedIdx, State = #{cfg := #cfg{effective_machine_version = MacVer}}, MacSt, Effects0, Notifys0, LastTs}) -> %% augment the meta data structure Meta = augment_command_meta(Idx, Term, MacVer, ReplyMode, CmdMeta), Ts = maps:get(ts, CmdMeta, LastTs), case ra_machine:apply(Module, Meta, Cmd, MacSt) of {NextMacSt, Reply, MacEffs} -> {Effects, Notifys} = add_reply(CmdMeta, Reply, ReplyMode, append_machine_effects(MacEffs, Effects0), Notifys0), {Module, Idx, State, NextMacSt, Effects, Notifys, Ts}; {NextMacSt, Reply} -> {Effects, Notifys} = add_reply(CmdMeta, Reply, ReplyMode, Effects0, Notifys0), {Module, Idx, State, NextMacSt, Effects, Notifys, Ts} end; apply_with({Idx, Term, {'$ra_cluster_change', CmdMeta, NewCluster, ReplyMode}}, {Mod, _, State0, MacSt, Effects0, Notifys0, LastTs}) -> {Effects, Notifys} = add_reply(CmdMeta, ok, ReplyMode, Effects0, Notifys0), State = case State0 of #{cluster_index_term := {CI, CT}} when Idx > CI andalso Term >= CT -> ?DEBUG("~ts: applying ra cluster change at index ~b to ~tw", [log_id(State0), Idx, maps:keys(NewCluster)]), %% we are recovering and should apply the cluster change State0#{cluster => NewCluster, membership => get_membership(NewCluster, State0), cluster_change_permitted => true, cluster_index_term => {Idx, Term}}; _ -> ?DEBUG("~ts: committing ra cluster change at index ~b to ~tw", [log_id(State0), Idx, maps:keys(NewCluster)]), %% else just enable further cluster changes again State0#{cluster_change_permitted => true} end, {Mod, Idx, State, MacSt, Effects, Notifys, LastTs}; apply_with({Idx, Term, {noop, CmdMeta, NextMacVer}}, {CurModule, LastAppliedIdx, #{cfg := #cfg{log_id = LogId, machine_version = MacVer, %% active machine versions and their index %% (from last snapshot) machine = Machine, machine_versions = MacVersions, effective_machine_version = OldMacVer } = Cfg0, current_term := CurrentTerm, cluster_change_permitted := ClusterChangePerm0} = State0, MacSt, Effects, Notifys, LastTs}) -> ClusterChangePerm = case CurrentTerm of Term -> ?DEBUG("~ts: enabling ra cluster changes in" " ~b, index ~b", [LogId, Term, Idx]), true; _ -> ClusterChangePerm0 end, put_counter(Cfg0, ?C_RA_SVR_METRIC_EFFECTIVE_MACHINE_VERSION, NextMacVer), %% can we understand the next machine version IsOk = MacVer >= NextMacVer, case NextMacVer > OldMacVer of true when IsOk -> %% discover the next module to use Module = ra_machine:which_module(Machine, NextMacVer), %% enable cluster change if the noop command is for the current term Cfg = Cfg0#cfg{effective_machine_version = NextMacVer, %% record this machine version "term" machine_versions = [{Idx, NextMacVer} | MacVersions], effective_machine_module = Module, effective_handle_aux_fun = ra_machine:which_aux_fun(Module) }, State = State0#{cfg => Cfg, cluster_change_permitted => ClusterChangePerm}, Meta = augment_command_meta(Idx, Term, MacVer, undefined, CmdMeta), ?DEBUG("~ts: applying new machine version ~b current ~b", [LogId, NextMacVer, OldMacVer]), apply_with({Idx, Term, {'$usr', Meta, {machine_version, OldMacVer, NextMacVer}, none}}, {Module, LastAppliedIdx, State, MacSt, Effects, Notifys, LastTs}); true -> %% we cannot make progress as we don't understand the new %% machine version so we %% update the effective machine version to stop any further entries %% being applied. This is ok as a restart will be needed to %% learn the new machine version which will reset it ?DEBUG("~ts: unknown machine version ~b current ~b" " cannot apply any further entries", [LogId, NextMacVer, MacVer]), Cfg = Cfg0#cfg{effective_machine_version = NextMacVer}, State = State0#{cfg => Cfg}, {CurModule, LastAppliedIdx, State, MacSt, Effects, Notifys, LastTs}; false -> State = State0#{cluster_change_permitted => ClusterChangePerm}, {CurModule, Idx, State, MacSt, Effects, Notifys, LastTs} end; apply_with({Idx, _, {'$ra_cluster', CmdMeta, delete, ReplyType}}, {Module, _, State0, MacSt, Effects0, Notifys0, _LastTs}) -> % cluster deletion {Effects1, Notifys} = add_reply(CmdMeta, ok, ReplyType, Effects0, Notifys0), NotEffs = make_notify_effects(Notifys, []), %% virtual "eol" state EOLEffects = ra_machine:state_enter(Module, 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 ++ NotEffs ++ Effects1}); apply_with({Idx, _, _} = Cmd, Acc) -> %% Catch-all for unhandled commands - logs a warning and advances last_applied. %% This ensures forward progress even with unexpected command types. ?WARN("~ts: apply_with: unhandled command: ~W", [log_id(element(2, Acc)), Cmd, 10]), setelement(2, Acc, Idx). augment_command_meta(Idx, Term, MacVer, undefined, CmdMeta) -> augment_command_meta(Idx, Term, MacVer, CmdMeta); augment_command_meta(Idx, Term, MacVer, ReplyMode, CmdMeta) -> augment_command_meta(Idx, Term, MacVer, CmdMeta#{reply_mode => ReplyMode}). augment_command_meta(Idx, Term, MacVer, CmdMeta) -> maps:fold(fun (ts, V, Acc) -> %% rename from compact key name Acc#{system_time => V}; (K, V, Acc) -> Acc#{K => V} end, #{index => Idx, machine_version => MacVer, term => Term}, CmdMeta). add_reply(_, '$ra_no_reply', _, Effects, Notifys) -> {Effects, Notifys}; add_reply(#{from := From}, Reply, await_consensus, Effects, Notifys) -> {[{reply, From, {wrap_reply, Reply}} | Effects], Notifys}; add_reply(#{from := From}, Reply, {await_consensus, Options}, Effects, Notifys) -> Replier = case Options of #{reply_from := local} -> local; #{reply_from := {member, Member}} -> {member, Member}; _ -> leader end, ReplyEffect = {reply, From, {wrap_reply, Reply}, Replier}, {[ReplyEffect | Effects], Notifys}; add_reply(_, Reply, {notify, Corr, Pid}, Effects, Notifys) -> % 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}, case Notifys of #{Pid := T} -> {Effects, Notifys#{Pid => [CorrData | T]}}; _ -> {Effects, Notifys#{Pid => [CorrData]}} end; add_reply(_, _, _, % From, Reply, Mode Effects, Notifys) -> {Effects, Notifys}. append_log_leader({CmdTag, _, _, _}, #{cluster_change_permitted := false} = State, Effects) when CmdTag == '$ra_join' orelse CmdTag == '$ra_leave' -> {not_appended, cluster_change_not_permitted, State, Effects}; append_log_leader({'$ra_join', From, #{id := JoiningNode, voter_status := Voter0}, ReplyMode}, #{cluster := OldCluster} = State, Effects) -> case ensure_promotion_target(Voter0, State) of {error, Reason} -> {not_appended, Reason, State}; {ok, Voter} -> case OldCluster of #{JoiningNode := #{voter_status := Voter}} -> already_member(State, Effects); #{JoiningNode := Peer} -> % Update member status. Cluster = OldCluster#{JoiningNode => Peer#{voter_status => Voter}}, append_cluster_change(Cluster, From, ReplyMode, State, Effects); _ -> % Insert new member. Cluster = OldCluster#{JoiningNode => new_peer_with(#{voter_status => Voter})}, append_cluster_change(Cluster, From, ReplyMode, State, Effects) end end; append_log_leader({'$ra_join', From, #{id := JoiningNode} = Config, ReplyMode}, State, Effects) -> append_log_leader({'$ra_join', From, #{id => JoiningNode, voter_status => maps:with([membership, uid, target], Config)}, ReplyMode}, State, Effects); append_log_leader({'$ra_join', From, JoiningNode, ReplyMode}, #{cluster := OldCluster} = State, Effects) -> % Legacy $ra_join, join as voter if no such member in the cluster. case OldCluster of #{JoiningNode := _} -> already_member(State, Effects); _ -> append_log_leader({'$ra_join', From, #{id => JoiningNode}, ReplyMode}, State, Effects) end; append_log_leader({'$ra_leave', From, LeavingServer, ReplyMode}, #{cfg := #cfg{log_id = LogId}, cluster := OldCluster} = State, Effects) -> case OldCluster of #{LeavingServer := _} -> Cluster = maps:remove(LeavingServer, OldCluster), append_cluster_change(Cluster, From, ReplyMode, State, Effects); _ -> ?DEBUG("~ts: member ~tw requested to leave but was not a member. " "Members: ~tw", [LogId, LeavingServer, maps:keys(OldCluster)]), % not a member - do nothing {not_appended, not_member, State, Effects} end; append_log_leader(Cmd, #{log := Log0, current_term := Term} = State, Effects) -> NextIdx = ra_log:next_index(Log0), try ra_log:append({NextIdx, Term, Cmd}, Log0) of Log -> {ok, NextIdx, Term, State#{log => Log}, Effects} catch error:wal_down -> {not_appended, wal_down, State, Effects} end. 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, _} -> ?DEBUG("~ts: ~ts: follower applying ra cluster change to ~tw", [log_id(State), ?FUNCTION_NAME, maps:keys(Cluster)]), State#{cluster => Cluster, membership => get_membership(Cluster, State), cluster_index_term => {Idx, Term}}; _ -> % revert back to previous cluster {PrevIdx, PrevTerm, PrevCluster} = maps:get(previous_cluster, State), ?DEBUG("~ts: ~ts: follower reverting cluster change to ~tw", [log_id(State), ?FUNCTION_NAME, maps:keys(PrevCluster)]), State1 = State#{cluster => PrevCluster, membership => get_membership(PrevCluster, State), cluster_index_term => {PrevIdx, PrevTerm}}, pre_append_log_follower(Entry, State1) end; pre_append_log_follower({Idx, Term, {'$ra_cluster_change', _, Cluster, _}}, #{cluster := PrevCluster, cluster_index_term := {PrevCITIdx, PrevCITTerm}} = State) -> ?DEBUG("~ts: ~ts: follower applying ra cluster change to ~tw", [log_id(State), ?FUNCTION_NAME, maps:keys(Cluster)]), State#{cluster => Cluster, membership => get_membership(Cluster, State), cluster_index_term => {Idx, Term}, previous_cluster => {PrevCITIdx, PrevCITTerm, PrevCluster}}; pre_append_log_follower(_, State) -> State. append_cluster_change(Cluster, From, ReplyMode, #{log := Log0, cluster := PrevCluster, cluster_index_term := {PrevCITIdx, PrevCITTerm}, current_term := Term} = State, Effects) -> % 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? try ra_log:append({NextIdx, Term, Command}, Log0) of Log -> {ok, NextIdx, Term, State#{log => Log, cluster => Cluster, cluster_change_permitted => false, cluster_index_term => IdxTerm, previous_cluster => {PrevCITIdx, PrevCITTerm, PrevCluster}}, Effects} catch error:wal_down -> {not_appended, wal_down, State, Effects} end. mismatch_append_entries_reply(Term, LastAppliedIdx, State0) -> {LATerm, State} = fetch_term(LastAppliedIdx, State0), % assert LATerm is found ?assert(LATerm =/= undefined), {#append_entries_reply{term = Term, success = false, next_index = LastAppliedIdx + 1, last_index = LastAppliedIdx, last_term = LATerm}, State}. append_entries_reply(Term, Success, #{log := Log} = State) -> {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(#{cfg := Cfg, commit_index := CI0} = State0, Effects0) -> % TODO: shortcut function if commit index was not incremented State = #{commit_index := CI} = increment_commit_index(State0), Effects = case CI > CI0 of true -> put_counter(Cfg, ?C_RA_SVR_METRIC_COMMIT_INDEX, CI), [{aux, eval} | Effects0]; false -> Effects0 end, {State1, Effects1} = apply_to(CI, State, Effects), maybe_emit_pending_release_cursor(State1, Effects1). increment_commit_index(State0 = #{current_term := CurrentTerm}) -> PotentialNewCommitIndex = agreed_commit(match_indexes(State0)), % 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, State0) of {CurrentTerm, State} -> State#{commit_index => PotentialNewCommitIndex}; {_, State} -> State end. query_indexes(#{cfg := #cfg{id = Id}, cluster := Cluster, query_index := QueryIndex}) -> maps:fold(fun (PeerId, _, Acc) when PeerId == Id -> Acc; (_K, #{voter_status := #{membership := Membership}}, Acc) when Membership =/= voter -> Acc; (_K, #{query_index := Idx}, Acc) -> [Idx | Acc] end, [QueryIndex], Cluster). match_indexes(#{cfg := #cfg{id = Id}, cluster := Cluster, log := Log}) -> {LWIdx, _} = ra_log:last_written(Log), maps:fold(fun (PeerId, _, Acc) when PeerId == Id -> Acc; (_K, #{voter_status := #{membership := Membership}}, Acc) when Membership =/= voter -> Acc; (_K, #{match_index := Idx}, Acc) -> [Idx | Acc] end, [LWIdx], Cluster). -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, #{cfg := #cfg{log_id = LogId}}) -> ?DEBUG("~ts: ~w received unhandled msg: ~W", [LogId, RaState, Msg, 6]). fold_log_from(From, Folder, {St, Log0}) -> {To, _} = ra_log:last_index_term(Log0), case ra_log:fold(From, To, Folder, St, Log0) of {St1, Log} -> {ok, {St1, Log}} end. drop_existing(Log0, [], LastIdx) -> {Log0, [], LastIdx}; drop_existing(Log0, [{Idx, Trm, _} | Tail] = Entries, LastIdx) -> case ra_log:exists({Idx, Trm}, Log0) of {true, Log} -> drop_existing(Log, Tail, Idx); {false, Log} -> {Log, Entries, LastIdx} end. cast_reply(From, To, Msg) -> {cast, To, {From, Msg}}. index_machine_version(Idx, #{cfg := #cfg{machine_versions = Versions}}) -> %% scan for versions index_machine_version0(Idx, Versions). index_machine_version0(Idx, []) -> %% this _should_ never happen as you should never get a release cursor %% for an index that is lower than the last snapshot index exit({machine_version_for_index_not_known, {index, Idx}}); index_machine_version0(Idx, [{MIdx, V} | _]) when Idx >= MIdx -> V; index_machine_version0(Idx, [_ | Rem]) -> index_machine_version0(Idx, Rem). heartbeat_reply(CurTerm, QueryIndex) when is_integer(CurTerm) andalso is_integer(QueryIndex) -> #heartbeat_reply{term = CurTerm, query_index = QueryIndex}. update_heartbeat_rpc_effects(#{query_index := QueryIndex, queries_waiting_heartbeats := Waiting, current_term := Term, cfg := #cfg{id = Id}} = State) -> Peers = peers(State), %% TODO: do a quorum evaluation to find a queries to apply and apply all %% queries until that point case maps:size(Peers) of 0 -> %% Apply all if there are no peers. {_, QueryRefs} = lists:unzip(queue:to_list(Waiting)), Effects = apply_consistent_queries_effects(QueryRefs, State), {State#{queries_waiting_heartbeats => queue:new()}, Effects}; _ -> Effects = heartbeat_rpc_effects(Peers, Id, Term, QueryIndex), {State, Effects} end. make_heartbeat_rpc_effects(QueryRef, #{query_index := QueryIndex, queries_waiting_heartbeats := Waiting0, current_term := Term, cfg := #cfg{id = Id}} = State0) -> Peers = peers(State0), %% TODO: do a quorum evaluation to find a queries to apply and apply all %% queries until that point case maps:size(Peers) of 0 -> Effects = apply_consistent_queries_effects([QueryRef], State0), {State0, Effects}; _ -> NewQueryIndex = QueryIndex + 1, State = State0#{query_index => NewQueryIndex}, Effects = heartbeat_rpc_effects(Peers, Id, Term, NewQueryIndex), Waiting1 = queue:in({NewQueryIndex, QueryRef}, Waiting0), {State#{queries_waiting_heartbeats => Waiting1}, Effects} end. reset_query_index(#{cluster := Cluster} = State) -> State#{cluster => maps:map(fun(_PeerId, Peer) -> Peer#{query_index => 0} end, Cluster)}. heartbeat_rpc_effects(Peers, Id, Term, QueryIndex) -> lists:filtermap(fun({PeerId, Peer}) -> heartbeat_rpc_effect_for_peer(PeerId, Peer, Id, Term, QueryIndex) end, maps:to_list(Peers)). heartbeat_rpc_effect_for_peer(PeerId, #{status := normal} = Peer, Id, Term, QueryIndex) -> case maps:get(query_index, Peer, 0) < QueryIndex of true -> {true, {send_rpc, PeerId, #heartbeat_rpc{query_index = QueryIndex, term = Term, leader_id = Id}}}; false -> false end; heartbeat_rpc_effect_for_peer(_PeerId, _Peer, _Id, _Term, _QueryIndex) -> false. heartbeat_rpc_quorum(NewQueryIndex, PeerId, #{queries_waiting_heartbeats := Waiting0} = State) -> State1 = update_peer_query_index(PeerId, NewQueryIndex, State), ConsensusQueryIndex = get_current_query_quorum(State1), {QueryRefs, Waiting1} = take_from_queue_while( fun({QueryIndex, QueryRef}) -> case QueryIndex > ConsensusQueryIndex of true -> false; false -> {true, QueryRef} end end, Waiting0), case QueryRefs of [] -> {[], State1}; _ -> {QueryRefs, State1#{queries_waiting_heartbeats := Waiting1}} end. update_peer_query_index(PeerId, QueryIndex, #{cluster := Cluster} = State0) -> case maps:get(PeerId, Cluster, undefined) of undefined -> State0; #{query_index := PeerQueryIndex} = Peer -> case QueryIndex > PeerQueryIndex of true -> put_peer(PeerId, Peer#{query_index => QueryIndex}, State0); false -> State0 end end. get_current_query_quorum(State) -> agreed_commit(query_indexes(State)). -spec take_from_queue_while(fun((El) -> {true, Res} | false), queue:queue(El)) -> {[Res], queue:queue(El)}. take_from_queue_while(Fun, Queue) -> take_from_queue_while(Fun, Queue, []). take_from_queue_while(Fun, Queue, Result) -> case queue:peek(Queue) of {value, El} -> case Fun(El) of {true, ResVal} -> take_from_queue_while(Fun, queue:drop(Queue), [ResVal | Result]); false -> {Result, Queue} end; empty -> {Result, Queue} end. apply_consistent_queries_effects(QueryRefs, #{last_applied := LastApplied} = State) -> lists:map(fun({_, _, _, ReadCommitIndex} = QueryRef) -> true = LastApplied >= ReadCommitIndex, process_consistent_query(QueryRef, State) end, QueryRefs). process_consistent_query({query, From, QueryFun, _ReadCommitIndex}, #{cfg := #cfg{id = Id, machine = {machine, MacMod, _}, effective_machine_version = MacVer}, machine_state := MacState, last_applied := Last, current_term := Term}) -> Ctx = #{index => Last, term => Term, machine_version => MacVer}, Result = ra_machine:query(MacMod, QueryFun, MacState, Ctx), {reply, From, {ok, Result, Id}}; process_consistent_query({aux, From, AuxCmd, _ReadCommitIndex}, _State0) -> {next_event, {call, From}, {aux_command, {'$wrap_reply', AuxCmd}}}. process_pending_consistent_queries(#{cluster_change_permitted := false} = State0, Effects0) -> {State0, Effects0}; process_pending_consistent_queries(#{pending_consistent_queries := []} = State0, Effects0) -> {State0, Effects0}; process_pending_consistent_queries(#{cluster_change_permitted := true, pending_consistent_queries := Pending} = State0, Effects0) -> %% TODO: submit all pending queries with a single query index. lists:foldl( fun(QueryRef, {State, Effects}) -> {NewState, NewEffects} = make_heartbeat_rpc_effects(QueryRef, State), {NewState, NewEffects ++ Effects} end, {State0#{pending_consistent_queries => []}, Effects0}, Pending). incr_counter(#cfg{counter = Cnt}, Ix, N) when Cnt =/= undefined -> counters:add(Cnt, Ix, N); incr_counter(#cfg{counter = undefined}, _Ix, _N) -> ok. put_counter(#cfg{counter = Cnt}, Ix, N) when Cnt =/= undefined -> counters:put(Cnt, Ix, N); put_counter(#cfg{counter = undefined}, _Ix, _N) -> ok. meta_name(#cfg{system_config = #{names := #{log_meta := Name}}}) -> Name; meta_name(#{names := #{log_meta := Name}}) -> Name. already_member(State, Effects) -> % already a member do nothing % TODO: reply? If we don't reply the caller may block until timeout {not_appended, already_member, State, Effects}. %%% ==================== %%% Voter status helpers %%% ==================== -spec ensure_promotion_target(ra_voter_status(), ra_server_state()) -> {ok, ra_voter_status()} | {error, term()}. ensure_promotion_target(#{membership := promotable, target := _, uid := _} = Status, _) -> {ok, Status}; ensure_promotion_target(#{membership := promotable, uid := _} = Status, #{log := Log}) -> %% The next index in the log is used by for a cluster change command: %% the caller of `ensure_promotion_target/2' also calls %% `append_cluster_change/5'. So even if a peer joins a cluster which isn't %% handling any other commands, this promotion target will be reachable. Target = ra_log:next_index(Log), {ok, Status#{target => Target}}; ensure_promotion_target(#{membership := promotable}, _) -> {error, missing_uid}; ensure_promotion_target(Voter, _) -> {ok, Voter}. %% Get membership of a given Id+UId from a (possibly new) cluster. -spec get_membership(ra_cluster() | ra_cluster_snapshot() | ra_cluster_servers(), ra_server_id(), ra_uid(), ra_membership()) -> ra_membership(). get_membership(_Cluster, _PeerId, _UId, Default) when is_list(_Cluster) -> %% Legacy cluster snapshot does not retain voter_status. Default; get_membership(Cluster, PeerId, UId, Default) -> case maps:get(PeerId, Cluster, undefined) of #{voter_status := #{uid := UId} = VoterStatus} -> maps:get(membership, VoterStatus, Default); _ -> Default end. -spec get_condition_timeout(ra_server_state(), Default :: term()) -> term() | integer(). get_condition_timeout(#{condition := #{timeout := #{duration := D}}}, _Def) -> D; get_condition_timeout(_, Def) -> Def. %% Get this node's membership from a (possibly new) cluster. %% Defaults to last known-locally value. -spec get_membership(ra_cluster() | ra_cluster_snapshot() | ra_cluster_servers(), ra_server_state()) -> ra_membership(). get_membership(Cluster, #{cfg := #cfg{id = Id, uid = UId}} = State) -> Default = maps:get(membership, State, voter), get_membership(Cluster, Id, UId, Default). %% Get this node's membership. %% Defaults to last known-locally value. -spec get_membership(ra_server_state()) -> ra_membership(). get_membership(#{cfg := #cfg{id = Id, uid = UId}, cluster := Cluster} = State) -> Default = maps:get(membership, State, voter), get_membership(Cluster, Id, UId, Default). -spec maybe_promote_peer(ra_server_id(), ra_server_state(), effects()) -> effects(). maybe_promote_peer(PeerId, #{cluster := Cluster}, Effects) -> case Cluster of #{PeerId := #{match_index := MI, voter_status := #{membership := promotable, target := Target} = OldStatus}} when MI >= Target -> Promote = {next_event, {command, {'$ra_join', #{ts => os:system_time(millisecond)}, #{id => PeerId, voter_status => OldStatus#{ membership => voter }}, noreply}}}, [Promote | Effects]; _ -> Effects end. -spec required_quorum(ra_cluster()) -> pos_integer(). required_quorum(Cluster) -> Voters = count_voters(Cluster), trunc(Voters / 2) + 1. count_voters(Cluster) -> maps:fold( fun (_, #{voter_status := #{membership := Membership}}, Count) when Membership =/= voter -> Count; (_, _, Count) -> Count + 1 end, 0, Cluster). append_error_reply(Cmd, Reason, Effects0) -> case Cmd of {_, #{from := From}, _, _} -> [{reply, From, {error, Reason}} | Effects0]; _ -> Effects0 end. after_log_append_reply(Cmd, Idx, Term, Effects0) -> case Cmd of {_, #{from := From}, _, after_log_append} -> [{reply, From, {wrap_reply, {Idx, Term}}} | Effects0]; _ -> Effects0 end. post_election_effects(#{cfg := #cfg{effective_machine_version = EffectiveMacVer, machine = Mac, system_config = SystemConfig}} = State) -> Peers = peers(State), PeerIds = maps:keys(Peers), MachineUpgradeStrategy = maps:get(machine_upgrade_strategy, SystemConfig, ?DEFAULT_MACHINE_UPGRADE_STRATEGY), UpgradeMachineNow = (PeerIds =:= [] orelse MachineUpgradeStrategy =:= quorum), case UpgradeMachineNow of true -> %% This node is alone in the cluster, we can send the `noop' %% command with the newer machine version right away. MacVer = ra_machine:version(Mac), Noop = {noop, #{ts => erlang:system_time(millisecond)}, MacVer}, NoopEffect = {next_event, cast, {command, Noop}}, [NoopEffect]; false -> %% We continue to send the `noop' command immediately after %% becoming a leader, but compared to Ra 2.15 and older, we don't %% set the machine version to the latest: we keep the same %% effective machine version for now. %% %% However, we query info keys from all peers, including their %% supported machine version. The replies will be used to %% determine the max supported machine version and when it is %% greater than the effective one, another `noop' command will be %% sent. Noop = {noop, #{ts => erlang:system_time(millisecond)}, EffectiveMacVer}, NoopEffect = {next_event, cast, {command, Noop}}, InfoRpcEffects = info_rpc_effects(State), [NoopEffect | InfoRpcEffects] end. info_rpc_effects(#{cfg := #cfg{id = Id}, cluster := Cluster} = State) -> InfoRpcEffects = maps:fold( fun (PeerId, _, Acc) when PeerId =:= Id -> Acc; (PeerId, _, Acc) -> Acc ++ info_rpc_effects_for_peer(State, PeerId) end, [], Cluster), InfoRpcEffects. info_rpc_effects_for_peer(#{cluster := Cluster, current_term := CurTerm} = State, PeerId) -> %% We determine if we need to ask (for the fist time or again) the info %% from a peer. SendRpc = case Cluster of #{PeerId := #{machine_version := PeerMacVer}} -> %% We have the machine version of the peer, but we want %% to ask again if that version is old, in the hope the %% peer restarted and was updated. MacVer = machine_version(State), PeerMacVer < MacVer; _ -> %% We don't have any details about the peer, we ask. true end, case SendRpc of true -> %% We ask for all info keys currently. If we ask for specific %% keys, we will have to handle merging of already known info keys %% and updates. Id = id(State), Command = #info_rpc{from = Id, term = CurTerm, keys = [machine_version]}, [{send_rpc, PeerId, Command}]; false -> [] end. info_reply_effect(#{current_term := CurTerm} = State, #info_rpc{from = FromId, keys = Keys}) -> Id = id(State), Info = ra_server_info(State, Keys), InfoReply = #info_reply{from = Id, term = CurTerm, keys = Keys, info = Info}, {cast, FromId, InfoReply}. empty_info_reply_effect(#{current_term := CurTerm} = State, #info_rpc{from = FromId, keys = Keys}) -> Id = id(State), InfoReply = #info_reply{from = Id, term = CurTerm, keys = Keys, info = #{}}, {cast, FromId, InfoReply}. ra_server_info(State) -> MacVer = machine_version(State), #{machine_version => MacVer}. ra_server_info(State, Keys) -> %% Note that the node that asked may ask for keys we don't know. Info0 = ra_server_info(State), Info1 = maps:filter( fun(Key, _Value) -> lists:member(Key, Keys) end, Info0), Info1. handle_info_reply(#{cluster := Cluster} = State, #info_reply{from = PeerId, keys = Keys, info = Info}) when is_map_key(PeerId, Cluster) -> PeerState0 = maps:get(PeerId, Cluster), PeerState1 = lists:foldl(fun(Key, PS) -> case Info of #{Key := Value} -> PS#{Key => Value}; _ -> maps:remove(Key, PS) end end, PeerState0, Keys), Cluster1 = Cluster#{PeerId => PeerState1}, State1 = State#{cluster => Cluster1}, determine_if_machine_upgrade_allowed(State1); handle_info_reply(State, #info_reply{} = _InfoReply) -> %% The Ra node who sent the `#info_reply{}' was removed from the cluster %% since this node asked for its info. We ignore its reply. {State, []}. determine_if_machine_upgrade_allowed( #{cfg := #cfg{effective_machine_version = EffectiveMacVer, log_id = LogId}} = State) -> Effects = case get_max_supported_machine_version(State) of MaxSupMacVer when MaxSupMacVer > EffectiveMacVer -> ?DEBUG( "~ts: max supported machine version = ~b, " "upgrading from ~b", [LogId, MaxSupMacVer, EffectiveMacVer]), Noop = {noop, #{ts => erlang:system_time(millisecond)}, MaxSupMacVer}, [{next_event, cast, {command, Noop}}]; _ -> [] end, {State, Effects}. get_max_supported_machine_version( #{cfg := #cfg{effective_machine_version = EffectiveMacVer, id = Id}, cluster := Cluster} = State) -> MacVer = machine_version(State), MaxSupMacVer = maps:fold( fun (PeerId, #{machine_version := PeerMacVer}, Max) when PeerId =/= Id andalso PeerMacVer < Max -> %% This peer has a lower machine version than the %% previous peers in the list so far. This becomes %% the highest machine version we can upgrade to. PeerMacVer; (PeerId, PeerState, _Max) when PeerId =/= Id andalso not is_map_key(machine_version, PeerState) -> %% We don't have the machine version of one of the %% peers yet. We need to stay on the effective %% machine version. EffectiveMacVer; (_PeerId, _PeerState, Max) -> %% Either this is this Ra server or this is a peer %% using a newer machine version. Max end, MacVer, Cluster), MaxSupMacVer. %%% =================== %%% Internal unit tests %%% =================== -ifdef(TEST). -include_lib("eunit/include/eunit.hrl"). index_machine_version0_test() -> S0 = [{0, 0}], ?assertEqual(0, index_machine_version0(0, S0)), ?assertEqual(0, index_machine_version0(1123456, S0)), S1 = [{100, 4}, {50, 3}, {25, 2}], ?assertEqual(4, index_machine_version0(101, S1)), ?assertEqual(4, index_machine_version0(100, S1)), ?assertEqual(3, index_machine_version0(99, S1)), ?assertEqual(2, index_machine_version0(49, S1)), ?assertEqual(2, index_machine_version0(25, S1)), ?assertExit({machine_version_for_index_not_known, _}, index_machine_version0(24, S1)), ok. 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.