ra_server.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-2023 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,
         metrics/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/3,
         promote_checkpoint/2,
         checkpoint/3,
         persist_last_applied/1,
         update_peer/3,
         register_external_log_reader/2,
         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(),
      stop_after => 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()),
      meta_fd => option(file:fd())
     }.

-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, term(), ra:query_fun()} |
                  #heartbeat_rpc{} |
                  #info_rpc{} |
                  #info_reply{} |
                  {ra_server_id, #heartbeat_reply{}} |
                  pipeline_rpcs.

-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{}} |
    {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.

-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.
%%
%% <ul>
%% <li>`all': a new machine version it used only when all members of the
%% cluster support it. This is the default behavior starting with Ra
%% 2.16.</li>
%% <li>`quorum': when a leader is elected and if it supports a newer machine
%% version, it switches to that new version. This was the default and only
%% possible behavior in Ra up to 2.15.</li>
%% </ul>

-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 `~w'
                              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(),
                              % 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()
                             }.

-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:
%% <ul>
%% <li>`machine_version': Highest machine version supported by this Ra
%% server.</li>
%% </ul>
%%
%% 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(),
                            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()]}}.

-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("~w", [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,
                                    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),
    DataDir = maps:get(data_dir, SystemConfig),
    MetaFile = ra_server_meta:path(DataDir, UId),

    {ok, {VotedFor, CurrentTerm, LastApplied}} = ra_server_meta:fetch(MetaFile, MetaName, UId),
    {ok, MetaFd} = file:open(MetaFile, [read, write, raw, binary]),

    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),
    Cfg = #cfg{id = Id,
               uid = UId,
               log_id = LogId,
               metrics_key = MetricKey,
               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},
    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,
      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 => [],
      meta_fd => MetaFd}.

