ra

%% This Source Code Form is subject to the terms of the Mozilla Public
%% License, v. 2.0. If a copy of the MPL was not distributed with this
%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
%%
%% Copyright (c) 2017-2025 Broadcom. All Rights Reserved. The term Broadcom refers to Broadcom Inc. and/or its subsidiaries.
%%
%% @hidden
-module(ra_log).

-compile([inline_list_funcs]).

-export([pre_init/1,
         init/1,
         close/1,
         begin_tx/1,
         commit_tx/1,
         append/2,
         write/2,
         append_sync/2,
         write_sync/2,
         fold/5,
         sparse_read/2,
         partial_read/3,
         execute_read_plan/4,
         read_plan_info/1,
         last_index_term/1,
         set_last_index/2,
         handle_event/2,
         last_written/1,
         fetch/2,
         fetch_term/2,
         next_index/1,
         snapshot_state/1,
         set_snapshot_state/2,
         install_snapshot/3,
         recover_snapshot/1,
         snapshot_index_term/1,
         update_release_cursor/5,
         checkpoint/5,
         promote_checkpoint/2,

         can_write/1,
         exists/2,
         overview/1,
         %% config
         write_config/2,
         read_config/1,

         delete_everything/1,
         release_resources/3,

         % external reader
         register_reader/2,
         readers/1,
         tick/2
        ]).

-include("ra.hrl").

-define(DEFAULT_RESEND_WINDOW_SEC, 20).
-define(MIN_SNAPSHOT_INTERVAL, 4096).
-define(MIN_CHECKPOINT_INTERVAL, 16384).
-define(LOG_APPEND_TIMEOUT, 5000).
-define(WAL_RESEND_TIMEOUT, 5000).

-type ra_meta_key() :: atom().
-type segment_ref() :: {ra_range:range(), File :: file:filename_all()}.
-type event_body() :: {written, ra_term(), ra:range()} |
                      {segments, [{ets:tid(), ra:range()}], [segment_ref()]} |
                      {resend_write, ra_index()} |
                      {snapshot_written, ra_idxterm(), ra_snapshot:kind()} |
                      {down, pid(), term()}.

-type event() :: {ra_log_event, event_body()}.
-type transform_fun() :: fun ((ra_index(), ra_term(), ra_server:command()) -> term()).

-type effect() ::
    {delete_snapshot, Dir :: file:filename_all(), ra_idxterm()} |
    {monitor, process, log, pid()} |
    ra_snapshot:effect() |
    ra_server:effect().
%% logs can have effects too so that they can be coordinated with other state
%% such as avoiding to delete old snapshots whilst they are still being
%% replicated

-type effects() :: [effect()].

-record(cfg, {uid :: ra_uid(),
              log_id :: unicode:chardata(),
              directory :: file:filename_all(),
              min_snapshot_interval = ?MIN_SNAPSHOT_INTERVAL :: non_neg_integer(),
              min_checkpoint_interval = ?MIN_CHECKPOINT_INTERVAL :: non_neg_integer(),
              snapshot_module :: module(),
              resend_window_seconds = ?DEFAULT_RESEND_WINDOW_SEC :: integer(),
              wal :: atom(),
              segment_writer :: atom(),
              counter :: undefined | counters:counters_ref(),
              names :: ra_system:names()}).

-record(?MODULE,
        {cfg = #cfg{},
         %% mutable data below
         first_index = -1 :: ra_index(),
         last_index = -1 :: -1 | ra_index(),
         last_term = 0 :: ra_term(),
         last_written_index_term = {0, 0} :: ra_idxterm(),
         snapshot_state :: ra_snapshot:state(),
         last_resend_time :: option({integer(), WalPid :: pid() | undefined}),
         last_wal_write :: {pid(), Ms :: integer()},
         reader :: ra_log_reader:state(),
         readers = [] :: [pid()],
         mem_table :: ra_mt:state(),
         tx = false :: boolean()
        }).

-record(read_plan, {dir :: file:filename_all(),
                    read :: #{ra_index() := log_entry()},
                    plan :: ra_log_reader:read_plan()}).

-opaque read_plan() :: #read_plan{}.
-opaque state() :: #?MODULE{}.

-type ra_log_init_args() :: #{uid := ra_uid(),
                              system_config => ra_system:config(),
                              log_id => unicode:chardata(),
                              %% Deprecated in favor of `min_snapshot_interval'
                              %% but this value is used as a fallback if
                              %% `min_snapshot_interval' is not provided.
                              snapshot_interval => non_neg_integer(),
                              min_snapshot_interval => non_neg_integer(),
                              min_checkpoint_interval => non_neg_integer(),
                              resend_window => integer(),
                              max_open_segments => non_neg_integer(),
                              snapshot_module => module(),
                              counter => counters:counters_ref(),
                              initial_access_pattern => sequential | random,
                              max_checkpoints => non_neg_integer()}.


-type overview() ::
    #{type := ra_log,
      last_index := ra_index(),
      last_term := ra_term(),
      first_index := ra_index(),
      last_written_index_term := ra_idxterm(),
      num_segments := non_neg_integer(),
      open_segments => non_neg_integer(),
      snapshot_index => undefined | ra_index(),
      snapshot_term => undefined | ra_index(),
      mem_table_size => non_neg_integer(),
      latest_checkpoint_index => undefined | ra_index(),
      atom() => term()}.

-export_type([state/0,
              read_plan/0,
              ra_log_init_args/0,
              ra_meta_key/0,
              segment_ref/0,
              event/0,
              event_body/0,
              effect/0,
              overview/0
             ]).

-define(SNAPSHOTS_DIR, <<"snapshots">>).
-define(CHECKPOINTS_DIR, <<"checkpoints">>).

