Packages
ra
3.1.1
3.1.9
3.1.8
3.1.7
3.1.6
3.1.5
3.1.4
3.1.3
3.1.2
3.1.1
3.1.0
3.0.2
3.0.1
3.0.0
3.0.0-beta.1
2.17.3
2.17.2
2.17.1
2.17.0
2.16.13
2.16.12
2.16.11
2.16.10
2.16.9
2.16.8
2.16.7
2.16.6
2.16.5
2.16.4
2.16.3
2.16.2
2.16.1
2.16.0
2.16.0-pre.12
2.16.0-pre.11
2.16.0-pre.10
2.16.0-pre.9
2.16.0-pre.8
2.16.0-pre.7
2.16.0-pre.6
2.16.0-pre.5
2.16.0-pre.4
2.16.0-pre.3
2.16.0-pre.2
2.16.0-pre.1
2.15.4
2.15.3
2.15.2
2.15.1
2.15.0
2.14.0
2.13.6
2.13.5
2.13.4
2.13.3
2.13.2
2.13.1
2.13.0
2.13.0-pre.1
2.12.0
2.11.0
2.11.0-pre.1
2.10.2-pre.2
2.10.2-pre.1
2.10.1
2.10.0
2.10.0-pre.3
2.10.0-pre.2
2.10.0-pre.1
2.9.10-pre.1
2.9.1
2.9.1-pre.2
2.9.1-pre.1
2.9.0
2.8.0
retired
2.7.3
2.7.2
2.7.1
2.7.0
2.7.0-pre.3
2.7.0-pre.2
2.7.0-pre.1
2.6.3
2.6.2
2.6.1
2.6.0-pre.1
2.5.1
2.5.1-pre.1
2.5.0
2.4.9
2.4.8
2.4.7
2.4.6
2.4.5
2.4.4
2.4.3
2.4.2
retired
2.4.1
2.4.0
2.3.0
2.2.0
2.1.0
2.0.13
2.0.12
2.0.11
2.0.10
2.0.9
2.0.8
2.0.7
2.0.6
2.0.5
2.0.4
2.0.3
2.0.2
2.0.1
2.0.0
1.1.9
1.1.8
1.1.7
1.1.6
1.1.5
1.1.4
1.1.3
1.1.2
1.1.1
1.1.0
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.9.6
0.9.5
0.9.4
0.9.2
0.3.3
retired
0.3.2
retired
0.3.1
retired
Raft library
Current section
Files
Jump to
Current section
Files
src/ra_server_proc.erl
%% This Source Code Form is subject to the terms of the Mozilla Public
%% 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_proc).
-behaviour(gen_statem).
-compile({inline, [handle_raft_state/3]}).
-include_lib("stdlib/include/assert.hrl").
-include("ra.hrl").
-include("ra_server.hrl").
%% State functions
-export([
post_init/3,
recover/3,
recovered/3,
leader/3,
pre_vote/3,
candidate/3,
follower/3,
receive_snapshot/3,
await_condition/3,
terminating_leader/3,
terminating_follower/3
]).
%% gen_statem callbacks
-export([
init/1,
format_status/1,
handle_event/4,
terminate/3,
code_change/4,
callback_mode/0
]).
%% API
-export([start_link/1,
command/3,
cast_command/2,
cast_command/3,
query/5,
state_query/3,
local_state_query/3,
trigger_election/2,
ping/2,
read_entries/4,
transfer_leadership/3,
force_shrink_members_to_current_member/1
]).
-export([send_rpc/3]).
-ifdef(TEST).
-export([leader_call/3,
local_call/3]).
-endif.
-define(DEFAULT_BROADCAST_TIME, 100).
-define(DEFAULT_ELECTION_MULT, 5).
-define(TICK_INTERVAL_MS, 1000).
-define(DEFAULT_AWAIT_CONDITION_TIMEOUT, 30000).
%% Utilisation average calculations are all in μs.
-define(INSTALL_SNAP_RPC_TIMEOUT, 120 * 1000).
-define(HANDLE_EFFECTS(Effects, EvtType, State0),
handle_effects(?FUNCTION_NAME, Effects, EvtType, State0)).
-define(ASYNC_DIST(Node, Send),
case Node == node() of
true ->
Send,
ok;
false ->
%% use async_dist for remote sends
process_flag(async_dist, true),
Send,
process_flag(async_dist, false),
ok
end).
-type query_fun() :: ra:query_fun().
-type query_options() :: #{condition => ra:query_condition()}.
-type ra_command() :: {ra_server:command_type(), term(),
ra_server:command_reply_mode()}.
-type ra_leader_call_ret(Result) :: {ok, Result, Leader::ra_server_id()} |
{error, term()} |
{timeout, ra_server_id()}.
-type ra_local_call_ret(Result) :: {ok, Result, LocalServer::ra_server_id()} |
{error, term()} |
{timeout, ra_server_id()}.
-type ra_cmd_ret() :: ra_leader_call_ret(term()).
-type gen_statem_start_ret() :: {ok, pid()} | ignore | {error, term()}.
%% ra_event types
-type ra_event_reject_detail() :: {not_leader, Leader :: option(ra_server_id()),
ra_server:command_correlation()}.
-type ra_event_body() ::
% used for notifying senders of the ultimate fate of their command
% sent using ra:pipeline_command/3|4
{applied, [{ra_server:command_correlation(), Reply :: term()}]} |
{rejected, ra_event_reject_detail()} |
% used to send message side-effects emitted by the state machine
{machine, term()}.
-type ra_event() :: {ra_event, Sender :: ra_server_id(), ra_event_body()}.
%% the Sender is the ra process that emitted the ra_event.
-type server_loc() :: ra_server_id() | [ra_server_id()].
-export_type([ra_leader_call_ret/1,
ra_local_call_ret/1,
ra_cmd_ret/0,
safe_call_ret/1,
ra_event_reject_detail/0,
ra_event/0,
ra_event_body/0,
query_options/0]).
%% the ra server proc keeps monitors on behalf of different components
%% the state machine, log and server code. The tag is used to determine
%% which component to dispatch the down to
-record(conf, {log_id :: unicode:chardata(),
name :: atom(),
cluster_name :: term(),
broadcast_time = ?DEFAULT_BROADCAST_TIME :: non_neg_integer(),
tick_timeout :: non_neg_integer(),
await_condition_timeout :: non_neg_integer(),
ra_event_formatter :: undefined | {module(), atom(), [term()]},
flush_commands_size = ?FLUSH_COMMANDS_SIZE :: non_neg_integer(),
snapshot_chunk_size = ?DEFAULT_SNAPSHOT_CHUNK_SIZE :: non_neg_integer(),
receive_snapshot_timeout = ?DEFAULT_RECEIVE_SNAPSHOT_TIMEOUT :: non_neg_integer(),
install_snap_rpc_timeout :: non_neg_integer(),
aten_poll_interval = 1000 :: non_neg_integer(),
counter :: undefined | counters:counters_ref(),
worker_pid :: pid()
}).
-record(state, {conf :: #conf{},
server_state :: ra_server:ra_server_state(),
monitors = ra_monitors:init() :: ra_monitors:state(),
pending_commands = [] :: [{{pid(), any()}, term()}],
leader_monitor :: reference() | undefined,
leader_last_seen :: integer() | undefined,
low_priority_commands :: ra_ets_queue:state(),
election_timeout_set = false :: boolean(),
%% the log index last time gc was forced
pending_notifys = #{} :: #{pid() => [term()]},
pending_queries = [] :: [{ra:query_condition(),
gen_statem:from(),
query_fun()}],
%% {LeakyIntegrator, LastCommitIndex, LastIntervalRate} for commit rate tracking
commit_rate :: {ra_li:state(), non_neg_integer(), float()}
}).
%%%===================================================================
%%% API
%%%===================================================================
-spec start_link(ra_server:ra_server_config()) -> gen_statem_start_ret().
start_link(Config = #{id := Id}) ->
Name = ra_lib:ra_server_id_to_local_name(Id),
gen_statem:start_link({local, Name}, ?MODULE, Config, []).
-spec command(server_loc(), ra_command(), timeout()) ->
ra_cmd_ret().
command(ServerLoc, Cmd, Timeout) ->
leader_call(ServerLoc, {command, normal, Cmd}, Timeout).
-spec cast_command(ra_server_id(), ra_command()) -> ok.
cast_command(ServerId, Cmd) ->
gen_statem:cast(ServerId, {command, low, Cmd}).
-spec cast_command(ra_server_id(), ra_server:command_priority(), ra_command()) -> ok.
cast_command(ServerId, Priority, Cmd) ->
gen_statem:cast(ServerId, {command, Priority, Cmd}).
-spec query(server_loc(), AuxCmd :: term() | query_fun(),
local | consistent | consistent_aux | leader,
query_options(),
timeout()) ->
ra_server_proc:ra_leader_call_ret({ra_idxterm(), Reply :: term()})
| ra_server_proc:ra_leader_call_ret(Reply :: term())
| {ok, {ra_idxterm(), Reply :: term()}, not_known}.
query(ServerLoc, QueryFun, local, Options, Timeout)
when map_size(Options) =:= 0 ->
statem_call(ServerLoc, {local_query, QueryFun}, Timeout);
query(ServerLoc, QueryFun, local, Options, Timeout) ->
statem_call(ServerLoc, {local_query, QueryFun, Options}, Timeout);
query(ServerLoc, QueryFun, leader, Options, Timeout)
when map_size(Options) =:= 0 ->
leader_call(ServerLoc, {local_query, QueryFun}, Timeout);
query(ServerLoc, QueryFun, leader, Options, Timeout) ->
leader_call(ServerLoc, {local_query, QueryFun, Options}, Timeout);
query(ServerLoc, QueryFun, consistent, _Options, Timeout) ->
leader_call(ServerLoc, {consistent_query, QueryFun}, Timeout);
query(ServerLoc, AuxCmd, consistent_aux, _Options, Timeout) ->
leader_call(ServerLoc, {consistent_aux, AuxCmd}, Timeout).
-spec read_entries(ra_server_id(), [ra:index()],
undefined | ra_flru:state(),
timeout()) ->
{ok, {map(), ra_flru:state()}} | {error, term()}.
read_entries({_, Node} = ServerId, Indexes, Flru0, Timeout)
when Node == node() ->
case local_call(ServerId, {read_entries, Indexes}, Timeout) of
{ok, ReadPlan} ->
{Reads, Flru} = ra_log:execute_read_plan(ReadPlan, Flru0,
fun (Idx, Term, Cmd) ->
{Idx, Term, Cmd}
end,
#{access_pattern => random,
file_advise => random,
index_mode => binary}),
{ok, {Reads, Flru}};
Err ->
Err
end.
%% used to query the raft state rather than the machine state
-spec state_query(server_loc(),
all |
overview |
voters |
members |
members_info |
initial_members |
machine, timeout()) ->
ra_leader_call_ret(term()).
state_query(ServerLoc, Spec, Timeout) ->
leader_call(ServerLoc, {state_query, Spec}, Timeout).
-spec local_state_query(server_loc(),
all |
overview |
voters |
members |
members_info |
initial_members |
machine, timeout()) ->
ra_local_call_ret(term()).
local_state_query(ServerLoc, Spec, Timeout) ->
local_call(ServerLoc, {state_query, Spec}, Timeout).
-spec trigger_election(ra_server_id(), timeout()) -> ok.
trigger_election(ServerId, Timeout) ->
gen_statem:call(ServerId, trigger_election, Timeout).
-spec transfer_leadership(ra_server_id(), ra_server_id(), timeout()) ->
ok | already_leader | {error, term()} | {timeout, ra_server_id()}.
transfer_leadership(ServerId, TargetServerId, Timeout) ->
leader_call(ServerId, {transfer_leadership, TargetServerId}, Timeout).
-spec force_shrink_members_to_current_member(ra_server_id()) -> ok.
force_shrink_members_to_current_member(ServerId) ->
gen_statem_safe_call(ServerId, force_member_change, 5000).
-spec ping(ra_server_id(), timeout()) -> safe_call_ret({pong, states()}).
ping(ServerId, Timeout) ->
gen_statem_safe_call(ServerId, ping, Timeout).
leader_call(ServerLoc, Msg, Timeout) ->
statem_call(ServerLoc, {leader_call, Msg}, Timeout).
local_call(ServerLoc, Msg, Timeout) ->
statem_call(ServerLoc, {local_call, Msg}, Timeout).
statem_call(ServerIds, Msg, Timeout)
when is_list(ServerIds) ->
multi_statem_call(ServerIds, Msg, [], Timeout);
statem_call(ServerId, Msg, Timeout) ->
case gen_statem_safe_call(ServerId, Msg, Timeout) of
{redirect, Leader} ->
statem_call(Leader, Msg, Timeout);
{wrap_reply, Reply} ->
{ok, Reply, ServerId};
{error, _} = E ->
E;
timeout ->
{timeout, ServerId};
Reply ->
Reply
end.
multi_statem_call([ServerId | ServerIds], Msg, Errs, Timeout) ->
case statem_call(ServerId, Msg, Timeout) of
{Tag, Info} = E
when Tag == timeout orelse
(Tag == error andalso
(Info == noproc orelse
Info == nodedown orelse
Info == shutdown orelse
Info == system_not_started)) ->
%% these are the retryable errors, any others we consider
%% genuine errors that a retry will not fix
case ServerIds of
[] ->
{error, {no_more_servers_to_try, [E | Errs]}};
_ ->
multi_statem_call(ServerIds, Msg, [E | Errs], Timeout)
end;
Reply ->
Reply
end.
%%%===================================================================
%%% gen_statem callbacks
%%%===================================================================
init(#{reply_to := ReplyTo} = Config) ->
{ok, post_init, maps:remove(reply_to, Config),
[{next_event, internal, {go, ReplyTo}}]}.
do_init(#{id := Id,
uid := UId,
parent := ParentPid,
cluster_name := ClusterName} = Config0)
when is_pid(ParentPid) ->
Key = ra_lib:ra_server_id_to_local_name(Id),
true = ets:insert(ra_state, {Key, init, unknown}),
process_flag(trap_exit, true),
MetricLabels = maps:get(metrics_labels, Config0, #{}),
Config = maps:merge(config_defaults(Id, MetricLabels), Config0),
#{counter := Counter,
system_config := #{names := Names} = SysConf} = Config,
MsgQData = maps:get(message_queue_data, SysConf, off_heap),
MinBinVheapSize = maps:get(server_min_bin_vheap_size, SysConf,
?MIN_BIN_VHEAP_SIZE),
MinHeapSize = maps:get(server_min_heap_size, SysConf, ?MIN_BIN_VHEAP_SIZE),
process_flag(message_queue_data, MsgQData),
process_flag(min_bin_vheap_size, MinBinVheapSize),
process_flag(min_heap_size, MinHeapSize),
#{names := Names} = SysConf,
%% New servers should register _after_ log initialisation to ensure the
%% config file is fully written as it is required for successful recovery
IsNew = not ra_directory:is_registered_uid(Names, UId),
Parent = maps:get(parent, Config, undefined),
if not IsNew ->
ok = ra_directory:register_name(Names, UId, self(),
Parent, Key, ClusterName);
true -> ok
end,
#{cluster := Cluster} = ServerState = ra_server:init(Config),
if IsNew ->
ok = ra_directory:register_name(Names, UId, self(),
Parent, Key, ClusterName);
true -> ok
end,
% ensure each relevant erlang node is connected
PeerNodes = [PeerNode ||
{_, PeerNode} <- maps:keys(maps:remove(Id, Cluster))],
case PeerNodes -- nodes() of
[] ->
%% all peer nodes are connected
ok;
DisconnectedNodes ->
%% as most messages are sent using noconnect we explicitly attempt to
%% connect to all relevant nodes
_ = spawn(fun () ->
[net_kernel:connect_node(N)
|| N <- DisconnectedNodes]
end),
ok
end,
TickTime = maps:get(tick_timeout, Config),
InstallSnapRpcTimeout = maps:get(install_snap_rpc_timeout, Config),
AwaitCondTimeout = maps:get(await_condition_timeout, Config),
RaEventFormatterMFA = maps:get(ra_event_formatter, Config, undefined),
FlushCommandsSize = maps:get(low_priority_commands_flush_size, SysConf,
?FLUSH_COMMANDS_SIZE),
SnapshotChunkSize = maps:get(snapshot_chunk_size, SysConf,
?DEFAULT_SNAPSHOT_CHUNK_SIZE),
ReceiveSnapshotTimeout = maps:get(receive_snapshot_timeout, SysConf,
?DEFAULT_RECEIVE_SNAPSHOT_TIMEOUT),
AtenPollInt = application:get_env(aten, poll_interval, 1000),
LogId = ra_server:log_id(ServerState),
%% TODO: full error handling
WorkerPid = case ra_server_sup:start_ra_worker(ParentPid, Config) of
{ok, P} -> P;
{error, {already_started, P}} ->
P
end,
ra_env:configure_logger(logger),
%% monitor worker process, it is easier to handle than linking as we're
%% already processing all downs
_ = monitor(process, WorkerPid),
#{commit_index := CommitIndex} = ServerState,
State = #state{conf = #conf{log_id = LogId,
cluster_name = ClusterName,
name = Key,
tick_timeout = TickTime,
await_condition_timeout = AwaitCondTimeout,
ra_event_formatter = RaEventFormatterMFA,
flush_commands_size = FlushCommandsSize,
snapshot_chunk_size = SnapshotChunkSize,
install_snap_rpc_timeout = InstallSnapRpcTimeout,
receive_snapshot_timeout = ReceiveSnapshotTimeout,
aten_poll_interval = AtenPollInt,
counter = Counter,
worker_pid = WorkerPid},
low_priority_commands = ra_ets_queue:new(),
server_state = ServerState,
commit_rate = {ra_li:new(TickTime * 6), CommitIndex, 0.0}},
ok = net_kernel:monitor_nodes(true, [nodedown_reason]),
State.
%% callback mode
callback_mode() -> [state_functions, state_enter].
%%%===================================================================
%%% State functions
%%%===================================================================
post_init(enter, _OldState, State) ->
{keep_state, State, []};
post_init(internal, {go, {ReplyToRef, ReplyToPid}}, Config) ->
State = do_init(Config),
ReplyToPid ! {ReplyToRef, ok},
{next_state, recover, State, [{next_event, internal, go}]}.
recover(enter, OldState, State0) ->
{State, Actions} = handle_enter(?FUNCTION_NAME, OldState, State0),
{keep_state, State, Actions};
recover(internal, go, State = #state{server_state = ServerState0}) ->
ServerState = ra_server:recover(ServerState0),
incr_counter(State#state.conf, ?C_RA_SRV_GCS, 1),
%% we have to issue the next_event here so that the recovered state is
%% only passed through very briefly
next_state(recovered, State#state{server_state = ServerState},
[{next_event, internal, next}]);
recover(_, _, State) ->
% all other events need to be postponed until we can return
% `next_event` from init
{keep_state, State, {postpone, true}}.
%% this is a passthrough state to allow state machines to emit node local
%% effects post recovery
recovered(enter, OldState, State0) ->
{State, Actions} = handle_enter(?FUNCTION_NAME, OldState, State0),
ok = record_cluster_change(State),
{keep_state, State, Actions};
recovered(internal, next, State) ->
true = erlang:garbage_collect(),
next_state(follower, State, set_tick_timer(State, [])).
leader(enter, OldState, #state{low_priority_commands = Delayed0} = State0) ->
{State, Actions} = handle_enter(?FUNCTION_NAME, OldState, State0),
Delayed = case OldState of
await_condition ->
%% if we're returning from await_condition we may still
%% have valid delayed commands to schedule
schedule_command_flush(Delayed0),
Delayed0;
_ ->
%% for any other state it is best to just reset the
%% delayed commands
ra_ets_queue:reset(Delayed0)
end,
ok = record_cluster_change(State),
{keep_state, State#state{leader_last_seen = undefined,
pending_notifys = #{},
low_priority_commands = Delayed,
election_timeout_set = false}, Actions};
leader(EventType, {leader_call, Msg}, State) ->
% no need to redirect
leader(EventType, Msg, State);
leader(EventType, {local_call, Msg}, State) ->
leader(EventType, Msg, State);
leader(EventType, {leader_cast, Msg}, State) ->
leader(EventType, Msg, State);
leader(EventType, {command, normal, {CmdType, Data, ReplyMode}},
#state{conf = Conf} = State0) ->
case validate_reply_mode(ReplyMode) of
ok ->
%% normal priority commands are written immediately
Cmd = make_command(CmdType, EventType, Data, ReplyMode),
{NextState, State1, Effects} = handle_leader({command, Cmd}, State0),
{State, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
case NextState of
leader ->
{keep_state, State, Actions};
_ ->
next_state(NextState, State, Actions)
end;
Error ->
ok = incr_counter(Conf, ?C_RA_SRV_INVALID_REPLY_MODE_COMMANDS, 1),
case EventType of
{call, From} ->
{keep_state, State0, [{reply, From, Error}]};
_ ->
{keep_state, State0, []}
end
end;
leader(EventType, {command, low, {'$usr', Data, ReplyMode}},
#state{conf = Conf,
low_priority_commands = Delayed} = State0) ->
%% only user commands can be low priority
case validate_reply_mode(ReplyMode) of
ok ->
%% cache the low priority command until the flush_commands message
%% arrives
Cmd = make_command('$usr', EventType, Data, ReplyMode),
%% if there are no prior delayed commands
%% (and thus no action queued to do so)
%% queue a state timeout to flush them
%% We use a cast to ourselves instead of a zero timeout as we want
%% to get onto the back of the erlang mailbox not just the current
%% gen_statem event buffer.
case ra_ets_queue:len(Delayed) of
0 ->
ok = gen_statem:cast(self(), flush_commands);
_ ->
ok
end,
State = State0#state{low_priority_commands =
ra_ets_queue:in(Cmd, Delayed)},
{keep_state, State, []};
Error ->
ok = incr_counter(Conf, ?C_RA_SRV_INVALID_REPLY_MODE_COMMANDS, 1),
case EventType of
{call, From} ->
{keep_state, State0, [{reply, From, Error}]};
_ ->
{keep_state, State0, []}
end
end;
leader(EventType, {command, low, Cmd}, #state{} = State) ->
%% non user low priority commands are upgraded to normal priority
leader(EventType, {command, normal, Cmd}, State);
leader(EventType, {aux_command, Cmd}, State0) ->
{_, ServerState, Effects} = ra_server:handle_aux(?FUNCTION_NAME, EventType,
Cmd, State0#state.server_state),
{State, Actions} =
?HANDLE_EFFECTS(Effects, EventType,
State0#state{server_state = ServerState}),
{keep_state, State, Actions};
leader(EventType, flush_commands,
#state{conf = #conf{flush_commands_size = Size},
low_priority_commands = Delayed0} = State0) ->
{Commands, Delayed} = ra_ets_queue:take(Size, Delayed0),
%% write a batch of delayed commands
{NextState, State1, Effects} = handle_leader({commands, Commands}, State0),
State2 = State1#state{low_priority_commands = Delayed},
{State, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State2),
case NextState of
leader ->
schedule_command_flush(Delayed),
{keep_state, State#state{low_priority_commands = Delayed}, Actions};
_ ->
next_state(NextState, State, Actions)
end;
leader({call, _From} = EventType, {local_query, QueryFun}, State) ->
leader(EventType, {local_query, QueryFun, #{}}, State);
leader({call, From} = EventType, {local_query, QueryFun, Options}, State) ->
evaluate_local_query(leader, EventType, From, QueryFun, Options, State);
leader({call, From}, {state_query, Spec}, State) ->
Reply = {ok, do_state_query(Spec, State), id(State)},
{keep_state, State, [{reply, From, Reply}]};
leader({call, From}, {consistent_query, QueryFun},
#state{conf = Conf,
server_state = ServerState0} = State0) ->
{leader, ServerState1, Effects} =
ra_server:handle_leader({consistent_query, From, QueryFun},
ServerState0),
incr_counter(Conf, ?C_RA_SRV_CONSISTENT_QUERIES, 1),
{State1, Actions} =
?HANDLE_EFFECTS(Effects, {call, From},
State0#state{server_state = ServerState1}),
{keep_state, State1, Actions};
leader({call, From}, {consistent_aux, AuxCmd},
#state{conf = Conf,
server_state = ServerState0} = State0) ->
{leader, ServerState1, Effects} =
ra_server:handle_leader({consistent_aux, From, AuxCmd},
ServerState0),
incr_counter(Conf, ?C_RA_SRV_CONSISTENT_QUERIES, 1),
{State1, Actions} =
?HANDLE_EFFECTS(Effects, {call, From},
State0#state{server_state = ServerState1}),
{keep_state, State1, Actions};
leader({call, From}, ping, State) ->
{keep_state, State, [{reply, From, {pong, leader}}]};
leader(info, {node_event, _Node, _Evt}, State) ->
{keep_state, State, []};
leader(info, {'DOWN', _MRef, process, Pid, Info}, State0) ->
handle_process_down(Pid, Info, ?FUNCTION_NAME, State0);
leader(info, {Status, Node, InfoList}, State0)
when Status =:= nodedown orelse
Status =:= nodeup ->
handle_node_status_change(Node, Status, InfoList, ?FUNCTION_NAME, State0);
leader(info, {unsuspend_peer, PeerId}, State0) ->
State = case ra_server:peer_status(PeerId, State0#state.server_state) of
suspended ->
update_peer(PeerId, #{status => normal}, State0);
_ ->
State0
end,
{keep_state, State, []};
leader(_, tick_timeout,
#state{conf = #conf{tick_timeout = TickTimeMs},
commit_rate = CommitRate0} = State0) ->
{State1, RpcEffs} = make_rpcs(State0),
ServerState0 = State1#state.server_state,
CommitRate = update_commit_rate(CommitRate0, ServerState0, TickTimeMs),
Effects = ra_server:tick(ServerState0),
ServerState = ra_server:log_tick(ServerState0),
{State2, Actions} = ?HANDLE_EFFECTS(RpcEffs ++ Effects ++ [{aux, tick}],
cast, State1#state{server_state = ServerState,
commit_rate = CommitRate}),
%% try sending any pending applied notifications again
State = send_applied_notifications(#{}, State2),
{keep_state, State,
set_tick_timer(State, Actions)};
leader({timeout, Name}, machine_timeout, State0) ->
% the machine timer timed out, add a timeout message
Cmd = make_command('$usr', cast, {timeout, Name}, noreply),
{leader, State1, Effects} = handle_leader({command, Cmd}, State0),
{State, Actions} = ?HANDLE_EFFECTS(Effects, cast, State1),
{keep_state, State, Actions};
leader({timeout, {snapshot_retry, PeerId}}, {snapshot_retry, PeerId},
#state{server_state = SS0} = State0) ->
%% Reset peer status from backoff to normal and trigger RPC evaluation
case ra_server:handle_leader({snapshot_retry_timeout, PeerId}, SS0) of
{leader, SS, Effects} ->
{State, Actions} = ?HANDLE_EFFECTS(Effects, cast,
State0#state{server_state = SS}),
{keep_state, State, Actions}
end;
leader({call, From}, trigger_election, State) ->
{keep_state, State, [{reply, From, ok}]};
leader({call, From}, {read_entries, Indexes}, State) ->
read_entries0(From, Indexes, State);
leader(EventType, Msg, State0) ->
case handle_leader(Msg, State0) of
{leader, State1, Effects1} ->
{State, Actions} = ?HANDLE_EFFECTS(Effects1, EventType, State1),
{keep_state, State, Actions};
{stop, State1, Effects} ->
% interact before shutting down in case followers need
% to know about the new commit index
{State, _Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
{stop, normal, State};
{delete_and_terminate, State1, Effects} ->
{State2, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
State = send_rpcs(State2),
case ra_server:is_fully_replicated(State#state.server_state) of
true ->
{stop, {shutdown, delete}, State};
false ->
next_state(terminating_leader, State, Actions)
end;
{await_condition, State1, Effects1} ->
{State, Actions} = ?HANDLE_EFFECTS(Effects1, EventType, State1),
?DEBUG_IF(is_command(Msg), "~ts: postponing ~0P",
[log_id(State0), Msg, 10]),
next_state(await_condition, State,
[{postpone, is_command(Msg)} | Actions]);
{NextState, State1, Effects1} ->
{State, Actions} = ?HANDLE_EFFECTS(Effects1, EventType, State1),
next_state(NextState, State, Actions)
end.
candidate(enter, OldState, State0) ->
{State1, Actions0} = handle_enter(?FUNCTION_NAME, OldState, State0),
{State, Actions} = maybe_set_election_timeout(short, State1, Actions0),
{keep_state, State, Actions};
candidate({call, From}, {leader_call, Msg},
#state{pending_commands = Pending} = State) ->
{keep_state, State#state{pending_commands = [{From, Msg} | Pending]}};
candidate(EventType, {local_call, Msg}, State) ->
candidate(EventType, Msg, State);
candidate(cast, {command, _Priority,
{_CmdType, _Data, {notify, Corr, Pid}}},
State) ->
_ = reject_command(Pid, Corr, State),
{keep_state, State, []};
candidate({call, _From} = EventType, {local_query, QueryFun}, State) ->
candidate(EventType, {local_query, QueryFun, #{}}, State);
candidate({call, From} = EventType, {local_query, QueryFun, Options}, State) ->
evaluate_local_query(candidate, EventType, From, QueryFun, Options, State);
candidate({call, From}, {state_query, Spec}, State) ->
Reply = {ok, do_state_query(Spec, State), id(State)},
{keep_state, State, [{reply, From, Reply}]};
candidate(EventType, {aux_command, Cmd}, State0) ->
{_, ServerState, Effects} = ra_server:handle_aux(?FUNCTION_NAME, EventType, Cmd,
State0#state.server_state),
{State, Actions} =
?HANDLE_EFFECTS(Effects, EventType,
State0#state{server_state = ServerState}),
{keep_state, State, Actions};
candidate({call, From}, ping, State) ->
{keep_state, State, [{reply, From, {pong, candidate}}]};
candidate(info, {'DOWN', _MRef, process, Pid, Info}, State0) ->
handle_process_down(Pid, Info, ?FUNCTION_NAME, State0);
candidate(info, {Status, Node, InfoList}, State0)
when Status =:= nodedown orelse Status =:= nodeup ->
handle_node_status_change(Node, Status, InfoList, ?FUNCTION_NAME, State0);
candidate(info, {node_event, _Node, _Evt}, State) ->
{keep_state, State};
candidate(_, tick_timeout,
#state{conf = #conf{tick_timeout = TickTimeMs},
commit_rate = CommitRate0,
server_state = ServerState} = State0) ->
CommitRate = update_commit_rate(CommitRate0, ServerState, TickTimeMs),
State = maybe_persist_last_applied(State0#state{commit_rate = CommitRate}),
{keep_state, State, set_tick_timer(State, [])};
candidate({call, From}, trigger_election, State) ->
{keep_state, State, [{reply, From, ok}]};
candidate({call, From}, {read_entries, Indexes}, State) ->
read_entries0(From, Indexes, State);
candidate(EventType, Msg, State0) ->
case handle_candidate(Msg, State0) of
{candidate, State1, Effects} ->
{State2, Actions0} = ?HANDLE_EFFECTS(Effects, EventType, State1),
{State, Actions} = maybe_set_election_timeout(medium, State2, Actions0),
{keep_state, State, Actions};
{follower, State1, Effects} ->
{State, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
next_state(follower, State, Actions);
{leader, State1, Effects} ->
{State, Actions0} = ?HANDLE_EFFECTS(Effects, EventType, State1),
%% reset the tick timer to avoid it triggering early after a leader
%% change
next_state(leader, State,
[{next_event, cast, tick_timeout} | Actions0])
end.
pre_vote(enter, OldState, #state{leader_monitor = MRef} = State0) ->
_ = stop_monitor(MRef),
{State1, Actions0} = handle_enter(?FUNCTION_NAME, OldState, State0),
{State, Actions} = maybe_set_election_timeout(long, State1, Actions0),
{keep_state, State#state{leader_monitor = undefined}, Actions};
pre_vote({call, From}, {leader_call, Msg},
State = #state{pending_commands = Pending}) ->
{keep_state, State#state{pending_commands = [{From, Msg} | Pending]}};
pre_vote(EventType, {local_call, Msg}, State) ->
pre_vote(EventType, Msg, State);
pre_vote(cast, {command, _Priority,
{_CmdType, _Data, {notify, Corr, Pid}}},
State) ->
_ = reject_command(Pid, Corr, State),
{keep_state, State, []};
pre_vote({call, _From} = EventType, {local_query, QueryFun}, State) ->
pre_vote(EventType, {local_query, QueryFun, #{}}, State);
pre_vote({call, From} = EventType, {local_query, QueryFun, Options}, State) ->
evaluate_local_query(pre_vote, EventType, From, QueryFun, Options, State);
pre_vote({call, From}, {state_query, Spec}, State) ->
Reply = {ok, do_state_query(Spec, State), id(State)},
{keep_state, State, [{reply, From, Reply}]};
pre_vote(EventType, {aux_command, Cmd}, State0) ->
{_, ServerState, Effects} = ra_server:handle_aux(?FUNCTION_NAME, EventType, Cmd,
State0#state.server_state),
{State, Actions} =
?HANDLE_EFFECTS(Effects, EventType,
State0#state{server_state = ServerState}),
{keep_state, State, Actions};
pre_vote({call, From}, ping, State) ->
{keep_state, State, [{reply, From, {pong, pre_vote}}]};
pre_vote(info, {node_event, _Node, _Evt}, State) ->
{keep_state, State};
pre_vote(info, {Status, Node, InfoList}, State0)
when Status =:= nodedown orelse Status =:= nodeup ->
handle_node_status_change(Node, Status, InfoList, ?FUNCTION_NAME, State0);
pre_vote(info, {'DOWN', _MRef, process, Pid, Info}, State0) ->
handle_process_down(Pid, Info, ?FUNCTION_NAME, State0);
pre_vote(_, tick_timeout,
#state{conf = #conf{tick_timeout = TickTimeMs},
commit_rate = CommitRate0,
server_state = ServerState} = State0) ->
CommitRate = update_commit_rate(CommitRate0, ServerState, TickTimeMs),
State = maybe_persist_last_applied(State0#state{commit_rate = CommitRate}),
{keep_state, State, set_tick_timer(State, [])};
pre_vote({call, From}, trigger_election, State) ->
{keep_state, State, [{reply, From, ok}]};
pre_vote({call, From}, {read_entries, Indexes}, State) ->
read_entries0(From, Indexes, State);
pre_vote(EventType, Msg, State0) ->
case handle_pre_vote(Msg, State0) of
{pre_vote, State1, Effects} ->
{State2, Actions0} = ?HANDLE_EFFECTS(Effects, EventType, State1),
{State, Actions} = maybe_set_election_timeout(long, State2, Actions0),
{keep_state, State, Actions};
{follower, State1, Effects} ->
{State, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
next_state(follower, State, Actions);
{candidate, State1, Effects} ->
{State, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
next_state(candidate, State, Actions)
end.
follower(enter, OldState, #state{low_priority_commands = Delayed,
server_state = ServerState} = State0) ->
%% New cluster starts should be coordinated and elections triggered
%% explicitly hence if this is a new one we wait here.
%% Else we set an election timer
%% Set the timeout length based on the previous state
TimeoutLen = case OldState of
recovered -> short;
_ -> long
end,
{State1, Actions0} = handle_enter(?FUNCTION_NAME, OldState, State0),
{State, Actions} = case ra_server:is_new(ServerState) of
true ->
{State1, Actions0};
false ->
?DEBUG("~ts: is not new, setting "
"election timeout.",
[log_id(State0)]),
maybe_set_election_timeout(TimeoutLen, State1,
Actions0)
end,
Monitors = ra_monitors:remove_all(machine, State#state.monitors),
{keep_state, State#state{low_priority_commands = ra_ets_queue:reset(Delayed),
monitors = Monitors}, Actions};
follower({call, From}, {leader_call, Msg}, State) ->
maybe_redirect(From, Msg, State);
follower(EventType, {local_call, Msg}, State) ->
follower(EventType, Msg, State);
follower(_, {command, Priority, {_CmdType, Data, noreply}},
State) ->
% forward to leader
case leader_id(State) of
undefined ->
?WARN("~ts: leader cast - leader not known. "
"Command is dropped.", [log_id(State)]),
{keep_state, State, []};
LeaderId ->
?DEBUG("~ts: follower leader cast - redirecting to ~tw ",
[log_id(State), LeaderId]),
ok = ra:pipeline_command(LeaderId, Data, no_correlation, Priority),
{keep_state, State, []}
end;
follower(cast, {command, _Priority,
{_CmdType, _Data, {notify, Corr, Pid}}},
State) ->
_ = reject_command(Pid, Corr, State),
{keep_state, State, []};
follower({call, _From} = EventType, {local_query, QueryFun}, State) ->
follower(EventType, {local_query, QueryFun, #{}}, State);
follower({call, From} = EventType, {local_query, QueryFun, Options}, State) ->
evaluate_local_query(follower, EventType, From, QueryFun, Options, State);
follower({call, From}, {state_query, Spec}, State) ->
Reply = {ok, do_state_query(Spec, State), id(State)},
{keep_state, State, [{reply, From, Reply}]};
follower(EventType, {aux_command, Cmd}, State0) ->
{_, ServerState, Effects} = ra_server:handle_aux(?FUNCTION_NAME, EventType, Cmd,
State0#state.server_state),
{State, Actions} =
?HANDLE_EFFECTS(Effects, EventType,
State0#state{server_state = ServerState}),
{keep_state, State, Actions};
follower({call, From}, trigger_election, State) ->
?DEBUG("~ts: election triggered by ~w", [log_id(State), element(1, From)]),
{keep_state, State, [{reply, From, ok},
{next_event, cast, election_timeout}]};
follower({call, From}, ping, State) ->
{keep_state, State, [{reply, From, {pong, follower}}]};
follower(info, {'DOWN', MRef, process, _Pid, Info},
#state{leader_monitor = MRef} = State0) ->
?INFO("~ts: Leader monitor down with ~W, setting election timeout",
[log_id(State0), Info, 8]),
%% If the DOWN reason is something else than `noconnection', we know that
%% the leader process is really gone. We want to clear the leader ID we
%% know at this point, while a new election is running.
%%
%% This is to make sure that `follower_leader_change()' won't create a
%% useless monitor and more importantly, it won't redirect pending
%% commands to that old leader. This would cause the commands callers to
%% get a `noproc' error or a timeout.
State1 = case Info of
noconnection ->
State0;
_ ->
clear_leader_id(State0)
end,
TimeoutLen = case Info of
noconnection ->
short;
_ ->
%% if it isn't a network related error
%% set the shortest timeout
really_short
end,
{State, Actions} = maybe_set_election_timeout(TimeoutLen, State1, []),
{keep_state, State#state{leader_monitor = undefined}, Actions};
follower(info, {'DOWN', _MRef, process, Pid, Info}, State0) ->
handle_process_down(Pid, Info, ?FUNCTION_NAME, State0);
follower(info, {node_event, Node, down}, State0) ->
case leader_id(State0) of
{_, Node} ->
?DEBUG("~ts: Leader node ~w may be down, setting pre-vote timeout",
[log_id(State0), Node]),
{State, Actions} = maybe_set_election_timeout(long, State0, []),
{keep_state, State, Actions};
_ ->
{keep_state, State0, []}
end;
follower(info, {node_event, Node, up}, State) ->
case leader_id(State) of
{_, Node} when State#state.election_timeout_set ->
?DEBUG("~ts: Leader node ~w is back up, cancelling pre-vote timeout",
[log_id(State), Node]),
{keep_state,
State#state{election_timeout_set = false},
[{state_timeout, infinity, election_timeout}]};
_ ->
{keep_state, State, []}
end;
follower(info, {Status, Node, InfoList}, State0)
when Status =:= nodedown orelse Status =:= nodeup ->
handle_node_status_change(Node, Status, InfoList, ?FUNCTION_NAME, State0);
follower(_, tick_timeout,
#state{conf = #conf{tick_timeout = TickTimeMs},
commit_rate = CommitRate0,
server_state = ServerState0} = State0) ->
CommitRate = update_commit_rate(CommitRate0, ServerState0, TickTimeMs),
ServerState = ra_server:log_tick(ServerState0),
{State, Actions} = ?HANDLE_EFFECTS([{aux, tick}], cast,
State0#state{server_state = ServerState,
commit_rate = CommitRate}),
{keep_state, State, set_tick_timer(State, Actions)};
follower({call, From}, {read_entries, Indexes}, State) ->
read_entries0(From, Indexes, State);
follower(EventType, Msg, #state{conf = #conf{name = Name},
server_state = SS0} = State0) ->
case handle_follower(Msg, State0) of
{follower, State1, Effects} ->
{State2, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
State = #state{server_state = SS} = follower_leader_change(State0, State2),
Membership0 = ra_server:get_membership(SS0),
case ra_server:get_membership(SS) of
Membership0 ->
ok;
Membership ->
true = ets:update_element(ra_state, Name, {3, Membership})
end,
{keep_state, State, Actions};
{pre_vote, State1, Effects} ->
{State, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
next_state(pre_vote, State, Actions);
{await_condition, State1, Effects} ->
{State2, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
State = follower_leader_change(State0, State2),
next_state(await_condition, State, Actions);
{receive_snapshot, State1, Effects} ->
{State, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
next_state(receive_snapshot, State, Actions);
{delete_and_terminate, State1, Effects} ->
{State, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
next_state(terminating_follower, State, Actions)
end.
%% TODO: handle leader down abort snapshot and revert to follower
receive_snapshot(enter, OldState, State0 = #state{conf = Conf}) ->
#conf{receive_snapshot_timeout = ReceiveSnapshotTimeout} = Conf,
{State, Actions} = handle_enter(?FUNCTION_NAME, OldState, State0),
{keep_state, State,
[{state_timeout, ReceiveSnapshotTimeout, receive_snapshot_timeout}
| Actions]};
receive_snapshot(_, tick_timeout,
#state{commit_rate = CommitRate0,
server_state = ServerState} = State0) ->
%% Reset the integrator to avoid counting snapshot index jump as throughput
CommitRate = reset_commit_rate(CommitRate0, ServerState),
{keep_state, State0#state{commit_rate = CommitRate}, set_tick_timer(State0, [])};
receive_snapshot({call, From}, {leader_call, Msg}, State) ->
maybe_redirect(From, Msg, State);
receive_snapshot(EventType, {local_call, Msg}, State) ->
receive_snapshot(EventType, Msg, State);
receive_snapshot({call, _From} = EventType, {local_query, QueryFun}, State) ->
receive_snapshot(EventType, {local_query, QueryFun, #{}}, State);
receive_snapshot({call, From} = EventType, {local_query, QueryFun, Options},
State) ->
evaluate_local_query(receive_snapshot, EventType, From, QueryFun,
Options, State);
receive_snapshot({call, From}, {state_query, Spec}, State) ->
Reply = {ok, do_state_query(Spec, State), id(State)},
{keep_state, State, [{reply, From, Reply}]};
receive_snapshot(EventType, {aux_command, Cmd}, State0) ->
{_, ServerState, Effects} = ra_server:handle_aux(?FUNCTION_NAME, EventType, Cmd,
State0#state.server_state),
{State, Actions} =
?HANDLE_EFFECTS(Effects, EventType,
State0#state{server_state = ServerState}),
{keep_state, State, Actions};
receive_snapshot(info, {'DOWN', MRef, process, _Pid, _Info},
#state{leader_monitor = MRef} = State) ->
%% leader is down
?INFO("~ts: receive_snapshot - Leader monitor down. Aborting snapshot receive. "
"Entering follower state.",
[log_id(State)]),
receive_snapshot(info, receive_snapshot_timeout,
State#state{leader_monitor = undefined});
receive_snapshot(EventType, Msg, State00) ->
%% HACK: if the EventType is a {call, From} we stash the from in the
%% ServerState, this is so ra_server can defer the snapshot reply in
%% some cases
State0 = case EventType of
{call, _From} ->
SS = State00#state.server_state,
State00#state{server_state =
SS#{current_event_type => EventType}};
_ ->
State00
end,
case handle_receive_snapshot(Msg, State0) of
{receive_snapshot, State1, Effects} ->
{#state{conf = Conf} = State, Actions} =
?HANDLE_EFFECTS(Effects, EventType, State1),
TimeoutActions = case Msg of
#install_snapshot_rpc{} ->
%% Reset timeout only on receive snapshot progress.
[{state_timeout,
Conf#conf.receive_snapshot_timeout,
receive_snapshot_timeout}];
_ ->
[]
end,
{keep_state, State, TimeoutActions ++ Actions};
{follower, State1, Effects} ->
{State2, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
State = follower_leader_change(State0, State2),
next_state(follower, State, Actions)
end.
terminating_leader(enter, OldState, State0) ->
{State, Actions} = handle_enter(?FUNCTION_NAME, OldState, State0),
{keep_state, State, Actions};
terminating_leader(_EvtType, {command, _, _}, State0) ->
% do not process any further commands
{keep_state, State0, []};
terminating_leader(EvtType, Msg, State0) ->
LogName = log_id(State0),
?DEBUG("~ts: terminating leader received ~W", [LogName, Msg, 10]),
{State, Actions} = case leader(EvtType, Msg, State0) of
{next_state, terminating_leader, S, A} ->
{S, A};
{keep_state, S, A} ->
{S, A};
{stop, shutdown, S} ->
{S, []};
{stop, {shutdown, delete}, S} ->
{S, []}
end,
NS = State#state.server_state,
case ra_server:is_fully_replicated(NS) of
true ->
{stop, {shutdown, delete}, State};
false ->
?DEBUG("~ts: is not fully replicated after ~W",
[LogName, Msg, 7]),
{keep_state, send_rpcs(State), Actions}
end.
terminating_follower(enter, OldState, State0) ->
{State, Actions} = handle_enter(?FUNCTION_NAME, OldState, State0),
{keep_state, State, Actions};
terminating_follower(EvtType, Msg, State0) ->
% only process ra_log_events
LogName = log_id(State0),
?DEBUG("~ts: terminating follower received ~W", [LogName, Msg, 10]),
{State, Actions} = case follower(EvtType, Msg, State0) of
{next_state, terminating_follower, S, A} ->
{S, A};
{next_state, NextState, S, A} ->
?DEBUG("~ts: terminating follower requested state '~s'"
" - remaining in current state",
[LogName, NextState]),
{S, A};
{keep_state, S, A} ->
{S, A}
end,
case ra_server:is_fully_persisted(State#state.server_state) of
true ->
{stop, {shutdown, delete}, State};
false ->
?DEBUG("~ts: is not fully persisted after ~W",
[log_id(State), Msg, 7]),
{keep_state, State, Actions}
end.
await_condition(enter, OldState, #state{conf = Conf,
server_state = ServerState} = State0) ->
{State, Actions0} = handle_enter(?FUNCTION_NAME, OldState, State0),
Timeout = ra_server:get_condition_timeout(ServerState,
Conf#conf.await_condition_timeout),
Actions = [{state_timeout, Timeout, await_condition_timeout} | Actions0],
{keep_state, State, Actions};
await_condition({call, From}, {leader_call, Msg}, State) ->
maybe_redirect(From, Msg, State);
await_condition(EventType, {local_call, Msg}, State) ->
await_condition(EventType, Msg, State);
await_condition({call, _From} = EventType, {local_query, QueryFun}, State) ->
await_condition(EventType, {local_query, QueryFun, #{}}, State);
await_condition({call, From} = EventType, {local_query, QueryFun, Options},
State) ->
evaluate_local_query(await_condition, EventType, From, QueryFun,
Options, State);
await_condition({call, From}, {state_query, Spec}, State) ->
Reply = {ok, do_state_query(Spec, State), id(State)},
{keep_state, State, [{reply, From, Reply}]};
await_condition(EventType, {aux_command, Cmd}, State0) ->
{_, ServerState, Effects} = ra_server:handle_aux(?FUNCTION_NAME, EventType,
Cmd, State0#state.server_state),
{State, Actions} =
?HANDLE_EFFECTS(Effects, EventType,
State0#state{server_state = ServerState}),
{keep_state, State, Actions};
await_condition({call, From}, ping, State) ->
{keep_state, State, [{reply, From, {pong, await_condition}}]};
await_condition({call, From}, trigger_election, State) ->
{keep_state, State, [{reply, From, ok},
{next_event, cast, election_timeout}]};
await_condition(info, {'DOWN', MRef, process, _Pid, _Info},
State = #state{leader_monitor = MRef}) ->
?INFO("~ts: await_condition - Leader monitor down. Entering follower state.",
[log_id(State)]),
next_state(follower, State#state{leader_monitor = undefined}, []);
await_condition(info, {'DOWN', _MRef, process, Pid, Info}, State0) ->
handle_process_down(Pid, Info, ?FUNCTION_NAME, State0);
await_condition(info, {node_event, Node, down}, State) ->
case leader_id(State) of
{_, Node} ->
?WARN("~ts: await_condition - Leader node ~w might be down."
" Re-entering follower state.",
[log_id(State), Node]),
next_state(follower, State, []);
_ ->
{keep_state, State}
end;
await_condition(info, {Status, Node, InfoList}, State0)
when Status =:= nodedown orelse Status =:= nodeup ->
handle_node_status_change(Node, Status, InfoList, ?FUNCTION_NAME, State0);
await_condition(_, tick_timeout,
#state{conf = #conf{tick_timeout = TickTimeMs},
commit_rate = CommitRate0,
server_state = ServerState} = State0) ->
CommitRate = update_commit_rate(CommitRate0, ServerState, TickTimeMs),
{State, Actions} = ?HANDLE_EFFECTS([{aux, tick}], cast,
State0#state{commit_rate = CommitRate}),
{keep_state, State, set_tick_timer(State, Actions)};
await_condition({timeout, {snapshot_retry, _}}, {snapshot_retry, _}, _State) ->
%% Postpone snapshot retry until we transition back to leader
{keep_state_and_data, [postpone]};
await_condition({call, From}, {read_entries, Indexes}, State) ->
read_entries0(From, Indexes, State);
await_condition(EventType, Msg, State0) ->
case handle_await_condition(Msg, State0) of
{follower, State1, Effects} ->
{State2, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
State = follower_leader_change(State0, State2),
next_state(follower, State, Actions);
{pre_vote, State1, Effects} ->
{State, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
next_state(pre_vote, State, Actions);
{leader, State1, Effects} ->
{State, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
next_state(leader, State, Actions);
{await_condition, State1, Effects} ->
{State, Actions} = ?HANDLE_EFFECTS(Effects, EventType, State1),
%% postpone commands such that they are retried when the
%% await_condition state is exited. Should help with client
%% liveness
?DEBUG_IF(is_command(Msg), "~ts: await_condition postponing ~0P",
[log_id(State0), Msg, 10]),
{keep_state, State, [{postpone, is_command(Msg)} | Actions]}
end.
is_command(Msg) when is_tuple(Msg) ->
element(1, Msg) == command;
is_command(_) ->
false.
handle_event(_EventType, EventContent, StateName, State) ->
?WARN("~ts: handle_event unknown ~P", [log_id(State), EventContent, 10]),
{next_state, StateName, State}.
terminate(Reason, StateName,
#state{conf = #conf{name = Key,
cluster_name = ClusterName,
worker_pid = WorkerPid},
server_state = ServerState} = State) ->
?DEBUG("~ts: terminating with ~w in state ~w",
[log_id(State), Reason, StateName]),
#{names := #{server_sup := SrvSup,
log_meta := MetaName} = Names} =
ra_server:system_config(ServerState),
UId = uid(State),
Id = id(State),
case Reason of
{shutdown, delete} ->
%% try to wait a bit for the worker to finish any pending work
%% before continuing
if is_pid(WorkerPid) ->
_ = catch gen_server:call(WorkerPid, undefined, 1000);
true -> ok
end,
Parent = ra_directory:where_is_parent(Names, UId),
%% we need to unregister _before_ the log closes
%% in the ra_server:terminate/2 function
%% as we want the directory to be deleted
%% after the server is removed from the ra directory.
%% This is so that the segment writer can avoid
%% crashing if it detects a missing key
catch ra_directory:unregister_name(Names, UId),
catch ra_log_wal:forget_writer(maps:get(wal, Names), UId),
_ = ra_server:terminate(ServerState, Reason),
catch ra_log_meta:delete_sync(MetaName, UId),
catch ra_counters:delete(Id),
Self = self(),
%% we have to terminate the child spec from the supervisor as it
%% won't do this automatically, even for transient children
%% for simple_one_for_one terminate also removes
_ = spawn(fun () ->
Ref = erlang:monitor(process, Self),
receive
{'DOWN', Ref, _, _, _} ->
ok = supervisor:terminate_child(
SrvSup, Parent)
after 5000 ->
ok
end
end),
ok;
_ ->
_ = ra_server:terminate(ServerState, Reason),
ok
end,
catch ra_leaderboard:clear(ClusterName),
_ = ets:delete(ra_state, Key),
ok;
%% This occurs if there is a crash in the init callback of the ra_machine,
%% before a state has been built
terminate(Reason, StateName, #{id := Id} = Config) ->
LogId = maps:get(friendly_name, Config,
lists:flatten(io_lib:format("~tw", [Id]))),
?DEBUG("~ts: terminating with ~w in state ~w",
[LogId, Reason, StateName]),
ok;
%% Unknown reason for termination
terminate(Reason, StateName, State) ->
?DEBUG("Terminating with ~w in state ~w with state ~w",
[Reason, StateName, State]),
ok.
code_change(_OldVsn, StateName, State, _Extra) ->
{ok, StateName, State}.
format_status(#{state := StateName,
data := #state{server_state = NS,
leader_last_seen = LastSeen,
pending_commands = Pending,
low_priority_commands = Delayed,
pending_notifys = PendingNots,
election_timeout_set = ElectionSet
}} = FormatStatus) ->
NumPendingNots = maps:fold(fun (_, Corrs, Acc) -> Acc + length(Corrs) end,
0, PendingNots),
FormatStatus#{data => [{id, ra_server:id(NS)},
{raft_state, StateName},
{leader_last_seen, LastSeen},
{num_pending_commands, length(Pending)},
{num_low_priority_commands, ra_ets_queue:len(Delayed)},
{num_pending_applied_notifications, NumPendingNots},
{election_timeout_set, ElectionSet},
{ra_server_state, ra_server:overview(NS)}
]}.
%%%===================================================================
%%% Internal functions
%%%===================================================================
handle_enter(RaftState, OldRaftState,
#state{conf = #conf{name = Name},
server_state = ServerState0} = State) ->
Membership = ra_server:get_membership(ServerState0),
true = ets:insert(ra_state, {Name, RaftState, Membership}),
{ServerState, Effects} = ra_server:handle_state_enter(RaftState,
OldRaftState,
ServerState0),
LastApplied = do_state_query(last_applied, State),
case RaftState == leader orelse OldRaftState == leader of
true ->
%% ensure transitions from and to leader are logged at a higher
%% level
?NOTICE("~ts: ~s -> ~s in term: ~b machine version: ~b, "
"last applied ~b",
[log_id(State), OldRaftState, RaftState,
current_term(State), machine_version(State),
LastApplied]);
false ->
?DEBUG("~ts: ~s -> ~s in term: ~b machine version: ~b, "
"last applied ~b",
[log_id(State), OldRaftState, RaftState,
current_term(State), machine_version(State),
LastApplied])
end,
handle_effects(RaftState, Effects, cast,
State#state{server_state = ServerState}).
handle_leader(Msg, #state{server_state = ServerState0} = State0) ->
try ra_server:handle_leader(Msg, ServerState0) of
{NextState, ServerState, Effects} ->
State1 = State0#state{server_state =
ra_server:persist_last_applied(ServerState)},
%% The last applied index made progress. Check if there are
%% pending queries that wait for this index.
{State, Actions} = perform_pending_queries(leader, State1),
maybe_record_cluster_change(State0, State),
{NextState, State, Effects ++ Actions}
catch
throw:{NextState, ServerState, Effects} when is_atom(NextState) andalso
is_map(ServerState) andalso
is_list(Effects) ->
State1 = State0#state{server_state =
ra_server:persist_last_applied(ServerState)},
{State, Actions} = perform_pending_queries(leader, State1),
maybe_record_cluster_change(State0, State),
{NextState, State, Effects ++ Actions};
Class:Reason:Stacktrace ->
?ERR("~ts: handle_leader err ~p ~0P~n~p",
[log_id(State0), Class, Reason, 10,
safe_stacktrace(Stacktrace)]),
exit(Reason)
end.
handle_raft_state(RaftState, Msg,
#state{server_state = ServerState0,
election_timeout_set = Set} = State0) ->
{NextState, ServerState1, Effects} =
ra_server:RaftState(Msg, ServerState0),
ElectionTimeoutSet = case Msg of
election_timeout -> false;
_ -> Set
end,
ServerState = ra_server:persist_last_applied(ServerState1),
State1 = State0#state{server_state = ServerState,
election_timeout_set = ElectionTimeoutSet},
%% The last applied index made progress. Check if there are pending
%% queries that wait for this index.
{State, Actions} = perform_pending_queries(RaftState, State1),
{NextState, State, Effects ++ Actions}.
handle_candidate(Msg, State) ->
handle_raft_state(?FUNCTION_NAME, Msg, State).
handle_pre_vote(Msg, State) ->
handle_raft_state(?FUNCTION_NAME, Msg, State).
handle_follower(Msg, State0) ->
Ret = handle_raft_state(?FUNCTION_NAME, Msg, State0),
{_NextState, State, _Effects} = Ret,
maybe_record_cluster_change(State0, State),
Ret.
handle_receive_snapshot(Msg, State) ->
handle_raft_state(?FUNCTION_NAME, Msg, State).
handle_await_condition(Msg, State) ->
handle_raft_state(?FUNCTION_NAME, Msg, State).
evaluate_local_query(RaftState, EventType, From, QueryFun, Options,
#state{conf = Conf,
server_state =
#{cfg := #cfg{id = ThisMember}} = ServerState,
pending_queries = PendingQueries} = State0) ->
%% The caller might decide it wants the query to be executed only after a
%% specific index has been applied on the local node. It can specify that
%% with the `condition' option.
%%
%% If the condition is unset or set to `undefined', the query is performed
%% immediatly. That is the default behavior.
%%
%% If the condition is set to `{applied, {Index, Term}}', the query is
%% added to a list of pending queries. It will be evaluated once that
%% index is applied locally.
{State3, Effects} =
case maps:get(condition, Options, undefined) of
undefined ->
Leader = determine_leader(RaftState, State0),
Reply = perform_local_query(QueryFun, Leader,
ServerState, Conf),
{State0, [{reply, From, Reply, {member, ThisMember}}]};
Condition ->
PendingQuery = {Condition, From, QueryFun},
PendingQueries1 = [PendingQuery | PendingQueries],
State1 = State0#state{pending_queries = PendingQueries1},
%% It's possible that the specified index was already applied.
%% That's why we evaluate pending queries just after adding the
%% query to the list.
{State2, Effects0} = perform_pending_queries(RaftState, State1),
{State2, Effects0}
end,
{State, Actions} = handle_effects(RaftState, Effects, EventType, State3),
{keep_state, State, Actions}.
perform_pending_queries(_RaftState, #state{pending_queries = []} = State) ->
{State, []};
perform_pending_queries(RaftState, State) ->
LastApplied = do_state_query(last_applied, State),
perform_pending_queries(RaftState, LastApplied, State, []).
perform_pending_queries(RaftState, LastApplied,
#state{conf = Conf,
server_state = ServerState0,
pending_queries = PendingQueries0} = State0,
Actions0) ->
Leader = determine_leader(RaftState, State0),
{PendingQueries, Actions, ServerState} =
lists:foldr(
fun (PendingQuery, Acc) ->
perform_pending_queries1(PendingQuery, Acc,
#{last_applied => LastApplied,
leader => Leader,
conf => Conf})
end, {[], Actions0, ServerState0}, PendingQueries0),
State = State0#state{server_state = ServerState,
pending_queries = PendingQueries},
{State, Actions}.
perform_pending_queries1({{applied, {TargetIndex, TargetTerm}}, From, QueryFun}
= PendingQuery,
{PendingQueries0, Actions0, #{cfg := #cfg{id = ThisMember}} = ServerState0},
#{last_applied := LastApplied,
leader := Leader,
conf := Conf})
when TargetIndex =< LastApplied ->
%% NB: this could mean a disk seek + reed
{Term, ServerState} = ra_server:fetch_term(TargetIndex, ServerState0),
case Term of
TargetTerm ->
%% The local node reached or passed the target index. We can
%% evaluate the query.
%%
%% Note that some queries may have timed out from the caller's
%% point of view. We can't tell that here, so they are still
%% evaluated. The reply will be discarded by Erlang because the
%% process alias in `From' is inactive after the timeout.
Reply = perform_local_query(QueryFun, Leader, ServerState, Conf),
Actions = [{reply, From, Reply, {member, ThisMember}} | Actions0],
{PendingQueries0, Actions, ServerState};
_ ->
PendingQueries = [PendingQuery | PendingQueries0],
{PendingQueries, Actions0, ServerState}
end;
perform_pending_queries1(PendingQuery,
{PendingQueries0, Actions, ServerState}, _Context) ->
PendingQueries = [PendingQuery | PendingQueries0],
{PendingQueries, Actions, ServerState}.
determine_leader(RaftState, #state{server_state = ServerState} = State) ->
case RaftState of
leader ->
id(State);
follower ->
case ra_server:leader_id(ServerState) of
undefined -> not_known;
L -> L
end;
_ ->
not_known
end.
perform_local_query(QueryFun, Leader, ServerState, Conf) ->
incr_counter(Conf, ?C_RA_SRV_LOCAL_QUERIES, 1),
try ra_server:machine_query(QueryFun, ServerState) of
Result ->
{ok, Result, Leader}
catch
_:_ = Err ->
{error, Err}
end.
handle_effects(RaftState, Effects0, EvtType, State0) ->
handle_effects(RaftState, Effects0, EvtType, State0, []).
% effect handler: either executes an effect or builds up a list of
% gen_statem 'Actions' to be returned.
handle_effects(RaftState, Effects0, EvtType, State0, Actions0) ->
{State, Actions} = lists:foldl(
fun(Effects, {State, Actions}) when is_list(Effects) ->
handle_effects(RaftState, Effects, EvtType,
State, Actions);
(Effect, {State, Actions}) ->
handle_effect(RaftState, Effect, EvtType,
State, Actions)
end, {State0, Actions0}, Effects0),
{State, lists:reverse(Actions)}.
handle_effect(_RaftState, {send_rpc, To, Rpc}, _,
#state{conf = Conf,
server_state = SS} = State0, Actions) ->
% fully qualified use only so that we can mock it for testing
% TODO: review / refactor to remove the mod call here
PeerStatus = ra_server:peer_status(To, SS),
case ?MODULE:send_rpc(To, Rpc, State0) of
ok ->
{State0, Actions};
nosuspend when PeerStatus == normal ->
%% update peer status to suspended and spawn a process
%% to send the rpc without nosuspend so that it will block until
%% the data can get through
Self = self(),
_Pid = spawn(fun () ->
%% AFAIK none of the below code will throw and
%% exception so we should always end up setting
%% the peer status back to normal
ok = gen_statem:cast(To, Rpc),
incr_counter(Conf, ?C_RA_SRV_MSGS_SENT, 1),
Self ! {unsuspend_peer, To}
end),
% ?DEBUG("~ts: temporarily suspending peer ~tw due to full distribution buffer ~W",
% [log_id(State0), To, Rpc, 5]),
{update_peer(To, #{status => suspended}, State0), Actions};
nosuspend ->
{State0, Actions};
noconnect ->
%% for noconnects just allow it to pipeline and catch up later
{State0, Actions}
end;
handle_effect(_, {next_event, Evt}, EvtType, State, Actions) ->
{State, [{next_event, EvtType, Evt} | Actions]};
handle_effect(_, {next_event, _, _} = Next, _, State, Actions) ->
{State, [Next | Actions]};
handle_effect(leader, {send_msg, To, Msg}, _, State, Actions) ->
%% default is to send without any wrapping
ToNode = get_node(To),
?ASYNC_DIST(ToNode, _ = send(To, Msg, State#state.conf)),
{State, Actions};
handle_effect(RaftState, {send_msg, To, _Msg, Options} = Eff,
_, State, Actions) ->
ToNode = get_node(To),
case has_local_opt(Options) of
true ->
case can_execute_locally(RaftState, ToNode, State) of
true ->
?ASYNC_DIST(ToNode, send_msg(Eff, State));
false ->
ok
end;
false when RaftState == leader ->
%% the effect got here so we can execute
?ASYNC_DIST(ToNode, send_msg(Eff, State));
false ->
ok
end,
{State, Actions};
handle_effect(leader, {append, Cmd}, _EvtType, State, Actions) ->
Evt = {command, normal, {'$usr', Cmd, noreply}},
{State, [{next_event, cast, Evt} | Actions]};
handle_effect(leader, {append, Cmd, ReplyMode}, _EvtType, State, Actions) ->
Evt = {command, normal, {'$usr', Cmd, ReplyMode}},
{State, [{next_event, cast, Evt} | Actions]};
handle_effect(_RaftState, {try_append, Cmd, ReplyMode}, _EvtType, State, Actions) ->
%% this is a special mode to retain the backwards compatibility of
%% certain prior uses of {append, when it wasn't (accidentally)
%% limited to the leader
Evt = {command, normal, {'$usr', Cmd, ReplyMode}},
{State, [{next_event, cast, Evt} | Actions]};
handle_effect(RaftState, {LogOrLogExt, Idxs, Fun, {local, Node}}, EvtType,
State0, Actions)
when LogOrLogExt == log orelse
LogOrLogExt == log_ext ->
case can_execute_locally(RaftState, Node, State0) of
true ->
{Effects, State} = handle_log_effect(LogOrLogExt, Idxs, Fun,
State0),
handle_effects(RaftState, Effects, EvtType,
State, Actions);
false ->
{State0, Actions}
end;
handle_effect(leader, {LogOrLogExt, Idxs, Fun}, EvtType, State0, Actions)
when is_list(Idxs) andalso
(LogOrLogExt == log orelse
LogOrLogExt == log_ext) ->
%% Useful to implement a batch send of data obtained from the log.
%% 1) Retrieve all data from the list of indexes
{Effects, State} = handle_log_effect(LogOrLogExt, Idxs, Fun,
State0),
handle_effects(leader, Effects, EvtType,
State, Actions);
handle_effect(RaftState, {aux, Cmd}, EventType, State0, Actions0) ->
{_, ServerState, Effects} = ra_server:handle_aux(RaftState, cast, Cmd,
State0#state.server_state),
{State, Actions} =
handle_effects(RaftState, Effects, EventType,
State0#state{server_state = ServerState}),
{State, Actions0 ++ Actions};
handle_effect(leader, {notify, Nots}, _, #state{} = State0, Actions) ->
%% should only be done by leader
State = send_applied_notifications(Nots, State0),
{State, Actions};
handle_effect(_AnyState, {cast, To, Msg}, _, State, Actions) ->
%% TODO: handle send failure
_ = gen_cast(To, Msg, State),
{State, Actions};
handle_effect(RaftState, {reply, {Pid, _Tag} = From, Reply, Replier}, _,
State, Actions) ->
case Replier of
leader when RaftState == leader ->
ok = gen_statem:reply(From, Reply);
leader ->
ok;
local ->
case can_execute_locally(RaftState, node(Pid), State) of
true ->
ok = gen_statem:reply(From, Reply);
false ->
ok
end;
{member, Member} ->
case can_execute_on_member(RaftState, Member, State) of
true ->
ok = gen_statem:reply(From, Reply);
false ->
ok
end;
_ ->
ok
end,
{State, Actions};
handle_effect(leader, {reply, From, Reply}, _, State, Actions) ->
% reply directly, this is only done from the leader
% this is like reply/4 above with the Replier=leader
ok = gen_statem:reply(From, Reply),
{State, Actions};
handle_effect(_RaftState, {reply, Reply}, {call, From}, State, Actions) ->
% this is the reply effect for replying to the current call, any state
% can use this
ok = gen_statem:reply(From, Reply),
{State, Actions};
handle_effect(_RaftState, {reply, _From, _Reply}, _EvtType, State, Actions) ->
{State, Actions};
handle_effect(leader, {send_snapshot, {_, ToNode} = To, {SnapState, _Id, Term}}, _,
#state{server_state = SS0,
monitors = Monitors,
conf = #conf{snapshot_chunk_size = ChunkSize,
log_id = LogId,
install_snap_rpc_timeout = InstallSnapTimeout} = Conf}
= State,
Actions) ->
case lists:member(ToNode, [node() | nodes()]) of
true ->
%% node is connected
%% leader effect only
Machine = ra_server:machine(SS0),
%% assertion - status must be normal or snapshot_backoff
%% (for retries)
#{To := #{status := Status}} = ra_server:peers(SS0),
AttemptCount = case Status of
{snapshot_backoff, N} ->
N;
normal ->
0
end,
Id = ra_server:id(SS0),
Pid = spawn(fun () ->
send_snapshots(Id, Term, To,
ChunkSize, InstallSnapTimeout,
SnapState, Machine, LogId)
end),
ok = incr_counter(Conf, ?C_RA_SRV_SNAPSHOTS_SENT, 1),
%% update the peer state so that no pipelined entries are sent
%% during the snapshot sending phase
SS = ra_server:update_peer(To, #{status =>
{sending_snapshot, Pid,
AttemptCount}}, SS0),
{State#state{server_state = SS,
monitors = ra_monitors:add(Pid, snapshot_sender,
Monitors)},
Actions};
false ->
?DEBUG("~ts: not sending snapshot to ~tw as not connected to node ~s",
[LogId, To, ToNode]),
SS = ra_server:update_peer(To, #{status => disconnected}, SS0),
{State#state{server_state = SS}, Actions}
end;
handle_effect(_, {delete_snapshot, Dir, SnapshotRef}, _, State0, Actions) ->
%% delete snapshots in separate process
_ = spawn(fun() ->
ra_snapshot:delete(Dir, SnapshotRef)
end),
{State0, Actions};
handle_effect(_, {send_vote_requests, VoteRequests}, _, % EvtType
#state{conf = #conf{aten_poll_interval = Poll}} = State, Actions) ->
% transient election processes
%% set the timeout to the aten poll interval which is the approximate maximum
%% election timeout value
Timeout = {dirty_timeout, Poll},
Me = self(),
[begin
_ = spawn(fun () ->
Reply = gen_statem:call(ServerId, Request, Timeout),
ok = gen_statem:cast(Me, Reply)
end)
end || {ServerId, Request} <- VoteRequests],
{State, Actions};
handle_effect(RaftState, {release_cursor, Index, MacState}, EvtType,
State, Actions) ->
handle_effect(RaftState, {release_cursor, Index, MacState, #{}}, EvtType,
State, Actions);
handle_effect(RaftState, {release_cursor, Index, MacState, Opts}, EvtType,
#state{server_state = ServerState0} = State0, Actions0) ->
incr_counter(State0#state.conf, ?C_RA_SRV_RELEASE_CURSORS, 1),
{ServerState, Effects} = ra_server:update_release_cursor(Index, MacState,
Opts, ServerState0),
State1 = State0#state{server_state = ServerState},
handle_effects(RaftState, Effects, EvtType, State1, Actions0);
handle_effect(RaftState, {release_cursor, Index}, EvtType,
#state{server_state = ServerState0} = State0, Actions0) ->
incr_counter(State0#state.conf, ?C_RA_SRV_RELEASE_CURSORS, 1),
{ServerState, Effects} = ra_server:promote_checkpoint(Index, ServerState0),
State1 = State0#state{server_state = ServerState},
handle_effects(RaftState, Effects, EvtType, State1, Actions0);
handle_effect(RaftState, {checkpoint, Index, MacState}, EvtType,
#state{server_state = ServerState0} = State0, Actions0) ->
incr_counter(State0#state.conf, ?C_RA_SRV_CHECKPOINTS, 1),
{ServerState, Effects} = ra_server:checkpoint(Index, MacState,
ServerState0),
State1 = State0#state{server_state = ServerState},
handle_effects(RaftState, Effects, EvtType, State1, Actions0);
handle_effect(_, garbage_collection, _EvtType, State, Actions) ->
true = erlang:garbage_collect(),
incr_counter(State#state.conf, ?C_RA_SRV_GCS, 1),
{State, Actions};
handle_effect(leader, {monitor, _ProcOrNode, PidOrNode}, _,
#state{monitors = Monitors} = State, Actions0) ->
%% this effect type is only emitted by state machines and thus will
%% only be monitored from the leader
{State#state{monitors = ra_monitors:add(PidOrNode, machine, Monitors)},
Actions0};
handle_effect(leader, {monitor, _ProcOrNode, machine, PidOrNode}, _,
#state{monitors = Monitors} = State, Actions0) ->
{State#state{monitors = ra_monitors:add(PidOrNode, machine, Monitors)},
Actions0};
handle_effect(_RaftState, {monitor, _ProcOrNode, machine, _PidOrNode}, _,
#state{} = State, Actions0) ->
%% AFAIK: there is nothing emitting this effect type but we have to
%% guard against it being actioned on the follower anyway
{State, Actions0};
handle_effect(_, {monitor, _ProcOrNode, Component, PidOrNode}, _,
#state{monitors = Monitors} = State, Actions0) ->
{State#state{monitors = ra_monitors:add(PidOrNode, Component, Monitors)},
Actions0};
handle_effect(_, {demonitor, _ProcOrNode, PidOrNode}, _,
#state{monitors = Monitors0} = State, Actions) ->
Monitors = ra_monitors:remove(PidOrNode, machine, Monitors0),
{State#state{monitors = Monitors}, Actions};
handle_effect(_, {demonitor, _ProcOrNode, Component, PidOrNode}, _,
#state{monitors = Monitors0} = State, Actions) ->
Monitors = ra_monitors:remove(PidOrNode, Component, Monitors0),
{State#state{monitors = Monitors}, Actions};
handle_effect(leader, {timer, Name, T}, _, State, Actions) ->
{State, [{{timeout, Name}, T, machine_timeout} | Actions]};
handle_effect(leader, {timer, Name, T0, Option}, _, State,
Actions) ->
T = case Option of
{abs, true} when is_number(T0) ->
M = erlang:monotonic_time(millisecond),
max(M, T0 - erlang:time_offset(millisecond));
_ ->
T0
end,
%% timers assume erlang system time type of timeouts so we need
%% to convert the absolute case to the monitonic time range
{State, [{{timeout, Name}, T, machine_timeout, [Option]} | Actions]};
handle_effect(leader, {mod_call, Mod, Fun, Args}, _,
State, Actions) ->
%% TODO: catch and log failures or rely on calling function never crashing
_ = erlang:apply(Mod, Fun, Args),
{State, Actions};
handle_effect(_RaftState, start_election_timeout, _, State, Actions) ->
maybe_set_election_timeout(long, State, Actions);
handle_effect(RaftState, {start_snapshot_retry_timer, PeerId, Delay}, _,
State, Actions)
when RaftState == leader orelse
RaftState == await_condition ->
%% Use gen_statem generic timeout with named timer
{State, [{{timeout, {snapshot_retry, PeerId}}, Delay,
{snapshot_retry, PeerId}} | Actions]};
handle_effect(_RaftState, {cancel_snapshot_retry_timer, PeerId}, _,
State, Actions) ->
{State, [{{timeout, {snapshot_retry, PeerId}}, infinity, undefined}
| Actions]};
handle_effect(follower, {record_leader_msg, _LeaderId}, _, State0, Actions) ->
%% record last time leader seen
State = State0#state{leader_last_seen = erlang:system_time(millisecond),
election_timeout_set = false},
%% always cancel state timeout when a valid leader message has been
%% received just in case a timeout is currently active
%% the follower ra_server logic will emit this
{State, [{state_timeout, infinity, undefined} | Actions]};
handle_effect(_, {record_leader_msg, _LeaderId}, _, State0, Actions) ->
%% non follower states don't need to reset state timeout after an effect
{State0, Actions};
handle_effect(_, {bg_work, FunOrMfa, ErrFun}, _,
#state{conf = #conf{worker_pid = WorkerPid}} = State0, Actions) ->
%% non follower states don't need to reset state timeout after an effect
ra_worker:queue_work(WorkerPid, FunOrMfa, ErrFun),
{State0, Actions};
handle_effect(_, _, _, State0, Actions) ->
{State0, Actions}.
send_rpcs(State0) ->
{State, Rpcs} = make_rpcs(State0),
% module call so that we can mock
% We can ignore send failures here as they have not incremented
% the peer's next index
[_ = ?MODULE:send_rpc(To, Rpc, State) || {send_rpc, To, Rpc} <- Rpcs],
State.
make_rpcs(State) ->
{ServerState, Rpcs} = ra_server:make_rpcs(State#state.server_state),
{State#state{server_state = ServerState}, Rpcs}.
send_rpc(To, Msg, State) ->
incr_counter(State#state.conf, ?C_RA_SRV_RPCS_SENT, 1),
% fake gen cast
gen_cast(To, Msg, State).
gen_cast(To, Msg, State) ->
send(To, {'$gen_cast', Msg}, State#state.conf).
send_ra_event(To, Msg, FromId, EvtType, State) ->
send(To, wrap_ra_event(State, FromId, EvtType, Msg), State#state.conf).
wrap_ra_event(#state{conf = #conf{ra_event_formatter = undefined}},
FromId, EvtType, Evt) ->
{ra_event, FromId, {EvtType, Evt}};
wrap_ra_event(#state{conf = #conf{ra_event_formatter = {M, F, A}}},
LeaderId, EvtType, Evt) ->
apply(M, F, [LeaderId, {EvtType, Evt} | A]).
parse_send_msg_options(ra_event) ->
{true, false};
parse_send_msg_options(cast) ->
{false, true};
parse_send_msg_options(local) ->
{false, false};
parse_send_msg_options(Options) when is_list(Options) ->
{lists:member(ra_event, Options), lists:member(cast, Options)}.
id(#state{server_state = ServerState}) ->
ra_server:id(ServerState).
log_id(#state{conf = #conf{log_id = N}}) ->
N.
uid(#state{server_state = ServerState}) ->
ra_server:uid(ServerState).
leader_id(#state{server_state = ServerState}) ->
ra_server:leader_id(ServerState).
clear_leader_id(#state{server_state = ServerState} = State) ->
ServerState1 = ra_server:clear_leader_id(ServerState),
State#state{server_state = ServerState1}.
current_term(#state{server_state = ServerState}) ->
ra_server:current_term(ServerState).
machine_version(#state{server_state = ServerState}) ->
ra_server:machine_version(ServerState).
process_pending_queries(NewLeader,
#state{server_state = ServerState0} = State) ->
{ServerState, Froms} = ra_server:process_new_leader_queries(ServerState0),
[_ = gen_statem:reply(F, {redirect, NewLeader})
|| F <- Froms],
State#state{server_state = ServerState}.
election_timeout_action(Length, #state{conf = Conf}) ->
election_timeout_action(Length, Conf);
election_timeout_action(really_short, #conf{broadcast_time = Timeout}) ->
T = rand:uniform(Timeout),
{state_timeout, T, election_timeout};
election_timeout_action(short, #conf{broadcast_time = Timeout}) ->
T = rand:uniform(Timeout * ?DEFAULT_ELECTION_MULT) + Timeout,
{state_timeout, T, election_timeout};
election_timeout_action(medium, #conf{broadcast_time = Timeout,
aten_poll_interval = Poll}) ->
%% this is roughly in between broadcast time and aten poll interval
T = rand:uniform(Poll) + Timeout,
{state_timeout, T, election_timeout};
election_timeout_action(long, #conf{broadcast_time = Timeout,
aten_poll_interval = Poll}) ->
%% this should be longer than aten detection poll interval so that
%% there is a chance a false positive from aten can be reversed without
%% triggering elections
T = rand:uniform(Timeout * ?DEFAULT_ELECTION_MULT * 2) + Poll,
{state_timeout, T, election_timeout}.
% sets the tick timer for periodic actions such as sending rpcs to servers
% that are stale to ensure liveness
set_tick_timer(#state{conf = #conf{tick_timeout = TickTimeout}}, Actions) ->
[{{timeout, tick}, TickTimeout, tick_timeout} | Actions].
follower_leader_change(Old, #state{pending_commands = Pending,
leader_monitor = OldMRef} = New) ->
OldLeader = leader_id(Old),
case leader_id(New) of
OldLeader when is_reference(OldMRef) ->
% no change and monitor is still intact
New;
undefined ->
New;
NewLeader ->
MRef = swap_monitor(OldMRef, NewLeader),
LeaderNode = ra_lib:ra_server_id_node(NewLeader),
ok = aten_register(LeaderNode),
OldLeaderNode = ra_lib:ra_server_id_node(OldLeader),
_ = aten:unregister(OldLeaderNode),
% leader has either changed or just been set
?INFO("~ts: detected a new leader ~tw in term ~b",
[log_id(New), NewLeader, current_term(New)]),
[ok = gen_statem:reply(From, {redirect, NewLeader})
|| {From, _Data} <- Pending],
process_pending_queries(NewLeader,
New#state{pending_commands = [],
leader_monitor = MRef})
end.
aten_register(Node) ->
case node() of
Node -> ok;
_ ->
case aten:register(Node) of
ignore ->
ok;
Res ->
Res
end
end.
swap_monitor(MRef, L) ->
stop_monitor(MRef),
erlang:monitor(process, L).
stop_monitor(undefined) ->
ok;
stop_monitor(MRef) ->
erlang:demonitor(MRef),
ok.
gen_statem_safe_call(ServerId, Msg, Timeout) ->
try
gen_statem:call(ServerId, Msg, Timeout)
catch
exit:{timeout, _} ->
timeout;
exit:{noproc, _} ->
{error, noproc};
exit:{{nodedown, _}, _} ->
{error, nodedown};
exit:{shutdown, _} ->
{error, shutdown};
exit:{Reason, _} ->
{error, Reason}
end.
do_state_query(overview, #state{server_state = ServerState,
commit_rate = {CommitRateLI, _, LastRate}}) ->
Overview = ra_server:state_query(overview, ServerState),
Rate = ra_li:rate(erlang:system_time(millisecond), CommitRateLI),
Overview#{commit_rate => Rate,
commit_rate_last => LastRate};
do_state_query(QueryName, #state{server_state = State}) ->
ra_server:state_query(QueryName, State).
update_commit_rate({LI0, LastCI, _LastRate}, ServerState, TickTimeMs) ->
#{commit_index := CI} = ServerState,
Now = erlang:system_time(millisecond),
Delta = CI - LastCI,
LastRate = Delta / TickTimeMs * 1000,
{ra_li:update(Delta, Now, LI0), CI, LastRate}.
reset_commit_rate({LI, _LastCI, _LastRate}, ServerState) ->
#{commit_index := CI} = ServerState,
{ra_li:reset(LI), CI, 0.0}.
config_defaults(ServerId, MetricLabels) ->
Counter = case ra_counters:fetch(ServerId) of
undefined ->
ra_counters:new(ServerId,
{persistent_term, ?FIELDSPEC_KEY},
MetricLabels);
C ->
C
end,
#{broadcast_time => ?DEFAULT_BROADCAST_TIME,
tick_timeout => ?TICK_INTERVAL_MS,
install_snap_rpc_timeout => ?INSTALL_SNAP_RPC_TIMEOUT,
await_condition_timeout => ?DEFAULT_AWAIT_CONDITION_TIMEOUT,
initial_members => [],
counter => Counter,
system_config => ra_system:default_config()
}.
maybe_redirect(From, Msg, #state{pending_commands = Pending,
leader_monitor = LeaderMon} = State) ->
Leader = leader_id(State),
case LeaderMon of
undefined ->
?DEBUG("~ts: leader call - leader not known. "
"Command will be forwarded once leader is known.",
[log_id(State)]),
{keep_state,
State#state{pending_commands = [{From, Msg} | Pending]}};
_ when Leader =/= undefined ->
{keep_state, State, {reply, From, {redirect, Leader}}}
end.
reject_command(Pid, Corr, #state{leader_monitor = _Mon} = State) ->
Id = id(State),
LeaderId = leader_id(State),
case LeaderId of
undefined ->
%% don't log these as they may never be resolved
ok;
Id ->
%% this can happen during an explicit leader change
%% best not rejecting them to oneself!
ok;
_ ->
?INFO("~ts: follower received leader command from ~w. "
"Rejecting to ~tw ", [log_id(State), Pid, LeaderId]),
send_ra_event(Pid, {not_leader, LeaderId, Corr},
id(State), rejected, State)
end.
maybe_persist_last_applied(#state{server_state = NS} = State) ->
State#state{server_state = ra_server:persist_last_applied(NS)}.
send(To, Msg, Conf) ->
% we do not want to block the ra server whilst attempting to set up
% a TCP connection to a potentially down node or when the distribution
% buffer is full, better to drop and try again later
case erlang:send(To, Msg, [noconnect, nosuspend]) of
ok ->
incr_counter(Conf, ?C_RA_SRV_MSGS_SENT, 1),
ok;
Res ->
incr_counter(Conf, ?C_RA_SRV_DROPPED_SENDS, 1),
Res
end.
read_entries0(From, Idxs, #state{server_state = #{log := Log}} = State) ->
ReadState = ra_log:partial_read(Idxs, Log, fun (Idx, Term, Cmd) ->
{Idx, Term, Cmd}
end),
{keep_state, State, [{reply, From, {ok, ReadState}}]}.
send_snapshots(Id, Term, {_, ToNode} = To, ChunkSize,
InstallTimeout, SnapState, Machine, LogId) ->
Context = ra_snapshot:context(SnapState, ToNode),
case ra_snapshot:begin_read(SnapState, Context) of
{error, Reason} ->
?DEBUG("~ts: failed to begin reading snapshot for ~tw: ~p",
[LogId, To, Reason]),
exit({snapshot_begin_read_failed, Reason});
{ok, #{index := SnapIdx,
machine_version := SnapMacVer} = Meta, ReadState} ->
%% TODO: consolidate getting the context, machine version and last
%% applied index in one rpc, and handle errors
TheirMacVer = erpc:call(ToNode, ra_machine, version, [Machine]),
%% only send the snapshot if the target server can accept it
case SnapMacVer > TheirMacVer of
true ->
?DEBUG("~ts: not sending snapshot to ~tw as their machine version ~b "
"is lower than snapshot machine version ~b",
[LogId, To, TheirMacVer, SnapMacVer]),
ok;
false ->
?DEBUG("~ts: sending snapshot at index ~b to ~tw",
[LogId, SnapIdx, To]),
RPC = #install_snapshot_rpc{term = Term,
leader_id = Id,
chunk_state = {0, init},
meta = Meta},
SnapDir = ra_snapshot:current_snapshot_dir(SnapState),
case ra_snapshot:indexes(SnapDir) of
{ok, [_|_] = Indexes0} ->
%% first send the init phase
try gen_statem:call(To, RPC,
{dirty_timeout, InstallTimeout}) of
#install_snapshot_result{} ->
ok;
Unexp ->
?INFO("~ts: ~tw returned an unexpected install snapshot "
"result ~w",
[LogId, To, Unexp]),
exit(unexpected_install_snapshot_result)
catch exit:{noproc, _}:_ ->
?INFO("~ts: ~tw not ready to receive snapshot, "
"reason: noproc", [LogId, To]),
%% no process, not ready yet, exit to reduce sasl logging
exit(snapshot_receiver_not_ready_yet);
C:E:S ->
erlang:raise(C, E, S)
end,
%% there are live indexes to send before the snapshot
#{last_applied := LastApplied} =
erpc:call(ToNode, ra_counters, counters,
[To, [last_applied]]),
%% remove all indexes lower than the target's last applied
Indexes = ra_seq:floor(LastApplied + 1, Indexes0),
?DEBUG("~ts: sending ~b live indexes in the range ~w to ~tw ",
[LogId, ra_seq:length(Indexes), ra_seq:range(Indexes), To]),
MaybeFlru = send_pre_snapshot_entries(Id, To, RPC, Indexes,
InstallTimeout, undefined),
_ = maybe_evict_flru(MaybeFlru),
ok;
_ ->
ok
end,
Result = send_snapshot_chunks(RPC, To, ReadState, 1,
ChunkSize, InstallTimeout,
SnapState),
?DEBUG("~ts: sending snapshot to ~tw completed",
[LogId, To]),
ok = gen_statem:cast(Id, {To, Result})
end
end.
send_snapshot_chunks(RPC0, To, ReadState0, Num, ChunkSize,
InstallTimeout, SnapState) ->
{ok, Data, ContState} = ra_snapshot:read_chunk(ReadState0, ChunkSize,
SnapState),
ChunkFlag = case ContState of
{next, _} -> next;
last -> last
end,
%% TODO: some of the metadata, e.g. the cluster is redundant in subsequent
%% rpcs
RPC1 = RPC0#install_snapshot_rpc{chunk_state = {Num, ChunkFlag},
data = Data},
Res1 = gen_statem:call(To, RPC1,
{dirty_timeout, InstallTimeout}),
case ContState of
{next, ReadState1} ->
send_snapshot_chunks(RPC0, To, ReadState1, Num + 1,
ChunkSize, InstallTimeout, SnapState);
last ->
Res1
end.
%% Send pre-snapshot entries using pipelined async reads.
%% This overlaps reading the next chunk with sending the current chunk
%% to reduce blocking time. Uses lazy chunking to avoid expanding
%% the entire sequence upfront.
send_pre_snapshot_entries(Id, To, RPC, Indexes, InstallTimeout, Flru0) ->
case ra_seq:list_chunk(16, Indexes) of
end_of_seq ->
Flru0;
{FirstIs, SeqIter} ->
%% First chunk: do a blocking read to get started
{ok, {Ents0, Flru1}} = ra_server_proc:read_entries(Id, FirstIs, Flru0,
infinity),
Ents = [map_get(I, Ents0) || I <- FirstIs],
send_pre_snapshot_entries0(Id, To, RPC, SeqIter, Ents,
InstallTimeout, Flru1)
end.
send_pre_snapshot_entries0(Id, To, RPC, SeqIter, CurrentEnts,
InstallTimeout, Flru0) ->
case ra_seq:list_chunk(16, SeqIter) of
end_of_seq ->
%% Last chunk: just send it, no more reads to start
RPC1 = RPC#install_snapshot_rpc{chunk_state = {0, pre},
data = CurrentEnts},
Res = gen_statem:call(To, RPC1, {dirty_timeout, InstallTimeout}),
?assert(is_record(Res, install_snapshot_result)),
Flru0;
{NextIs, NextSeqIter} ->
%% Pipeline: start async read for next chunk, then send current chunk
%% Start async request for the next chunk's read plan
ReqId = gen_statem:send_request(Id, {local_call, {read_entries, NextIs}}),
%% Send current entries to remote server (blocking)
RPC1 = RPC#install_snapshot_rpc{chunk_state = {0, pre},
data = CurrentEnts},
Res = gen_statem:call(To, RPC1, {dirty_timeout, InstallTimeout}),
?assert(is_record(Res, install_snapshot_result)),
%% Wait for the async read plan response
{ok, ReadPlan} = case gen_statem:receive_response(ReqId, infinity) of
{reply, {ok, RP}} ->
{ok, RP};
{error, Reason} ->
error(Reason);
timeout ->
error(read_entries_timeout)
end,
%% Execute the read plan to get actual entries
{Reads, Flru1} = ra_log:execute_read_plan(ReadPlan, Flru0,
fun (Idx, Term, Cmd) ->
{Idx, Term, Cmd}
end,
#{access_pattern => random,
file_advise => random,
index_mode => binary}),
NextEnts = [map_get(I, Reads) || I <- NextIs],
send_pre_snapshot_entries0(Id, To, RPC, NextSeqIter, NextEnts,
InstallTimeout, Flru1)
end.
maybe_evict_flru(undefined) ->
ok;
maybe_evict_flru(Flru) ->
ra_flru:evict_all(Flru).
validate_reply_mode(after_log_append) ->
ok;
validate_reply_mode(await_consensus) ->
ok;
validate_reply_mode({await_consensus, Options}) when is_map(Options) ->
validate_reply_mode_options(Options);
validate_reply_mode({notify, Correlation, Pid})
when (is_integer(Correlation) orelse
is_reference(Correlation)) andalso
is_pid(Pid) ->
ok;
validate_reply_mode({notify, Correlation, Pid, Options})
when (is_integer(Correlation) orelse
is_reference(Correlation)) andalso
is_pid(Pid) andalso
is_map(Options) ->
validate_reply_mode_options(Options);
validate_reply_mode(noreply) ->
ok;
validate_reply_mode(ReplyMode) ->
{error, {invalid_reply_mode, ReplyMode}}.
validate_reply_mode_options(Options) when is_map(Options) ->
maps:fold(fun (Key, Value, ok) ->
case {Key, Value} of
{reply_from, local} ->
ok;
{reply_from, {member, _}} ->
ok;
{reply_from, leader} ->
ok;
{_, _} ->
{error, {unknown_option, Key, Value}}
end;
(_Key, _Value, Error) ->
Error
end, ok, Options).
maybe_set_election_timeout(_TimeoutLen,
#state{election_timeout_set = true} = State,
Actions) ->
%% already set, don't update
{State, Actions};
maybe_set_election_timeout(TimeoutLen, State, Actions) ->
{State#state{election_timeout_set = true},
[election_timeout_action(TimeoutLen, State) | Actions]}.
next_state(leader, #state{pending_commands = Pending} = State, Actions) ->
NextEvents = [{next_event, {call, F}, Cmd} || {F, Cmd} <- Pending],
{next_state, leader, State#state{election_timeout_set = false,
pending_commands = []},
Actions ++ NextEvents};
next_state(Next, State, Actions) ->
%% as changing states will always cancel the state timeout we need
%% to set our own state tracking to false here
{next_state, Next, State#state{election_timeout_set = false}, Actions}.
send_msg({send_msg, To, Msg, Options}, State) ->
incr_counter(State#state.conf, ?C_RA_SRV_SEND_MSG_EFFS_SENT, 1),
case parse_send_msg_options(Options) of
{true, true} ->
gen_cast(To, wrap_ra_event(State, leader_id(State), machine, Msg),
State);
{true, false} ->
send(To, wrap_ra_event(State, leader_id(State), machine, Msg),
State#state.conf);
{false, true} ->
gen_cast(To, Msg, State);
{false, false} ->
send(To, Msg, State#state.conf)
end.
has_local_opt(local) ->
true;
has_local_opt(Opt) when is_atom(Opt) ->
false;
has_local_opt(Opts) ->
lists:member(local, Opts).
get_node(P) when is_pid(P) ->
node(P);
get_node({_, Node}) ->
Node;
get_node(Proc) when is_atom(Proc) ->
node().
can_execute_locally(RaftState, TargetNode,
#state{server_state = ServerState} = State) ->
case RaftState of
leader when TargetNode =/= node() ->
%% We need to evaluate whether to send the message.
%% Only send if there isn't a local node for the target pid.
Members = do_state_query(voters, State),
not lists:any(fun ({_, N}) -> N == TargetNode end, Members);
leader ->
true;
_ when RaftState =/= leader ->
TargetNode == node() andalso
voter == ra_server:get_membership(ServerState)
end.
can_execute_on_member(_RaftState, Member,
#state{server_state = #{cfg := #cfg{id = Member}}}) ->
true;
can_execute_on_member(leader, Member, State) ->
Members = do_state_query(members, State),
not lists:member(Member, Members);
can_execute_on_member(_RaftState, _Member, _State) ->
false.
handle_node_status_change(Node, Status, InfoList, RaftState,
#state{monitors = Monitors0,
server_state = ServerState0} = State0) ->
{Comps, Monitors} = ra_monitors:handle_down(Node, Monitors0),
{_, ServerState1, Effects} =
lists:foldl(
fun (Comp, {R, S0, E0}) ->
{R, S, E} = ra_server:handle_node_status(R, Comp, Node,
Status, InfoList,
S0),
{R, S, E0 ++ E}
end, {RaftState, ServerState0, []}, Comps),
ServerState = ra_server:update_disconnected_peers(Node, Status,
ServerState1),
{State, Actions} = handle_effects(RaftState, Effects, cast,
State0#state{server_state = ServerState,
monitors = Monitors}),
{keep_state, State, Actions}.
handle_process_down(Pid, Info, _RaftState,
#state{conf = #conf{worker_pid = Pid}} = State) ->
?WARN("~ts: worker exited with ~w",
[log_id(State), Info]),
{stop, Info, State};
handle_process_down(Pid, Info, RaftState,
#state{monitors = Monitors0,
pending_notifys = Nots,
server_state = ServerState0} = State0) ->
{Comps, Monitors} = ra_monitors:handle_down(Pid, Monitors0),
{ServerState, Effects} =
lists:foldl(
fun(Comp, {S0, E0}) ->
{_, S, E} = ra_server:handle_down(RaftState, Comp,
Pid, Info, S0),
{S, E0 ++ E}
end, {ServerState0, []}, Comps),
{State, Actions} =
handle_effects(RaftState, Effects, cast,
State0#state{server_state = ServerState,
pending_notifys = maps:remove(Pid, Nots),
monitors = Monitors}),
{keep_state, State, Actions}.
maybe_record_cluster_change(#state{conf = #conf{cluster_name = ClusterName},
server_state = ServerStateA},
#state{server_state = ServerStateB}) ->
LeaderA = ra_server:leader_id(ServerStateA),
LeaderB = ra_server:leader_id(ServerStateB),
if (map_get(cluster_index_term, ServerStateA) =/=
map_get(cluster_index_term, ServerStateB) orelse
LeaderA =/= LeaderB) ->
MembersB = ra_server:state_query(members, ServerStateB),
ok = ra_leaderboard:record(ClusterName, LeaderB, MembersB);
true ->
ok
end.
record_cluster_change(#state{conf = #conf{cluster_name = ClusterName},
server_state = ServerState}) ->
Leader = ra_server:state_query(leader, ServerState),
Members = ra_server:state_query(members, ServerState),
ok = ra_leaderboard:record(ClusterName, Leader, Members).
incr_counter(#conf{counter = Cnt}, Ix, N) when Cnt =/= undefined ->
counters:add(Cnt, Ix, N);
incr_counter(#conf{counter = undefined}, _Ix, _N) ->
ok.
update_peer(PeerId, Update,
#state{server_state = ServerState} = State0)
when is_map(Update) ->
State0#state{server_state =
ra_server:update_peer(PeerId, Update, ServerState)}.
send_applied_notifications(Nots0, #state{pending_notifys = PendingNots} = State)
when map_size(PendingNots) > 0 ->
Nots = maps:merge_with(fun(_K, V1, V2) ->
V1 ++ V2
end, PendingNots, Nots0),
send_applied_notifications(Nots, State#state{pending_notifys = #{}});
send_applied_notifications(Nots, #state{} = State) ->
Id = id(State),
%% any notifications that could not be sent
%% will be kept and retried
RemNots = maps:filter(
fun(Who, Correlations0) ->
%% correlations are build up in reverse order so we need
%% to reverse before sending
Correlations = lists:reverse(Correlations0),
ok =/= send_ra_event(Who, Correlations, Id,
applied, State)
end, Nots),
case map_size(RemNots) of
0 ->
State;
_ ->
State#state{pending_notifys = RemNots}
end.
make_command(Type, {call, From}, Data, Mode) ->
Ts = erlang:system_time(millisecond),
{Type, #{from => From, ts => Ts}, Data, Mode};
make_command(Type, _, Data, Mode) ->
Ts = erlang:system_time(millisecond),
{Type, #{ts => Ts}, Data, Mode}.
schedule_command_flush(Delayed) ->
case ra_ets_queue:len(Delayed) of
0 ->
ok;
_ ->
ok = gen_statem:cast(self(), flush_commands)
end.
handle_log_effect(log, Idxs, Fun,
#state{server_state = SS0} = State)
when is_list(Idxs) ->
%% Useful to implement a batch send of data obtained from the log.
%% 1) Retrieve all data from the list of indexes
{ok, Cmds, SS} = ra_server:log_read(Idxs, SS0),
%% 2) Apply the fun to the list of commands as a whole and deal with any effects
{Fun(Cmds), State#state{server_state = SS}};
handle_log_effect(log_ext, Idxs, Fun,
#state{server_state = SS0} = State)
when is_list(Idxs) ->
ReadState = ra_server:log_partial_read(Idxs, SS0),
{Fun(ReadState), State}.
%% Strip function arguments from stacktrace frames to avoid formatting
%% potentially huge terms (e.g. the full Ra machine state).
safe_stacktrace(Stacktrace) ->
lists:map(fun({M, F, Args, Info}) when is_list(Args) ->
{M, F, length(Args), Info};
(Frame) ->
Frame
end, Stacktrace).