recover(#{cfg := #cfg{log_id = LogId,
                      machine_version = MacVer,
                      effective_machine_version = EffMacVer} = Cfg,
          commit_index := CommitIndex,
          last_applied := LastApplied} = State0) ->
    put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_APPLIED, LastApplied),
    ?DEBUG("~ts: recovering state machine version ~b:~b from index ~b to ~b",
           [LogId, EffMacVer, MacVer, LastApplied, CommitIndex]),
    Before = erlang:system_time(millisecond),
    {#{log := Log0,
       cfg := #cfg{effective_machine_version = EffMacVerAfter}} = State1, _} =
        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, Idx),
                         {Mod, LastAppl, S, MacSt, [], #{}, LastTs}
                 end,
                 State0, []),
    After = erlang:system_time(millisecond),
    ?DEBUG("~ts: recovery of state machine version ~b:~b "
           "from index ~b to ~b took ~bms",
           [LogId, EffMacVerAfter, MacVer, LastApplied, 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]),
    {State, Log} = ra_log:fold(FromScan, ToScan,
                               fun cluster_scan_fun/2,
                               State1, Log0),

    put_counter(Cfg, ?C_RA_SVR_METRIC_COMMIT_LATENCY, 0),
    State#{log => Log,
           %% 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 ~w",
                  [LogId, PeerId]),
            {leader, State0, []};
        Peer0 = #{match_index := MI, next_index := NI} ->
            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 ~w",
                  [LogId, PeerId]),
            {leader, State0, []};
        _ ->
            ?NOTICE("~ts: leader saw append_entries_reply from ~w 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 ~w",
          [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 ~w 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 ~w",
                                 [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 ~w 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,
                              %% TODO: make duration configurable?
                              timeout => #{duration => 5000,
                                           effects => CondEffs,
                                           transition_to => leader}}},

            {await_condition, State, Effects0};
        {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 higest 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 ~w",
                  [LogId, PeerId]),
            {leader, State0, []};
        _ ->
            ?DEBUG("~ts: leader saw install_snapshot_result from ~w 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 ~w",
                  [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,

            {State, _, Effects} = make_pipelined_rpc_effects(State1, Effects0),
            {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(#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 ~w",
                  [LogId, Leader]),
            {leader, State0, []};
        _ ->
            ?INFO("~ts: leader saw install_snapshot_rpc from ~w 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 ~w 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 = {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 = {From, QueryFun, 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 ~w 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 ~w 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 ~w",
                  [LogId, Cand]),
            {leader, State0, []};
        _ ->
            ?INFO("~ts: leader saw request_vote_rpc from ~w 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 ~w",
                  [LogId, Cand]),
            {leader, State0, []};
        _ ->
            ?INFO("~ts: leader saw pre_vote_rpc from ~w 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 ~w 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 ~w 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 ~w",
           [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 ~w",
                  [LogId, ServerId]),
            {leader, State, [{reply, {error, non_voter}}]};
        _ ->
            ?DEBUG("~ts: transfer leadership to ~w 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({register_external_log_reader, Pid}, #{log := Log0} = State) ->
    {Log, Effs} = ra_log:register_reader(Pid, Log0),
    {leader, State#{log => Log}, Effs};
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({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({register_external_log_reader, Pid}, #{log := Log0} = State) ->
    {Log, Effs} = ra_log:register_reader(Pid, Log0),
    {candidate, State#{log => Log}, Effs};
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 ~w 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({register_external_log_reader, Pid}, #{log := Log0} = State) ->
    {Log, Effs} = ra_log:register_reader(Pid, Log0),
    {pre_vote, State#{log => Log}, Effs};
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),
                            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 ~w 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} ->
            LastApplied = maps:get(last_applied, State00),
            ?INFO("~ts: term mismatch - follower had entry at ~b with term ~b "
                  "but not with term ~b. "
                  "Asking leader ~w 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,
                                                           State0),
            Effects = [cast_reply(Id, LeaderId, Reply) | Effects0],
            {await_condition,
             State#{log => Log0,
                    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 ~w 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},
                                       leader_id := LeaderId,
                                       current_term := Term} = State0) ->
    % 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},
    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, State),
            {follower, State, [cast_reply(Id, LeaderId, Reply) | Effects]};
        _ ->
            {follower, State, Effects}
    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 ~w already voted for ~w 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 ~w 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 ~w 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 ~w 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 = {1, _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 ->
    %% only begin snapshot procedure if Idx is higher than the last_applied
    %% index.
    ?DEBUG("~ts: begin_accept snapshot at index ~b in term ~b",
           [LogId, SnapIdx, Term]),
    SnapState0 = ra_log:snapshot_state(Log0),
    {ok, SS} = ra_snapshot:begin_accept(Meta, SnapState0),
    Log = ra_log:set_snapshot_state(SS, Log0),
    {receive_snapshot, update_term(Term, State0#{log => Log,
                                                 leader_id => LeaderId}),
     [{next_event, Rpc}, {record_leader_msg, LeaderId}]};
handle_follower(#install_snapshot_rpc{term = Term,
                                      meta = #{index := LastIndex,
                                               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 lower last index ~b in ~b",
           [LogId, LastIndex, Term]),
    %% 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({register_external_log_reader, Pid}, #{log := Log0} = State) ->
    {Log, Effs} = ra_log:register_reader(Pid, Log0),
    {follower, State#{log => Log}, Effs};
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 ~w",
          [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,
                                                       term := SnapTerm} = SnapMeta,
                                              chunk_state = {Num, ChunkFlag},
                                              data = Data},
                        #{cfg := #cfg{id = Id,
                                      log_id = LogId,
                                      effective_machine_version = CurEffMacVer,
                                      machine_versions = MachineVersions,
                                      machine = Machine} = Cfg0,
                          log := Log0,
                          cluster := Cluster,
                          current_term := CurTerm,
                          last_applied := LastApplied,
                          machine_state := OldMacState} = State0)
  when Term >= CurTerm ->
    ?DEBUG("~ts: receiving snapshot chunk: ~b / ~w, index ~b, term ~b",
           [LogId, Num, ChunkFlag, SnapIndex, SnapTerm]),
    SnapState0 = ra_log:snapshot_state(Log0),
    {ok, SnapState} = ra_snapshot:accept_chunk(Data, Num, ChunkFlag,
                                               SnapState0),
    Reply = #install_snapshot_result{term = CurTerm,
                                     last_term = SnapTerm,
                                     last_index = SnapIndex},
    case ChunkFlag of
        last ->
            %% this is the last chunk so we can "install" it
            {Log, Effs} = ra_log:install_snapshot({SnapIndex, SnapTerm},
                                                  SnapState, Log0),
            %% 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,

            {#{cluster := ClusterIds}, MacState} = ra_log:recover_snapshot(Log),

            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),
            State = 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}),
            %% it was the last snapshot chunk so we can revert back to
            %% follower status
            {follower, persist_last_applied(State), [{reply, Reply} |
                                                     Effs ++ SnapInstalledEffs]};
        next ->
            Log = ra_log:set_snapshot_state(SnapState, Log0),
            State = update_term(Term, State0#{log => Log}),
            {receive_snapshot, State, [{reply, Reply}]}
    end;
handle_receive_snapshot(#append_entries_rpc{term = Term} = Msg,
                        #{current_term := CurTerm,
                          cfg := #cfg{log_id = LogId},
                          log := Log0} = State)
  when Term > CurTerm ->
    ?INFO("~ts: follower receiving snapshot saw append_entries_rpc from ~w for term ~b "
          "abdicates term: ~b!",
          [LogId, Msg#append_entries_rpc.leader_id,
           Term, CurTerm]),
    SnapState0 = ra_log:snapshot_state(Log0),
    SnapState = ra_snapshot:abort_accept(SnapState0),
    Log = ra_log:set_snapshot_state(SnapState, Log0),
    {follower, update_term(Term, clear_leader_id(State#{log => Log})),
     [{next_event, Msg}]};
handle_receive_snapshot({ra_log_event, Evt},
                        State = #{cfg := #cfg{id = _Id, log_id = LogId},
                                  log := Log0}) ->
    ?DEBUG("~ts: ~s ra_log_event received: ~w",
          [LogId, ?FUNCTION_NAME, Evt]),
    % simply forward all other events to ra_log
    % whilst the snapshot is being received
    {Log, Effects} = ra_log:handle_event(Evt, Log0),
    {receive_snapshot, State#{log => Log}, Effects};
handle_receive_snapshot(receive_snapshot_timeout, #{log := Log0} = State) ->
    SnapState0 = ra_log:snapshot_state(Log0),
    SnapState = ra_snapshot:abort_accept(SnapState0),
    Log = ra_log:set_snapshot_state(SnapState, Log0),
    {follower, State#{log => Log}, []};
handle_receive_snapshot({register_external_log_reader, Pid}, #{log := Log0} = State) ->
    {Log, Effs} = ra_log:register_reader(Pid, Log0),
    {receive_snapshot, State#{log => Log}, Effs};
handle_receive_snapshot(#info_rpc{term = Term} = Msg,
                        #{current_term := CurTerm,
                          cfg := #cfg{log_id = LogId},
                          log := Log0} = State)
  when CurTerm < Term ->
    ?INFO("~ts: follower receiving snapshot saw info_rpc from ~w for term ~b "
          "abdicates term: ~b!",
          [LogId, Msg#info_rpc.from,
           Term, CurTerm]),
    SnapState0 = ra_log:snapshot_state(Log0),
    SnapState = ra_snapshot:abort_accept(SnapState0),
    Log = ra_log:set_snapshot_state(SnapState, Log0),
    {follower, update_term(Term, clear_leader_id(State#{log => Log})),
     [{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},
                          log := Log0} = State)
  when CurTerm < Term ->
    ?INFO("~ts: follower receiving snapshot saw info_reply from ~w for term ~b "
          "abdicates term: ~b!",
          [LogId, Msg#info_reply.from,
           Term, CurTerm]),
    SnapState0 = ra_log:snapshot_state(Log0),
    SnapState = ra_snapshot:abort_accept(SnapState0),
    Log = ra_log:set_snapshot_state(SnapState, Log0),
    {follower, update_term(Term, clear_leader_id(State#{log => Log})),
     [{next_event, Msg}]};
handle_receive_snapshot(#info_reply{}, State) ->
    {receive_snapshot, State, []};
handle_receive_snapshot(Msg, State) ->
    log_unhandled_msg(receive_snapshot, Msg, State),
    %% drop all other events??
    %% TODO: work out what else to handle
    {receive_snapshot, State, [{reply, {error, {unsupported_call, Msg}}}]}.

-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({register_external_log_reader, Pid}, #{log := Log0} = State) ->
    {Log, Effs} = ra_log:register_reader(Pid, Log0),
    {await_condition, State#{log => Log}, Effs};
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
            % log_unhandled_msg(await_condition, Msg, State),
            {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 metrics(ra_server_state()) ->
    {atom(), ra_term(),
     ra_index(), ra_index(),
     ra_index(), ra_index(), non_neg_integer()}.
metrics(#{cfg := #cfg{metrics_key = Key},
          commit_index := CI,
          last_applied := LA,
          current_term := CT,
          log := Log} = State) ->
    SnapIdx = case ra_log:snapshot_index_term(Log) of
                  undefined -> 0;
                  {I, _} -> I
              end,
    CL = case  State of
             #{commit_latency := L} ->
                 L;
             _ ->
                 0
         end,
    {LW, _} = ra_log:last_index_term(Log),
    {Key, CT, SnapIdx, LA, CI, LW, CL}.

-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, Cmd,
           #{cfg := #cfg{effective_machine_module = MacMod,
                         effective_handle_aux_fun = {handle_aux, 5}},
             aux_state := Aux0} = State0) ->
    %% NEW API
    case ra_machine:handle_aux(MacMod, RaftState, Type, Cmd, Aux0,
                               State0) of
        {reply, Reply, Aux, State} ->
            {RaftState, State#{aux_state => Aux},
             [{reply, Reply}]};
        {reply, Reply, Aux, State, Effects} ->
            {RaftState, State#{aux_state => Aux},
             [{reply, 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, Cmd,
           #{cfg := #cfg{effective_machine_module = MacMod,
                         effective_handle_aux_fun = {handle_aux, 6}},
             aux_state := Aux0,
             machine_state := MacState,
             log := Log0} = State0) ->
    %% 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, Reply}]};
        {reply, Reply, Aux, Log, Effects} ->
            {RaftState, State0#{log => Log, aux_state => Aux},
             [{reply, 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

-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(fun((term()) -> term()), ra_server_state()) ->
    {ra_idxterm(), term()}.
machine_query(QueryFun, #{cfg := #cfg{effective_machine_module = MacMod},
                          machine_state := MacState,
                          last_applied := Last,
                          current_term := Term
                         }) ->
    Res = ra_machine:query(MacMod, QueryFun, MacState),
    {{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, _, #{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)};
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} ->
            case LastApplied > LastApplied0 of
                true ->
                    %% entries were applied, append eval_aux effect
                    {follower, State, [{aux, eval} | Effects]};
                false ->
                    %% no entries were applied
                    {follower, State, Effects}
            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),
    {State2, EffectsAER} = make_rpcs_for(peers_with_normal_status(State1), State1),
    {State2, 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 ~w 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(), ra_server_state()) ->
    {ra_server_state(), effects()}.
update_release_cursor(Index, MacState,
                      State = #{log := Log0, cluster := Cluster}) ->
    MacVersion = index_machine_version(Index, State),
    % simply pass on release cursor index to log
    {Log, Effects} = ra_log:update_release_cursor(Index, Cluster,
                                                  MacVersion,
                                                  MacState, Log0),
    {State#{log => Log}, Effects}.

-spec checkpoint(ra_index(), term(), ra_server_state()) ->
      {ra_server_state(), effects()}.
checkpoint(Index, MacState,
           State = #{log := Log0, cluster := Cluster}) ->
    MacVersion = index_machine_version(Index, State),
    {Log, Effects} = ra_log:checkpoint(Index, Cluster,
                                       MacVersion, 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 PL for some reason do nothing
    State;
persist_last_applied(#{last_applied := LastApplied,
                       meta_fd := MetaFd} = State) ->
    ok = ra_server_meta:update_last_applied(MetaFd, LastApplied),
    State#{persisted_last_applied => LastApplied}.


-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 register_external_log_reader(pid(), ra_server_state()) ->
    {ra_server_state(), effects()}.
register_external_log_reader(Pid, #{log := Log0} = State) ->
    {Log, Effs} = ra_log:register_reader(Pid, Log0),
    {State#{log => Log}, Effs}.

-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};
                            (_, Peer) ->
                                Peer
                        end, Peers)};
update_disconnected_peers(_Node, _Status, State) ->
    State.

peer_snapshot_process_exited(SnapshotPid, #{cluster := Peers} = State) ->
     PeerKv =
         maps:to_list(
           maps:filter(fun(_, #{status := {sending_snapshot, Pid}})
                             when Pid =:= SnapshotPid ->
                               true;
                          (_, _) -> false
                       end, Peers)),
     case PeerKv of
         [{PeerId, Peer}] ->
             put_peer(PeerId, Peer#{status => normal}, State);
         _ ->
             State
     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}} = State)
  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
    ?DEBUG_IF(Info /= normal,
              "~ts: Snapshot sender process ~w exited with ~W",
              [LogId, Pid, Info, 10]),
    {leader, peer_snapshot_process_exited(Pid, State), []};
handle_down(RaftState, snapshot_writer, Pid, Info,
            #{cfg := #cfg{log_id = LogId}, log := Log0} = State)
  when is_pid(Pid) ->
    case Info of
        noproc -> ok;
        normal -> ok;
        _ ->
            ?WARN("~ts: Snapshot write process ~w exited with ~w",
                  [LogId, Pid, Info])
    end,
    SnapState0 = ra_log:snapshot_state(Log0),
    SnapState = ra_snapshot:handle_down(Pid, Info, SnapState0),
    Log = ra_log:set_snapshot_state(SnapState, Log0),
    {RaftState, State#{log => Log}, []};
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,
            meta_fd := MetaFd,
            cfg := #cfg{log_id = LogId}} = _State, {shutdown, delete}) ->
    ?NOTICE("~ts: terminating with reason 'delete'", [LogId]),
    _ = file:close(MetaFd),
    catch ra_log:delete_everything(Log),
    ok;
terminate(#{cfg := #cfg{log_id = LogId}, meta_fd := MetaFd} = State, Reason) ->
    ?DEBUG("~ts: terminating with reason '~w'", [LogId, Reason]),
    #{log := Log} = persist_last_applied(State),
    _ = file:sync(MetaFd),
    _ = file:close(MetaFd),
    catch ra_log:close(Log),
    ok.

-spec log_fold(ra_server_state(), fun((term(), State) -> State), State) ->
    {ok, State, ra_server_state()} |
    {error, term(), ra_server_state()}.
log_fold(#{log := Log} = RaState, Fun, State) ->
    Idx = case ra_log:snapshot_index_term(Log) of
              {PrevIdx, _PrevTerm} ->
                  PrevIdx;
              undefined ->
                  1
          end,
    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  ->
    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 ~w 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 ~w"
                   " machine version (their:ours:effective) ~b:~b:~b"
                   " with last {index, term} ~w"
                   " for term ~b previous term ~b",
                   [log_id(State0), Cand, TheirMacVer, OurMacVer, EffMacVer,
                    {LLIdx, LLTerm}, Term, CurTerm]),
            {FsmState, State#{voted_for => Cand},
             [{reply, pre_vote_result(Term, Token, true)}]};
        true ->
            ?DEBUG("~ts: declining pre-vote for ~w 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 ~w for term ~b,"
                   " candidate last log {index, term} was: ~w "
                   "last log entry {index, term} seen is: ~w",
                   [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 ~w 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).

%% remove any peers that are currently receiving a snapshot
peers_with_normal_status(State) ->
    maps:filter(fun (_, #{status := normal}) -> true;
                    (_, _) -> false
                end, peers(State)).

% 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,
                                            current_term := CurTerm,
                                            meta_fd := MetaFd} = State) ->
    CurVotedFor = maps:get(voted_for, State, undefined),
    case Term =:= CurTerm andalso VotedFor =:= CurVotedFor of
        true ->
            %% no update needed
            State;
        false ->
            LastApplied = maps:get(last_applied, State, 0),
            ok = ra_server_meta:store_sync(MetaFd, VotedFor, Term, LastApplied),
            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 ~w",
           [log_id(State0), ?FUNCTION_NAME, 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}};
cluster_scan_fun(_Cmd, State) ->
    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 ~w",
                           [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 ~w",
                           [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) ->
    % TODO: remove to make more strict, ideally we should not need a catch all
    ?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 ~w requested to leave but was not a member. "
                   "Members: ~w",
                   [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 ~w",
                   [log_id(State), ?FUNCTION_NAME, maps:keys(Cluster)]),
            State#{cluster => Cluster,
                   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 ~w",
                   [log_id(State), ?FUNCTION_NAME, maps:keys(PrevCluster)]),
            State1 = State#{cluster => PrevCluster,
                            cluster_index_term => {PrevIdx, PrevTerm}},
            pre_append_log_follower(Entry, State1)
    end;
pre_append_log_follower({Idx, Term, {'$ra_cluster_change', _, Cluster, _}},
                        State) ->
    ?DEBUG("~ts: ~ts: follower applying ra cluster change to ~w",
           [log_id(State), ?FUNCTION_NAME, maps:keys(Cluster)]),
    State#{cluster => Cluster,
           membership => get_membership(Cluster, State),
           cluster_index_term => {Idx, Term}};
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,
    apply_to(CI, State, Effects).

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, 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_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.

-spec apply_consistent_queries_effects([consistent_query_ref()],
                                       ra_server_state()) ->
    effects().
apply_consistent_queries_effects(QueryRefs,
                                 #{last_applied := LastApplied} = State) ->
    lists:map(fun({_, _, ReadCommitIndex} = QueryRef) ->
                      true = LastApplied >= ReadCommitIndex,
                      consistent_query_reply(QueryRef, State)
              end, QueryRefs).

-spec consistent_query_reply(consistent_query_ref(), ra_server_state()) -> effect().
consistent_query_reply({From, QueryFun, _ReadCommitIndex},
                       #{cfg := #cfg{id = Id,
                                     machine = {machine, MacMod, _}},
                         machine_state := MacState
                         }) ->
    Result = ra_machine:query(MacMod, QueryFun, MacState),
    {reply, From, {ok, Result, Id}}.

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(#{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.