pre_init(#{uid := UId,
           system_config := #{data_dir := DataDir}} = Conf) ->
    Dir = server_data_dir(DataDir, UId),
    SnapModule = maps:get(snapshot_module, Conf, ?DEFAULT_SNAPSHOT_MODULE),
    MaxCheckpoints = maps:get(max_checkpoints, Conf, ?DEFAULT_MAX_CHECKPOINTS),
    SnapshotsDir = filename:join(Dir, ?SNAPSHOTS_DIR),
    CheckpointsDir = filename:join(Dir, ?CHECKPOINTS_DIR),
    _ = ra_snapshot:init(UId, SnapModule, SnapshotsDir,
                         CheckpointsDir, undefined, MaxCheckpoints),
    ok.

-spec init(ra_log_init_args()) -> state().
init(#{uid := UId,
       system_config := #{data_dir := DataDir,
                          names := #{wal := Wal,
                                     segment_writer := SegWriter} = Names}
      } = Conf) ->
    Dir = server_data_dir(DataDir, UId),
    MaxOpen = maps:get(max_open_segments, Conf, 5),
    SnapModule = maps:get(snapshot_module, Conf, ?DEFAULT_SNAPSHOT_MODULE),
    %% this has to be patched by ra_server
    LogId = maps:get(log_id, Conf, UId),
    ResendWindow = maps:get(resend_window, Conf, ?DEFAULT_RESEND_WINDOW_SEC),
    SnapInterval = maps:get(min_snapshot_interval, Conf,
                            maps:get(snapshot_interval, Conf,
                                     ?MIN_SNAPSHOT_INTERVAL)),
    CPInterval = maps:get(min_checkpoint_interval, Conf,
                          ?MIN_CHECKPOINT_INTERVAL),
    MaxCheckpoints = maps:get(max_checkpoints, Conf, ?DEFAULT_MAX_CHECKPOINTS),
    SnapshotsDir = filename:join(Dir, ?SNAPSHOTS_DIR),
    CheckpointsDir = filename:join(Dir, ?CHECKPOINTS_DIR),
    Counter = maps:get(counter, Conf, undefined),

    %% ensure directories are there
    ok = ra_lib:make_dir(Dir),
    ok = ra_lib:make_dir(SnapshotsDir),
    ok = ra_lib:make_dir(CheckpointsDir),
    % initialise metrics for this server
    SnapshotState = ra_snapshot:init(UId, SnapModule, SnapshotsDir,
                                     CheckpointsDir, Counter, MaxCheckpoints),
    {SnapIdx, SnapTerm} = case ra_snapshot:current(SnapshotState) of
                              undefined -> {-1, -1};
                              Curr -> Curr
                          end,

    AccessPattern = maps:get(initial_access_pattern, Conf, sequential),
    {ok, Mt0} = ra_log_ets:mem_table_please(Names, UId),
    % recover current range and any references to segments
    % this queries the segment writer and thus blocks until any
    % segments it is currently processed have been finished
    MtRange = ra_mt:range(Mt0),
    {{FirstIdx, LastIdx0}, SegRefs} = case recover_ranges(UId, MtRange, SegWriter) of
                                          {undefined, SRs} ->
                                              {{-1, -1}, SRs};
                                          R ->  R
                                      end,
    %% TODO: don't think this is necessary given the range is calculated from this
    %% but can't hurt as it may trigger some cleanup
    {DeleteSpecs, Mt} = ra_mt:set_first(FirstIdx, Mt0),

    ok = exec_mem_table_delete(Names, UId, DeleteSpecs),
    Reader = ra_log_reader:init(UId, Dir, MaxOpen, AccessPattern, SegRefs,
                                Names, Counter),
    %% assert there is no gap between the snapshot
    %% and the first index in the log
    case (FirstIdx - SnapIdx) > 1 of
        true ->
            exit({corrupt_log, gap_between_snapshot_and_first_index,
                  {SnapIdx, FirstIdx}});
        false -> ok
    end,
    Cfg = #cfg{directory = Dir,
               uid = UId,
               log_id = LogId,
               min_snapshot_interval = SnapInterval,
               min_checkpoint_interval = CPInterval,
               wal = Wal,
               segment_writer = SegWriter,
               resend_window_seconds = ResendWindow,
               snapshot_module = SnapModule,
               counter = Counter,
               names = Names},
    State0 = #?MODULE{cfg = Cfg,
                      first_index = max(SnapIdx + 1, FirstIdx),
                      last_index = max(SnapIdx, LastIdx0),
                      reader = Reader,
                      mem_table = Mt,
                      snapshot_state = SnapshotState,
                      last_wal_write = {whereis(Wal), now_ms()}
                     },
    put_counter(Cfg, ?C_RA_SVR_METRIC_SNAPSHOT_INDEX, SnapIdx),
    LastIdx = State0#?MODULE.last_index,
    put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_INDEX, LastIdx),
    put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_WRITTEN_INDEX, LastIdx),
    put_counter(Cfg, ?C_RA_SVR_METRIC_NUM_SEGMENTS, ra_log_reader:segment_ref_count(Reader)),
    case ra_snapshot:latest_checkpoint(SnapshotState) of
        undefined ->
            ok;
        {ChIdx, _ChTerm} ->
            put_counter(Cfg, ?C_RA_SVR_METRIC_CHECKPOINT_INDEX, ChIdx)
    end,

    % recover the last term
    {LastTerm0, State2} = case LastIdx of
                               SnapIdx ->
                                   {SnapTerm, State0};
                               -1 ->
                                   {0, State0};
                               LI ->
                                   fetch_term(LI, State0)
                           end,
    LastSegRefIdx = case SegRefs of
                        [] ->
                            -1;
                        [{{_, L}, _} | _] ->
                            L
                    end,
    LastWrittenIdx = case ra_log_wal:last_writer_seq(Wal, UId) of
                         {ok, undefined} ->
                             %% take last segref index
                             max(SnapIdx, LastSegRefIdx);
                         {ok, Idx} ->
                             max(Idx, LastSegRefIdx);
                         {error, wal_down} ->
                             ?ERROR("~ts: ra_log:init/1 cannot complete as wal process is down.",
                                    [State2#?MODULE.cfg#cfg.log_id]),
                             exit(wal_down)
                     end,
    {LastWrittenTerm, State3} = case LastWrittenIdx of
                                    SnapIdx ->
                                        {SnapTerm, State2};
                                    _ ->
                                        fetch_term(LastWrittenIdx, State2)
                                end,

    LastTerm = ra_lib:default(LastTerm0, -1),
    State4 = State3#?MODULE{last_term = LastTerm,
                            last_written_index_term =
                                {LastWrittenIdx, LastWrittenTerm}},

    % initialized with a default 0 index 0 term dummy value
    % and an empty meta data map
    State = maybe_append_first_entry(State4),
    ?DEBUG("~ts: ra_log:init recovered last_index_term ~w"
           " snapshot_index_term ~w, last_written_index_term ~w",
           [State#?MODULE.cfg#cfg.log_id,
            last_index_term(State),
            {SnapIdx, SnapTerm},
            State#?MODULE.last_written_index_term
           ]),
    element(1, delete_segments(SnapIdx, State)).

-spec close(state()) -> ok.
close(#?MODULE{cfg = #cfg{uid = _UId},
               reader = Reader}) ->
    % deliberately ignoring return value
    % close all open segments
    _ = ra_log_reader:close(Reader),
    ok.

-spec begin_tx(state()) -> state().
begin_tx(State) ->
    State#?MODULE{tx = true}.

-spec commit_tx(state()) -> {ok, state()} | {error, wal_down, state()}.
commit_tx(#?MODULE{cfg = #cfg{uid = UId,
                              wal = Wal} = Cfg,
                   tx = true,
                   mem_table = Mt1} = State) ->
    {Entries, Mt} = ra_mt:commit(Mt1),
    Tid = ra_mt:tid(Mt),
    WriterId = {UId, self()},
    {WalCommands, Num} =
        lists:foldl(fun ({Idx, Term, Cmd0}, {WC, N}) ->
                            Cmd = {ttb, term_to_iovec(Cmd0)},
                            WalC = {append, WriterId, Tid, Idx, Term, Cmd},
                            {[WalC | WC], N+1}
                    end, {[], 0}, Entries),

    case ra_log_wal:write_batch(Wal, lists:reverse(WalCommands)) of
        {ok, Pid} ->
            ok = incr_counter(Cfg, ?C_RA_LOG_WRITE_OPS, Num),
            {ok, State#?MODULE{tx = false,
                               last_wal_write = {Pid, now_ms()},
                               mem_table = Mt}};
        {error, wal_down} ->
            %% still need to return the state here
            {error, wal_down, State#?MODULE{tx = false,
                                            mem_table = Mt}}
    end;
commit_tx(#?MODULE{tx = false} = State) ->
    State.

-spec append(Entry :: log_entry(), State :: state()) ->
    state() | no_return().
append({Idx, Term, Cmd0} = Entry,
       #?MODULE{cfg = #cfg{uid = UId,
                           wal = Wal} = Cfg,
                last_index = LastIdx,
                tx = false,
                mem_table = Mt0} = State)
      when Idx =:= LastIdx + 1 ->
    case ra_mt:insert(Entry, Mt0) of
        {ok, Mt} ->
            Cmd = {ttb, term_to_iovec(Cmd0)},
            case ra_log_wal:write(Wal, {UId, self()}, ra_mt:tid(Mt),
                                  Idx, Term, Cmd) of
                {ok, Pid} ->
                    ok = incr_counter(Cfg, ?C_RA_LOG_WRITE_OPS, 1),
                    put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_INDEX, Idx),
                    State#?MODULE{last_index = Idx,
                                  last_term = Term,
                                  last_wal_write = {Pid, now_ms()},
                                  mem_table = Mt};
                {error, wal_down} ->
                    error(wal_down)
            end;
        {error, Reason} ->
            ?DEBUG("~ts: mem table ~s detected appending index ~b, "
                   "opening new mem table",
                   [Cfg#cfg.log_id, Reason, Idx]),
            %% this function uses the infinity timeout
            {ok, M0} = ra_log_ets:new_mem_table_please(Cfg#cfg.names,
                                                       Cfg#cfg.uid, Mt0),
            append(Entry, State#?MODULE{mem_table = M0})
    end;
append({Idx, Term, _Cmd} = Entry,
       #?MODULE{cfg = Cfg,
                last_index = LastIdx,
                tx = true,
                mem_table = Mt0} = State)
      when Idx =:= LastIdx + 1 ->
    case ra_mt:stage(Entry, Mt0) of
        {ok, Mt} ->
            put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_INDEX, Idx),
            State#?MODULE{last_index = Idx,
                          last_term = Term,
                          mem_table = Mt};
        {error, Reason} ->
            ?DEBUG("~ts: mem table ~s detected appending index ~b, tx=true "
                   "opening new mem table",
                   [Cfg#cfg.log_id, Reason, Idx]),
            %% this function uses the infinity timeout
            {ok, M0} = ra_log_ets:new_mem_table_please(Cfg#cfg.names,
                                                       Cfg#cfg.uid, Mt0),
            append(Entry, State#?MODULE{mem_table = M0})
    end;
append({Idx, _, _}, #?MODULE{last_index = LastIdx}) ->
    Msg = lists:flatten(io_lib:format("tried writing ~b - expected ~b",
                                      [Idx, LastIdx+1])),
    exit({integrity_error, Msg}).

-spec write(Entries :: [log_entry()], State :: state()) ->
    {ok, state()} |
    {error, {integrity_error, term()} | wal_down}.
write([{FstIdx, _, _} | _Rest] = Entries,
      #?MODULE{cfg = Cfg,
               last_index = LastIdx,
               mem_table = Mt0} = State0)
  when FstIdx =< LastIdx + 1 andalso
       FstIdx >= 0 ->
    case stage_entries(Cfg, Entries, Mt0) of
        {ok, Mt} ->
            wal_write_batch(State0#?MODULE{mem_table = Mt}, Entries);
        Error ->
            Error
    end;
write([], State) ->
    {ok, State};
write([{Idx, _, _} | _], #?MODULE{cfg = #cfg{uid = UId},
                                  last_index = LastIdx}) ->
    Msg = lists:flatten(io_lib:format("~s: ra_log:write/2 "
                                      "tried writing ~b - expected ~b",
                                      [UId, Idx, LastIdx+1])),
    {error, {integrity_error, Msg}}.

-spec fold(FromIdx :: ra_index(), ToIdx :: ra_index(),
           fun((log_entry(), Acc) -> Acc), Acc, state()) ->
    {Acc, state()} when Acc :: term().
fold(From0, To0, Fun, Acc0,
     #?MODULE{cfg = Cfg,
              mem_table = Mt,
              first_index = FirstIdx,
              last_index = LastIdx,
              reader = Reader0} = State)
  when To0 >= From0 andalso
       To0 >= FirstIdx ->
    From = max(From0, FirstIdx),
    To = min(To0, LastIdx),
    ok = incr_counter(Cfg, ?C_RA_LOG_READ_OPS, 1),

    MtOverlap = ra_mt:range_overlap({From, To}, Mt),
    case MtOverlap of
        {undefined, {RemStart, RemEnd}} ->
            {Reader, Acc} = ra_log_reader:fold(RemStart, RemEnd, Fun, Acc0, Reader0),
            {Acc, State#?MODULE{reader = Reader}};
        {{MtStart, MtEnd}, {RemStart, RemEnd}} ->
            {Reader, Acc1} = ra_log_reader:fold(RemStart, RemEnd, Fun, Acc0, Reader0),
            Acc = ra_mt:fold(MtStart, MtEnd, Fun, Acc1, Mt),
            NumRead = MtEnd - MtStart + 1,
            ok = incr_counter(Cfg, ?C_RA_LOG_READ_MEM_TBL, NumRead),
            {Acc, State#?MODULE{reader = Reader}};
        {{MtStart, MtEnd}, undefined} ->
            Acc = ra_mt:fold(MtStart, MtEnd, Fun, Acc0, Mt),
            NumRead = MtEnd - MtStart + 1,
            ok = incr_counter(Cfg, ?C_RA_LOG_READ_MEM_TBL, NumRead),
            {Acc, State}
    end;
fold(_From, _To, _Fun, Acc, State) ->
    {Acc, State}.

%% @doc Reads a list of indexes.
%% Found indexes are returned in the same order as the input list of indexes
%% @end
-spec sparse_read([ra_index()], state()) ->
    {[log_entry()], state()}.
sparse_read(Indexes0, #?MODULE{cfg = Cfg,
                               reader = Reader0,
                               last_index = LastIdx,
                               mem_table = Mt} = State) ->
    ok = incr_counter(Cfg, ?C_RA_LOG_READ_OPS, 1),
    %% indexes need to be sorted high -> low for correct and efficient reading
    Sort = ra_lib:lists_detect_sort(Indexes0),
    Indexes1 = case Sort of
                   unsorted ->
                       lists:sort(fun erlang:'>'/2, Indexes0);
                   ascending ->
                       lists:reverse(Indexes0);
                   _ ->
                       % descending or undefined
                       Indexes0
               end,

    %% drop any indexes that are larger than the last index available
    Indexes2 = lists:dropwhile(fun (I) -> I > LastIdx end, Indexes1),
    {Entries0, MemTblNumRead, Indexes} = ra_mt:get_items(Indexes2, Mt),
    ok = incr_counter(Cfg, ?C_RA_LOG_READ_MEM_TBL, MemTblNumRead),
    {Entries1, Reader} = ra_log_reader:sparse_read(Reader0, Indexes, Entries0),
    %% here we recover the original order of indexes
    Entries = case Sort of
                  descending ->
                      lists:reverse(Entries1);
                  unsorted ->
                      Lookup = lists:foldl(
                                 fun ({I, _, _} = E, Acc) ->
                                         maps:put(I, E, Acc)
                                 end, #{}, Entries1),
                      maps_with_values(Indexes0, Lookup);
                  _ ->
                      %% nothing to do for ascending or undefined
                      Entries1
              end,
    {Entries, State#?MODULE{reader = Reader}}.


%% read a list of indexes,
%% found indexes be returned in the same order as the input list of indexes
-spec partial_read([ra_index()], state(),
                   fun ((ra_index(), ra_term(), ra_server:command()) -> term())
                   ) ->
    read_plan().
partial_read(Indexes0, #?MODULE{cfg = Cfg,
                                reader = Reader0,
                                last_index = LastIdx,
                                mem_table = Mt},
            TransformFun) ->
    ok = incr_counter(Cfg, ?C_RA_LOG_READ_OPS, 1),
    %% indexes need to be sorted high -> low for correct and efficient reading
    Sort = ra_lib:lists_detect_sort(Indexes0),
    Indexes1 = case Sort of
                   unsorted ->
                       lists:sort(fun erlang:'>'/2, Indexes0);
                   ascending ->
                       lists:reverse(Indexes0);
                   _ ->
                       % descending or undefined
                       Indexes0
               end,

    %% drop any indexes that are larger than the last index available
    Indexes2 = lists:dropwhile(fun (I) -> I > LastIdx end, Indexes1),
    {Entries0, MemTblNumRead, Indexes} = ra_mt:get_items(Indexes2, Mt),
    ok = incr_counter(Cfg, ?C_RA_LOG_READ_MEM_TBL, MemTblNumRead),
    Read = lists:foldl(fun ({I, T, Cmd}, Acc) ->
                               maps:put(I, TransformFun(I, T, Cmd), Acc)
                       end, #{}, Entries0),

    Plan = ra_log_reader:read_plan(Reader0, Indexes),
    #read_plan{dir = Cfg#cfg.directory,
               read = Read,
               plan = Plan}.


-spec execute_read_plan(read_plan(), undefined | ra_flru:state(),
                        TransformFun :: transform_fun(),
                        ra_log_reader:read_plan_options()) ->
    {#{ra_index() => Command :: term()}, ra_flru:state()}.
execute_read_plan(#read_plan{dir = Dir,
                             read = Read,
                             plan = Plan}, Flru0, TransformFun,
                  Options) ->
    ra_log_reader:exec_read_plan(Dir, Plan, Flru0, TransformFun,
                                 Options, Read).

-spec read_plan_info(read_plan()) -> map().
read_plan_info(#read_plan{read = Read,
                          plan = Plan}) ->
    NumSegments = length(Plan),
    NumInSegments = lists:foldl(fun ({_, Idxs}, Acc) ->
                                        Acc + length(Idxs)
                                end, 0, Plan),
    #{num_read => map_size(Read),
      num_in_segments => NumInSegments,
      num_segments => NumSegments}.


-spec last_index_term(state()) -> ra_idxterm().
last_index_term(#?MODULE{last_index = LastIdx, last_term = LastTerm}) ->
    {LastIdx, LastTerm}.

-spec last_written(state()) -> ra_idxterm().
last_written(#?MODULE{last_written_index_term = LWTI}) ->
    LWTI.

%% forces the last index and last written index back to a prior index
-spec set_last_index(ra_index(), state()) ->
    {ok, state()} | {not_found, state()}.
set_last_index(Idx, State0) ->
    case fetch_term(Idx, State0) of
        {undefined, State} ->
            case snapshot_index_term(State) of
                {Idx, SnapTerm} ->
                    set_last_index0(Idx, SnapTerm, State);
                _ ->
                    {not_found, State}
            end;
        {Term, State} ->
            set_last_index0(Idx, Term, State)
    end.

set_last_index0(Idx, Term,
                #?MODULE{cfg = Cfg,
                         last_written_index_term = {LWIdx0, _}} = State0) ->
    LWIdx = min(Idx, LWIdx0),
    {LWTerm, State1} = fetch_term(LWIdx, State0),
    %% this should always be found but still assert just in case
    %% _if_ this ends up as a genuine reversal next time we try
    %% to write to the mem table it will detect this and open
    %% a new one
    true = LWTerm =/= undefined,
    put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_INDEX, Idx),
    put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_WRITTEN_INDEX, LWIdx),
    {ok, State1#?MODULE{last_index = Idx,
                        last_term = Term,
                        last_written_index_term = {LWIdx, LWTerm}}}.

-spec handle_event(event_body(), state()) ->
    {state(), [effect()]}.
handle_event({written, _Term, {FromIdx, _ToIdx}},
             #?MODULE{last_index = LastIdx} = State)
  when FromIdx > LastIdx ->
    %% we must have reverted back, either by explicit reset or by a snapshot
    %% installation taking place whilst the WAL was processing the write
    %% Just drop the event in this case as it is stale
    {State, []};
handle_event({written, Term, {FromIdx, ToIdx}},
             #?MODULE{cfg = Cfg,
                      last_written_index_term = {LastWrittenIdx0,
                                                 _LastWrittenTerm0},
                      first_index = FirstIdx} = State0)
  when FromIdx =< LastWrittenIdx0 + 1 ->
    % We need to ignore any written events for the same index
    % but in a prior term if we do not we may end up confirming
    % to a leader writes that have not yet
    % been fully flushed
    case fetch_term(ToIdx, State0) of
        {Term, State} when is_integer(Term) ->
            ok = put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_WRITTEN_INDEX, ToIdx),
            {State#?MODULE{last_written_index_term = {ToIdx, Term}}, []};
        {undefined, State} when ToIdx < FirstIdx ->
            % A snapshot happened before the written event came in
            % This can only happen on a leader when consensus is achieved by
            % followers returning appending the entry and the leader committing
            % and processing a snapshot before the written event comes in.
            {State, []};
        {OtherTerm, State} ->
            %% term mismatch, let's reduce the range and try again to see
            %% if any entries in the range are valid
            case ra_range:new(FromIdx, ToIdx-1) of
                undefined ->
                    ?DEBUG("~ts: written event did not find term ~b for index ~b "
                           "found ~w",
                           [State#?MODULE.cfg#cfg.log_id, Term, ToIdx, OtherTerm]),
                    {State, []};
                NextWrittenRange ->
                    %% retry with a reduced range
                    handle_event({written, Term, NextWrittenRange}, State0)
            end
    end;
handle_event({written, _Term, {FromIdx, _}} = Evt,
             #?MODULE{cfg = #cfg{log_id = LogId},
                      mem_table = Mt,
                      last_written_index_term = {LastWrittenIdx, _}} = State0)
  when FromIdx > LastWrittenIdx + 1 ->
    % leaving a gap is not ok - may need to resend from mem table
    Expected = LastWrittenIdx + 1,
    MtRange = ra_mt:range(Mt),
    case ra_range:in(Expected, MtRange) of
        true ->
            ?INFO("~ts: ra_log: written gap detected at ~b expected ~b!",
                  [LogId, FromIdx, Expected]),
            {resend_from(Expected, State0), []};
        false ->
            ?DEBUG("~ts: ra_log: written gap detected at ~b but is outside
                  of mem table range ~w. Updating last written index to ~b!",
                   [LogId, FromIdx, MtRange, Expected]),
            %% if the entry is not in the mem table we may have missed a
            %% written event due to wal crash. Accept written event by updating
            %% last written index term and recursing
            {Term, State} = fetch_term(Expected, State0),
            handle_event(Evt,
                         State#?MODULE{last_written_index_term = {Expected, Term}})
    end;
handle_event({segments, TidRanges, NewSegs},
             #?MODULE{cfg = #cfg{uid = UId, names = Names} = Cfg,
                      reader = Reader0,
                      mem_table = Mt0,
                      readers = Readers
                     } = State0) ->
    Reader = ra_log_reader:update_segments(NewSegs, Reader0),
    put_counter(Cfg, ?C_RA_SVR_METRIC_NUM_SEGMENTS,
                ra_log_reader:segment_ref_count(Reader)),
    %% the tid ranges arrive in the reverse order they were written
    %% (new -> old) so we need to foldr here to process the oldest first
    Mt = lists:foldr(
           fun ({Tid, Range}, Acc0) ->
                   {Spec, Acc} = ra_mt:record_flushed(Tid, Range, Acc0),
                    ok = ra_log_ets:execute_delete(Names, UId, Spec),
                    Acc
           end, Mt0, TidRanges),
    State = State0#?MODULE{reader = Reader,
                           mem_table = Mt},
    case Readers of
        [] ->
            {State, []};
        _ ->
            %% HACK: but this feature is deprecated anyway
            %% Dummy pid to swallow update notifications
            Pid = spawn(fun () -> ok end),
            {State, log_update_effects(Readers, Pid, State)}
    end;
handle_event({snapshot_written, {SnapIdx, _} = Snap, SnapKind},
             #?MODULE{cfg = #cfg{uid = UId,
                                 names = Names} = Cfg,
                      first_index = FstIdx,
                      last_index = LstIdx,
                      mem_table = Mt0,
                      last_written_index_term = {LastWrittenIdx, _} = LWIdxTerm0,
                      snapshot_state = SnapState0} = State0)
%% only update snapshot if it is newer than the last snapshot
  when SnapIdx >= FstIdx ->
    SnapState1 = ra_snapshot:complete_snapshot(Snap, SnapKind, SnapState0),
    case SnapKind of
        snapshot ->
            put_counter(Cfg, ?C_RA_SVR_METRIC_SNAPSHOT_INDEX, SnapIdx),
            % delete any segments outside of first_index
            {State, Effects0} = delete_segments(SnapIdx, State0),
            %% Delete old snapshot files. This is done as an effect
            %% so that if an old snapshot is still being replicated
            %% the cleanup can be delayed until it is safe.
            DeleteCurrentSnap = {delete_snapshot,
                                 ra_snapshot:directory(SnapState1, snapshot),
                                 ra_snapshot:current(SnapState0)},
            %% Also delete any checkpoints older than this snapshot.
            {SnapState, Checkpoints} =
                ra_snapshot:take_older_checkpoints(SnapIdx, SnapState1),
            CPEffects = [{delete_snapshot,
                          ra_snapshot:directory(SnapState, checkpoint),
                          Checkpoint} || Checkpoint <- Checkpoints],
            Effects1 = [DeleteCurrentSnap | CPEffects] ++ Effects0,

            {LWIdxTerm, Effects} =
                case LastWrittenIdx > SnapIdx of
                    true ->
                        {LWIdxTerm0, Effects1};
                    false ->
                        {Snap, Effects1}
                end,
            %% this will race with the segment writer but if the
            %% segwriter detects a missing index it will query the snaphost
            %% state and if that is higher it will resume flush
            {Spec, Mt1} = ra_mt:set_first(SnapIdx + 1, Mt0),
            ok = exec_mem_table_delete(Names, UId, Spec),

            {State#?MODULE{first_index = SnapIdx + 1,
                           last_index = max(LstIdx, SnapIdx),
                           last_written_index_term = LWIdxTerm,
                           mem_table = Mt1,
                           snapshot_state = SnapState}, Effects};
        checkpoint ->
            put_counter(Cfg, ?C_RA_SVR_METRIC_CHECKPOINT_INDEX, SnapIdx),
            %% If we already have the maximum allowed number of checkpoints,
            %% remove some checkpoints to make space.
            {SnapState, CPs} = ra_snapshot:take_extra_checkpoints(SnapState1),
            Effects = [{delete_snapshot,
                        ra_snapshot:directory(SnapState, SnapKind),
                        CP} || CP <- CPs],
            {State0#?MODULE{snapshot_state = SnapState}, Effects}
    end;
handle_event({snapshot_written, {Idx, Term} = Snap, SnapKind},
             #?MODULE{cfg =#cfg{log_id = LogId},
                      snapshot_state = SnapState} = State0) ->
    %% if the snapshot/checkpoint is stale we just want to delete it
    Current = ra_snapshot:current(SnapState),
    ?INFO("~ts: old snapshot_written received for index ~b in term ~b
          current snapshot ~w, deleting old ~s",
           [LogId, Idx, Term, Current, SnapKind]),
    Effects = [{delete_snapshot,
                ra_snapshot:directory(SnapState, SnapKind),
                Snap}],
    {State0, Effects};
handle_event({resend_write, Idx},
             #?MODULE{cfg =#cfg{log_id = LogId}} = State) ->
    % resend missing entries from mem tables.
    ?INFO("~ts: ra_log: wal requested resend from ~b",
          [LogId, Idx]),
    {resend_from(Idx, State), []};
handle_event({down, Pid, _Info},
             #?MODULE{readers = Readers} =
             State) ->
    {State#?MODULE{readers = lists:delete(Pid, Readers)}, []}.

-spec next_index(state()) -> ra_index().
next_index(#?MODULE{last_index = LastIdx}) ->
    LastIdx + 1.

-spec fetch(ra_index(), state()) ->
    {option(log_entry()), state()}.
fetch(Idx, State0) ->
    case fold(Idx, Idx, fun(E, Acc) -> [E | Acc] end, [], State0) of
        {[], State} ->
            {undefined, State};
        {[Entry], State} ->
            {Entry, State}
    end.

-spec fetch_term(ra_index(), state()) ->
    {option(ra_term()), state()}.
fetch_term(Idx, #?MODULE{last_index = LastIdx,
                         first_index = FirstIdx} = State0)
  when Idx < FirstIdx orelse Idx > LastIdx ->
    {undefined, State0};
fetch_term(Idx, #?MODULE{mem_table = Mt, reader = Reader0} = State0) ->
    case ra_mt:lookup_term(Idx, Mt) of
        undefined ->
            {Term, Reader} = ra_log_reader:fetch_term(Idx, Reader0),
            {Term, State0#?MODULE{reader = Reader}};
        Term when is_integer(Term) ->
            {Term, State0}
    end.

-spec snapshot_state(State :: state()) -> ra_snapshot:state().
snapshot_state(State) ->
    State#?MODULE.?FUNCTION_NAME.

-spec set_snapshot_state(ra_snapshot:state(), state()) -> state().
set_snapshot_state(SnapState, State) ->
    State#?MODULE{snapshot_state = SnapState}.

-spec install_snapshot(ra_idxterm(), ra_snapshot:state(), state()) ->
    {state(), effects()}.
install_snapshot({SnapIdx, SnapTerm} = IdxTerm, SnapState0,
                 #?MODULE{cfg = #cfg{uid = UId,
                                     names = Names} = Cfg,
                          mem_table = Mt0
                         } = State0) ->
    ok = incr_counter(Cfg, ?C_RA_LOG_SNAPSHOTS_INSTALLED, 1),
    ok = put_counter(Cfg, ?C_RA_SVR_METRIC_SNAPSHOT_INDEX, SnapIdx),
    put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_INDEX, SnapIdx),
    put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_WRITTEN_INDEX, SnapIdx),
    {State, Effs} = delete_segments(SnapIdx, State0),
    {SnapState, Checkpoints} =
        ra_snapshot:take_older_checkpoints(SnapIdx, SnapState0),
    CPEffects = [{delete_snapshot,
                  ra_snapshot:directory(SnapState, checkpoint),
                  Checkpoint} || Checkpoint <- Checkpoints],
    {Spec, Mt} = ra_mt:set_first(SnapIdx, Mt0),
    ok = exec_mem_table_delete(Names, UId, Spec),
    {State#?MODULE{snapshot_state = SnapState,
                   first_index = SnapIdx + 1,
                   last_index = SnapIdx,
                   last_term = SnapTerm,
                   mem_table = Mt,
                   last_written_index_term = IdxTerm},
     Effs ++ CPEffects}.

-spec recover_snapshot(State :: state()) ->
    option({ra_snapshot:meta(), term()}).
recover_snapshot(#?MODULE{snapshot_state = SnapState}) ->
    case ra_snapshot:recover(SnapState) of
        {ok, Meta, MacState} ->
            {Meta, MacState};
        {error, no_current_snapshot} ->
            undefined
    end.

-spec snapshot_index_term(State :: state()) -> option(ra_idxterm()).
snapshot_index_term(#?MODULE{snapshot_state = SS}) ->
    ra_snapshot:current(SS).

-spec update_release_cursor(Idx :: ra_index(), Cluster :: ra_cluster(),
                            MacVersion :: ra_machine:version(),
                            MacState :: term(), State :: state()) ->
    {state(), effects()}.
update_release_cursor(Idx, Cluster, MacVersion, MacState, State) ->
    suggest_snapshot(snapshot, Idx, Cluster, MacVersion, MacState, State).

-spec checkpoint(Idx :: ra_index(), Cluster :: ra_cluster(),
                 MacVersion :: ra_machine:version(),
                 MacState :: term(), State :: state()) ->
    {state(), effects()}.
checkpoint(Idx, Cluster, MacVersion, MacState, State) ->
    suggest_snapshot(checkpoint, Idx, Cluster, MacVersion, MacState, State).

suggest_snapshot(SnapKind, Idx, Cluster, MacVersion, MacState,
                 #?MODULE{snapshot_state = SnapshotState} = State) ->
    case ra_snapshot:pending(SnapshotState) of
        undefined ->
            suggest_snapshot0(SnapKind, Idx, Cluster, MacVersion, MacState, State);
        _ ->
            %% Only one snapshot or checkpoint may be written at a time to
            %% prevent excessive I/O usage.
            {State, []}
    end.

promote_checkpoint(Idx, #?MODULE{cfg = Cfg,
                                 snapshot_state = SnapState0} = State) ->
    case ra_snapshot:pending(SnapState0) of
        {_WriterPid, _IdxTerm, snapshot} ->
            %% If we're currently writing a snapshot, skip promoting a
            %% checkpoint.
            {State, []};
        _ ->
            {WasPromoted, SnapState, Effects} =
                ra_snapshot:promote_checkpoint(Idx, SnapState0),
            if WasPromoted ->
                   ok = incr_counter(Cfg, ?C_RA_LOG_CHECKPOINTS_PROMOTED, 1);
               true ->
                   ok
            end,

            {State#?MODULE{snapshot_state = SnapState}, Effects}
    end.

-spec tick(Now :: integer(), state()) -> state().
tick(Now, #?MODULE{cfg = #cfg{wal = Wal},
                   mem_table = Mt,
                   last_written_index_term = {LastWrittenIdx, _},
                   last_wal_write = {WalPid, Ms}} = State) ->
    CurWalPid = whereis(Wal),
    MtRange = ra_mt:range(Mt),
    case Now > Ms + ?WAL_RESEND_TIMEOUT andalso
         is_pid(CurWalPid) andalso
         CurWalPid =/= WalPid andalso
         ra_range:in(LastWrittenIdx + 1, MtRange)
    of
        true ->
            %% the wal has restarted, it has been at least 5s and there are
            %% cached items, resend them
            resend_from(LastWrittenIdx + 1, State);
        false ->
            State
    end.

suggest_snapshot0(SnapKind, Idx, Cluster, MacVersion, MacState, State0) ->
    ClusterServerIds = maps:map(fun (_, V) ->
                                        maps:with([voter_status], V)
                                end, Cluster),
    Meta = #{index => Idx,
             cluster => ClusterServerIds,
             machine_version => MacVersion},
    case should_snapshot(SnapKind, Idx, State0) of
        true ->
            % TODO: here we use the current cluster configuration in
            % the snapshot,
            % _not_ the configuration at the snapshot point.
            % Given cluster changes
            % are applied as they are received I cannot think of any scenarios
            % where this can cause a problem. That said there may
            % well be :dragons: here.
            % The MacState is a reference to the state at
            % the release_cursor point.
            % It can be some dehydrated form of the state itself
            % or a reference for external storage (e.g. ETS table)
            case fetch_term(Idx, State0) of
                {undefined, _} ->
                    {State0, []};
                {Term, State} ->
                    write_snapshot(Meta#{term => Term}, MacState,
                                   SnapKind, State)
            end;
        false ->
            {State0, []}
    end.

should_snapshot(snapshot, Idx,
                #?MODULE{cfg = #cfg{min_snapshot_interval = SnapInter},
                         reader = Reader,
                         snapshot_state = SnapState}) ->
    SnapLimit = case ra_snapshot:current(SnapState) of
                    undefined -> SnapInter;
                    {I, _} -> I + SnapInter
                end,
    % The release cursor index is the last entry _not_ contributing
    % to the current state. I.e. the last entry that can be discarded.
    % We should take a snapshot if the new snapshot index would allow us
    % to discard any segments or if the we've handled enough commands
    % since the last snapshot.
    CanFreeSegments = case ra_log_reader:range(Reader) of
                          undefined ->
                              false;
                          {Start, _End} ->
                              %% this isn't 100% guaranteed to free a segment
                              %% but there is a good chance
                              Idx > Start
                      end,
    CanFreeSegments orelse Idx > SnapLimit;
should_snapshot(checkpoint, Idx,
                #?MODULE{cfg = #cfg{min_checkpoint_interval = CheckpointInter},
                         snapshot_state = SnapState}) ->
    CheckpointLimit = case ra_snapshot:latest_checkpoint(SnapState) of
                          undefined -> CheckpointInter;
                          {I, _} -> I + CheckpointInter
                      end,
    Idx > CheckpointLimit.

-spec append_sync(Entry :: log_entry(), State :: state()) ->
    state() | no_return().
append_sync({Idx, Term, _} = Entry, Log0) ->
    Log = append(Entry, Log0),
    await_written_idx(Idx, Term, Log).

-spec write_sync(Entries :: [log_entry()], State :: state()) ->
    {ok, state()} |
    {error, {integrity_error, term()} | wal_down}.
write_sync(Entries, Log0) ->
    {Idx, Term, _} = lists:last(Entries),
    case ra_log:write(Entries, Log0) of
        {ok, Log} ->
            {ok, await_written_idx(Idx, Term, Log)};
        {error, _} = Err ->
            Err
    end.

-spec can_write(state()) -> boolean().
can_write(#?MODULE{cfg = #cfg{wal = Wal}}) ->
    undefined =/= whereis(Wal).

-spec exists(ra_idxterm(), state()) ->
    {boolean(), state()}.
exists({Idx, Term}, Log0) ->
    case fetch_term(Idx, Log0) of
        {Term, Log} when is_integer(Term) ->
            {true, Log};
        {_, Log} ->
            {false, Log}
    end.

-spec overview(state()) -> overview().
overview(#?MODULE{last_index = LastIndex,
                  last_term = LastTerm,
                  first_index = FirstIndex,
                  last_written_index_term = LWIT,
                  snapshot_state = SnapshotState,
                  reader = Reader,
                  last_wal_write = {_LastPid, LastMs},
                  mem_table = Mt
                 }) ->
    CurrSnap = ra_snapshot:current(SnapshotState),
    #{type => ?MODULE,
      last_index => LastIndex,
      last_term => LastTerm,
      first_index => FirstIndex,
      last_written_index_term => LWIT,
      num_segments => ra_log_reader:segment_ref_count(Reader),
      segments_range => ra_log_reader:range(Reader),
      open_segments => ra_log_reader:num_open_segments(Reader),
      snapshot_index => case CurrSnap of
                            undefined -> undefined;
                            {I, _} -> I
                        end,
      snapshot_term => case CurrSnap of
                            undefined -> undefined;
                            {_, T} -> T
                        end,
      latest_checkpoint_index =>
          case ra_snapshot:latest_checkpoint(SnapshotState) of
              undefined -> undefined;
              {I, _} -> I
          end,
      mem_table_range => ra_mt:range(Mt),
      mem_table_info => ra_mt:info(Mt),
      last_wal_write => LastMs
     }.

-spec write_config(ra_server:config(), state()) -> ok.
write_config(Config0, #?MODULE{cfg = #cfg{directory = Dir}}) ->
    ConfigPath = filename:join(Dir, <<"config">>),
    TmpConfigPath = filename:join(Dir, <<"config.tmp">>),
    % clean config of potentially unserialisable data
    Config = maps:without([parent,
                           counter,
                           has_changed,
                           %% don't write system config to disk as it will
                           %% be updated each time
                           system_config], Config0),
    ok = ra_lib:write_file(TmpConfigPath,
                           list_to_binary(io_lib:format("~p.", [Config]))),
    ok = prim_file:rename(TmpConfigPath, ConfigPath),
    ok.

-spec read_config(state() | file:filename_all()) ->
    {ok, ra_server:config()} | {error, term()}.
read_config(#?MODULE{cfg = #cfg{directory = Dir}}) ->
    read_config(Dir);
read_config(Dir) ->
    ConfigPath = filename:join(Dir, <<"config">>),
    ra_lib:consult(ConfigPath).

-spec delete_everything(state()) -> ok.
delete_everything(#?MODULE{cfg = #cfg{uid = UId,
                                      names = Names,
                                      directory = Dir},
                           snapshot_state = SnapState} = Log) ->
    _ = close(Log),
    %% if there is a snapshot process pending it could cause the directory
    %% deletion to fail, best kill the snapshot process first
    ok = ra_log_ets:delete_mem_tables(Names, UId),
    catch ets:delete(ra_log_snapshot_state, UId),
    case ra_snapshot:pending(SnapState) of
        {Pid, _, _} ->
            case is_process_alive(Pid) of
                true ->
                    exit(Pid, kill),
                    ok;
                false ->
                    ok
            end;
        _ ->
            ok
    end,
    try ra_lib:recursive_delete(Dir) of
        ok -> ok
    catch
        _:_ = Err ->
            ?WARN("ra_log:delete_everything/1 failed to delete "
                  "directory ~ts. Error: ~p", [Dir, Err])
    end,
    ok.

-spec release_resources(non_neg_integer(),
                        sequential | random, state()) -> state().
release_resources(MaxOpenSegments,
                  AccessPattern,
                  #?MODULE{cfg = #cfg{uid = UId,
                                      directory = Dir,
                                      counter = Counter,
                                      names = Names},
                           reader = Reader} = State) ->
    ActiveSegs = ra_log_reader:segment_refs(Reader),
    % close all open segments
    % deliberately ignoring return value
    _ = ra_log_reader:close(Reader),
    %% open a new segment with the new max open segment value
    State#?MODULE{reader = ra_log_reader:init(UId, Dir, MaxOpenSegments,
                                              AccessPattern,
                                              ActiveSegs, Names, Counter)}.

-spec register_reader(pid(), state()) ->
    {state(), effects()}.
register_reader(Pid, #?MODULE{cfg = #cfg{uid = UId,
                                         directory = Dir,
                                         names = Names},
                              reader = Reader,
                              readers = Readers} = State) ->
    SegRefs = ra_log_reader:segment_refs(Reader),
    NewReader = ra_log_reader:init(UId, Dir, 1, SegRefs, Names),
    {State#?MODULE{readers = [Pid | Readers]},
     [{reply, {ok, NewReader}},
      {monitor, process, log, Pid}]}.

readers(#?MODULE{readers = Readers}) ->
    Readers.


%%% Local functions

log_update_effects(Pids, ReplyPid, #?MODULE{first_index = Idx,
                                            reader = Reader}) ->
    SegRefs = ra_log_reader:segment_refs(Reader),
    [{send_msg, P, {ra_log_update, ReplyPid, Idx, SegRefs},
      [ra_event, local]} || P <- Pids].


%% deletes all segments where the last index is lower than
%% the Idx argument
delete_segments(SnapIdx, #?MODULE{cfg = #cfg{log_id = LogId,
                                             segment_writer = SegWriter,
                                             uid = UId} = Cfg,
                                  readers = Readers,
                                  reader = Reader0} = State0) ->
    case ra_log_reader:update_first_index(SnapIdx + 1, Reader0) of
        {Reader, []} ->
            State = State0#?MODULE{reader = Reader},
            {State, log_update_effects(Readers, undefined, State)};
        {Reader, [Pivot | _] = Obsolete} ->
            Pid = spawn(
                    fun () ->
                            ok = log_update_wait_n(length(Readers)),
                            ok = ra_log_segment_writer:truncate_segments(SegWriter,
                                                                         UId, Pivot)
                    end),
            NumActive = ra_log_reader:segment_ref_count(Reader),
            ?DEBUG("~ts: ~b obsolete segments at ~b - remaining: ~b, pivot ~0p",
                   [LogId, length(Obsolete), SnapIdx, NumActive, Pivot]),
            put_counter(Cfg, ?C_RA_SVR_METRIC_NUM_SEGMENTS, NumActive),
            State = State0#?MODULE{reader = Reader},
            {State, log_update_effects(Readers, Pid, State)}
    end.

%% unly used by resend to wal functionality and doesn't update the mem table
wal_rewrite(#?MODULE{cfg = #cfg{uid = UId,
                                wal = Wal} = Cfg} = State,
            Tid, {Idx, Term, Cmd}) ->
    case ra_log_wal:write(Wal, {UId, self()}, Tid, Idx, Term, Cmd) of
        {ok, Pid} ->
            ok = incr_counter(Cfg, ?C_RA_LOG_WRITE_OPS, 1),
            put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_INDEX, Idx),
            State#?MODULE{last_index = Idx,
                          last_term = Term,
                          last_wal_write = {Pid, now_ms()}
                         };
        {error, wal_down} ->
            error(wal_down)
    end.

wal_write_batch(#?MODULE{cfg = #cfg{uid = UId,
                                    wal = Wal} = Cfg,
                         mem_table = Mt0} = State,
                Entries) ->
    WriterId = {UId, self()},
    %% all entries in a transaction are written to the same tid
    Tid = ra_mt:tid(Mt0),
    {WalCommands, Num} =
        lists:foldl(fun ({Idx, Term, Cmd0}, {WC, N}) ->
                            Cmd = {ttb, term_to_iovec(Cmd0)},
                            WalC = {append, WriterId, Tid, Idx, Term, Cmd},
                            {[WalC | WC], N+1}
                    end, {[], 0}, Entries),

    [{_, _, _, LastIdx, LastTerm, _} | _] = WalCommands,
    {_, Mt} = ra_mt:commit(Mt0),
    put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_INDEX, LastIdx),
    ok = incr_counter(Cfg, ?C_RA_LOG_WRITE_OPS, Num),
    case ra_log_wal:write_batch(Wal, lists:reverse(WalCommands)) of
        {ok, Pid} ->
            {ok, State#?MODULE{last_index = LastIdx,
                               last_term = LastTerm,
                               last_wal_write = {Pid, now_ms()},
                               mem_table = Mt}};
        {error, wal_down} = Err ->
            %% if we get there the entry has already been inserted
            %% into the mem table but never reached the wal
            %% the resend logic will take care of that
            Err
    end.

maybe_append_first_entry(State0 = #?MODULE{last_index = -1}) ->
    State = append({0, 0, undefined}, State0),
    receive
        {ra_log_event, {written, 0, {0, 0}}} ->
            ok
    after 60000 ->
              exit({?FUNCTION_NAME, timeout})
    end,
    State#?MODULE{first_index = 0,
                  last_written_index_term = {0, 0}};
maybe_append_first_entry(State) ->
    State.

resend_from(Idx, #?MODULE{cfg = #cfg{uid = UId}} = State0) ->
    try resend_from0(Idx, State0) of
        State -> State
    catch
        error:wal_down ->
            ?WARN("~ts: ra_log: resending from ~b failed with wal_down",
                  [UId, Idx]),
            State0
    end.

resend_from0(Idx, #?MODULE{cfg = Cfg,
                           last_index = LastIdx,
                           last_resend_time = undefined,
                           mem_table = Mt} = State) ->
    ?DEBUG("~ts: ra_log: resending from ~b to ~b",
           [State#?MODULE.cfg#cfg.log_id, Idx, LastIdx]),
    ok = incr_counter(Cfg, ?C_RA_LOG_WRITE_RESENDS, LastIdx - Idx + 1),
    lists:foldl(fun (I, Acc) ->
                        {I, T, C} = ra_mt:lookup(I, Mt),
                        Tid = ra_mt:tid_for(I, T, Mt),
                        wal_rewrite(Acc, Tid, {I, T, C})
                end,
                State#?MODULE{last_resend_time = {erlang:system_time(seconds),
                                                  whereis(Cfg#cfg.wal)}},
                lists:seq(Idx, LastIdx));
resend_from0(Idx, #?MODULE{last_resend_time = {LastResend, WalPid},
                           cfg = #cfg{resend_window_seconds = ResendWindow}} = State) ->
    case erlang:system_time(seconds) > LastResend + ResendWindow orelse
         (is_pid(WalPid) andalso not is_process_alive(WalPid)) of
        true ->
            % it has been more than the resend window since last resend
            % _or_ the wal has been restarted since then
            % ok to try again
            resend_from(Idx, State#?MODULE{last_resend_time = undefined});
        false ->
            State
    end.

stage_entries(Cfg, [Entry | Rem] = Entries, Mt0) ->
    case ra_mt:stage(Entry, Mt0) of
        {ok, Mt} ->
            stage_entries0(Cfg, Rem, Mt);
        {error, OverwritingOrLimitReached} ->
            ?DEBUG("~ts: mem table ~s detected whilst staging entries, opening new mem table",
                   [Cfg#cfg.log_id, OverwritingOrLimitReached]),
            %% TODO: do we need error handling here - this function uses the infinity
            %% timeout
            {ok, Mt} = ra_log_ets:new_mem_table_please(Cfg#cfg.names,
                                                       Cfg#cfg.uid, Mt0),
            stage_entries(Cfg, Entries, Mt)
    end.

stage_entries0(_Cfg, [], Mt) ->
    {ok, Mt};
stage_entries0(Cfg, [Entry | Rem], Mt0) ->
    case ra_mt:stage(Entry, Mt0) of
        {ok, Mt} ->
            stage_entries0(Cfg, Rem, Mt);
        {error, overwriting} ->
            Range = ra_mt:range(Mt0),
            Msg = io_lib:format("ra_log:verify_entries/2 "
                                "tried writing ~p - mem table range ~w",
                                [Rem, Range]),
            {error, {integrity_error, lists:flatten(Msg)}}
    end.



write_snapshot(Meta, MacRef, SnapKind,
               #?MODULE{cfg = Cfg,
                        snapshot_state = SnapState0} = State) ->
    Counter = case SnapKind of
                  snapshot -> ?C_RA_LOG_SNAPSHOTS_WRITTEN;
                  checkpoint -> ?C_RA_LOG_CHECKPOINTS_WRITTEN
              end,
    ok = incr_counter(Cfg, Counter, 1),
    {SnapState, Effects} = ra_snapshot:begin_snapshot(Meta, MacRef, SnapKind,
                                                      SnapState0),
    {State#?MODULE{snapshot_state = SnapState}, Effects}.

recover_ranges(UId, MtRange, SegWriter) ->
    % 1. check mem_tables (this assumes wal has finished recovering
    % which means it is essential that ra_servers are part of the same
    % supervision tree
    % 2. check segments
    SegFiles = ra_log_segment_writer:my_segments(SegWriter, UId),
    SegRefs = lists:foldl(
                fun (File, Acc) ->
                        %% if a server recovered when a segment had been opened
                        %% but never had any entries written the segref would be
                        %% undefined
                        case ra_log_segment:segref(File) of
                            undefined ->
                                Acc;
                            SegRef ->
                                [SegRef | Acc]
                        end
                end, [], SegFiles),
    SegRanges = [Range || {Range, _} <- SegRefs],
    Ranges = [MtRange | SegRanges],
    {pick_range(Ranges, undefined), SegRefs}.

% picks the current range from a sorted (newest to oldest) list of ranges
pick_range([], Res) ->
    Res;
pick_range([H | Tail], undefined) ->
    pick_range(Tail, H);
pick_range([{Fst, _Lst} | Tail], {CurFst, CurLst}) ->
    pick_range(Tail, {min(Fst, CurFst), CurLst}).


%% TODO: implement synchronous writes using gen_batch_server:call/3
await_written_idx(Idx, Term, Log0) ->
    IDX = Idx,
    receive
        {ra_log_event, {written, Term, {_, IDX}} = Evt} ->
            {Log, _} = handle_event(Evt, Log0),
            Log;
        {ra_log_event, {written, _, _} = Evt} ->
            {Log, _} = handle_event(Evt, Log0),
            await_written_idx(Idx, Term, Log)
    after ?LOG_APPEND_TIMEOUT ->
              throw(ra_log_append_timeout)
    end.

log_update_wait_n(0) ->
    ok;
log_update_wait_n(N) ->
    receive
        ra_log_update_processed ->
            log_update_wait_n(N - 1)
    after 1500 ->
              %% just go ahead anyway
              ok
    end.

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.

server_data_dir(Dir, UId) ->
    filename:join(Dir, UId).

maps_with_values(Keys, Map) ->
    lists:foldr(
      fun (K, Acc) ->
              case Map of
                  #{K := Value} ->
                      [Value | Acc];
                  _ ->
                      Acc
              end
      end, [], Keys).

now_ms() ->
    erlang:system_time(millisecond).

exec_mem_table_delete(#{} = Names, UId, Specs)
  when is_list(Specs) ->
    [ra_log_ets:execute_delete(Names, UId, Spec)
     || Spec <- Specs],
    ok.

%%%% TESTS

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").

pick_range_test() ->
    Ranges1 = [{76, 90}, {50, 75}, {1, 100}],
    {1, 90} = pick_range(Ranges1, undefined),

    Ranges2 = [{76, 110}, {50, 75}, {1, 49}],
    {1, 110} = pick_range(Ranges2, undefined),

    Ranges3 = [{25, 30}, {25, 35}, {1, 50}],
    {1, 30} = pick_range(Ranges3, undefined),
    ok.
-endif.