diff --git a/src/eqwalizer_specs.erl b/src/eqwalizer_specs.erl
new file mode 100644
index 00000000..dda428fd
--- /dev/null
+++ b/src/eqwalizer_specs.erl
@@ -0,0 +1,124 @@
+-module(eqwalizer_specs).
+-compile(warn_missing_spec).
+% elp:ignore W0054 (no_nowarn_suppressions)
+-compile([export_all, nowarn_export_all]).
+
+
+-spec 'erpc:call'(node(), module(), atom(), [term()]) -> dynamic().
+'erpc:call'(_, _, _, _) -> error(eqwalizer_specs).
+
+-spec 'lists:foldl'(fun((T, Acc) -> Acc), Acc, [T]) -> Acc.
+'lists:foldl'(_, _, _) -> error(eqwalizer_specs).
+
+-spec 'lists:foldr'(fun((T, Acc) -> Acc), Acc, [T]) -> Acc.
+'lists:foldr'(_, _, _) -> error(eqwalizer_specs).
+
+ -spec 'lists:nth'(pos_integer(), [T]) -> T.
+'lists:nth'(_, _) -> error(eqwalizer_specs).
+
+ -spec 'lists:max'([T]) -> T.
+'lists:max'(_) -> error(eqwalizer_specs).
+
+ -spec 'lists:sort'([T]) -> [T].
+'lists:sort'(_) -> error(eqwalizer_specs).
+
+ -spec 'lists:usort'([T]) -> [T].
+'lists:usort'(_) -> error(eqwalizer_specs).
+
+ -spec 'lists:delete'(T, [T]) -> [T].
+'lists:delete'(_, _) -> error(eqwalizer_specs).
+
+-spec 'lists:sort'(fun((T, T) -> boolean()), [T]) -> [T].
+'lists:sort'(_, _) -> error(eqwalizer_specs).
+
+-spec 'lists:splitwith'(fun((T) -> boolean()), [T]) -> [T].
+'lists:splitwith'(_, _) -> error(eqwalizer_specs).
+
+-spec 'lists:foreach'(fun((T) -> ok), [T]) -> [T].
+'lists:foreach'(_, _) -> error(eqwalizer_specs).
+
+ -spec 'lists:reverse'([T]) -> [T].
+'lists:reverse'(_) -> error(eqwalizer_specs).
+
+ -spec 'lists:reverse'([T], [T]) -> [T].
+'lists:reverse'(_, _) -> error(eqwalizer_specs).
+
+ -spec 'erlang:hd'([A, ...]) -> A.
+'erlang:hd'(_) -> error(eqwalizer_specs).
+
+-spec 'erlang:max'(A, B) -> A | B.
+'erlang:max'(_, _) -> error(eqwalizer_specs).
+
+-spec 'erlang:min'(A, B) -> A | B.
+'erlang:min'(_, _) -> error(eqwalizer_specs).
+
+-spec 'erlang:system_time'() -> pos_integer().
+'erlang:system_time'() -> error(eqwalizer_specs).
+
+-spec 'erlang:system_time'(erlang:time_unit()) -> pos_integer().
+'erlang:system_time'(_) -> error(eqwalizer_specs).
+
+%% -------- gen_statem --------
+
+-spec 'gen_statem:call'(gen_statem:server_ref(), term()) -> dynamic().
+'gen_statem:call'(_, _) -> error(eqwalizer_specs).
+
+-spec 'gen_statem:call'(gen_statem:server_ref(), term(), Timeout) -> dynamic() when
+    Timeout :: timeout() | {clean_timeout, timeout()} | {dirty_timeout, timeout()}.
+'gen_statem:call'(_, _, _) -> error(eqwalizer_specs).
+
+-spec 'gen_statem:receive_response'(gen_statem:request_id(), Timeout) ->
+     dynamic() when
+    Timeout :: timeout() |
+    {clean_timeout, timeout()} |
+    {dirty_timeout, timeout()}.
+'gen_statem:receive_response'(_, _) -> error(eqwalizer_specs).
+
+ %% -------- application --------
+
+-spec 'application:get_all_env'(App :: atom()) -> [{atom(), dynamic()}].
+'application:get_all_env'(_) -> error(eqwalizer_specs).
+
+-spec 'application:get_env'(Param :: atom()) -> undefined | {ok, dynamic()}.
+'application:get_env'(_) -> error(eqwalizer_specs).
+
+-spec 'application:get_env'(App :: atom(), Param :: atom()) ->
+    undefined | {ok, dynamic()}.
+'application:get_env'(_, _) -> error(eqwalizer_specs).
+
+-spec 'application:get_env'(App :: atom(), Param :: atom(), Default :: term()) ->
+    dynamic().
+'application:get_env'(_, _, _) -> error(eqwalizer_specs).
+
+-spec 'application:get_key'(Key :: atom()) -> undefined | {ok, dynamic()}.
+'application:get_key'(_) -> error(eqwalizer_specs).
+
+-spec 'application:get_key'(App :: atom(), Key :: atom()) -> undefined | {ok, dynamic()}.
+'application:get_key'(_, _) -> error(eqwalizer_specs).
+
+-spec 'erlang:binary_to_term'(binary()) -> dynamic().
+'erlang:binary_to_term'(_) -> error(eqwalizer_specs).
+
+ -spec 'ets:info'
+    (ets:table(), compressed | decentralized_counters | fixed | named_table | read_concurrency | write_concurrency) ->
+        boolean();
+    (ets:table(), binary) -> list();
+    (ets:table(), heir) -> pid() | none;
+    (ets:table(), id) -> ets:tid();
+    (ets:table(), keypos | memory | size) -> non_neg_integer();
+    (ets:table(), name) -> atom();
+    (ets:table(), node) -> node();
+    (ets:table(), owner) -> pid();
+    (ets:table(), safe_fixed | safe_fixed_monotonic_time) -> tuple() | false;
+    (ets:table(), stats) -> tuple();
+    (ets:table(), protection) -> ets:table_access();
+    (ets:table(), type) -> ets:table_type().
+'ets:info'(_, _) -> error(eqwalizer_specs).
+
+%% -------- gen_server --------
+
+-spec 'gen_server:call'(gen_server:server_ref(), term()) -> dynamic().
+'gen_server:call'(_, _) -> error(eqwalizer_specs).
+
+-spec 'gen_server:call'(gen_server:server_ref(), term(), timeout()) -> dynamic().
+'gen_server:call'(_, _, _) -> error(eqwalizer_specs).
diff --git a/src/ra.erl b/src/ra.erl
index 6bf6e61d..c25ddb47 100644
--- a/src/ra.erl
+++ b/src/ra.erl
@@ -82,6 +82,7 @@
 -define(START_TIMEOUT, ?DEFAULT_TIMEOUT).
 
 -type ra_cmd_ret() :: ra_server_proc:ra_cmd_ret().
+-type ra_cmd_ret(T) :: ra_server_proc:ra_cmd_ret(T).
 
 -type environment_param() ::
     {data_dir, file:filename()} |
@@ -566,7 +567,7 @@ delete_cluster(ServerIds, Timeout) ->
 %% @end
 -spec add_member(ra_server_id() | [ra_server_id()],
                  ra_server_id() | ra_new_server()) ->
-    ra_cmd_ret() |
+    ra_cmd_ret(ra_idxterm()) |
     {error, already_member} |
     {error, cluster_change_not_permitted}.
 add_member(ServerLoc, ServerId) ->
@@ -578,7 +579,7 @@ add_member(ServerLoc, ServerId) ->
 -spec add_member(ra_server_id() | [ra_server_id()],
                  ra_server_id() | ra_new_server(),
                  timeout()) ->
-    ra_cmd_ret() |
+    ra_cmd_ret(ra_idxterm()) |
     {error, already_member} |
     {error, cluster_change_not_permitted}.
 add_member(ServerLoc, ServerId, Timeout) ->
@@ -661,7 +662,7 @@ leave_and_terminate(System, ServerRef, ServerId) ->
 -spec leave_and_terminate(atom(),
                           ra_server_id() | [ra_server_id()],
                           ra_server_id(), timeout()) ->
-    ok | timeout | {error, noproc | system_not_started}.
+    ok | timeout | {error, nodedown | noproc | system_not_started}.
 leave_and_terminate(System, ServerRef, ServerId, Timeout) ->
     LeaveCmd = {'$ra_leave', ServerId, await_consensus},
     case ra_server_proc:command(ServerRef, LeaveCmd, Timeout) of
@@ -1101,7 +1102,9 @@ members(ServerId, Timeout) ->
 %%
 %% @param ServerId the Ra server(s) to send the query to
 %% @end
--spec members_info(ra_server_id() | [ra_server_id()] | {local, ra_server_id()}) ->
+-spec members_info(ra_server_id() |
+                   [ra_server_id()] |
+                   {local, ra_server_id()}) ->
     ra_server_proc:ra_leader_call_ret(ra_cluster()).
 members_info(ServerId) ->
     members_info(ServerId, ?DEFAULT_TIMEOUT).
@@ -1120,7 +1123,7 @@ members_info(ServerId) ->
 %% @param Timeout the timeout to use
 %% @end
 -spec members_info(ra_server_id() | [ra_server_id()] | {local, ra_server_id()},
-              timeout()) ->
+                   timeout()) ->
     ra_server_proc:ra_leader_call_ret(ra_cluster()).
 members_info({local, ServerId}, Timeout) ->
     ra_server_proc:local_state_query(ServerId, members_info, Timeout);
diff --git a/src/ra.hrl b/src/ra.hrl
index f3b150e0..a52a8665 100644
--- a/src/ra.hrl
+++ b/src/ra.hrl
@@ -13,8 +13,8 @@
 %%
 
 %% taken from gen_statem as this type isn't exported for some reason.
--type from() ::
-	{To :: pid(), Tag :: term()}.
+-type from() :: gen_statem:from().
+	% {To :: pid(), Tag :: term()}.
 
 %% Sections 5.1 in the paper.
 -type ra_index() :: non_neg_integer().
@@ -232,6 +232,8 @@
             true -> ?DISPATCH_LOG(debug, Fmt, Args);
             false -> ok
         end).
+
+% eqwalizer:ignore we know what this is
 -define(DEBUG(Fmt, Args), ?DISPATCH_LOG(debug, Fmt, Args)).
 -define(INFO(Fmt, Args), ?DISPATCH_LOG(info, Fmt, Args)).
 -define(NOTICE(Fmt, Args), ?DISPATCH_LOG(notice, Fmt, Args)).
@@ -243,13 +245,13 @@
 -define(DISPATCH_LOG(Level, Fmt, Args),
         %% same as OTP logger does when using the macro
         try
-            (persistent_term:get('$ra_logger')):log(Level, Fmt, Args,
-                                                    #{mfa => {?MODULE,
-                                                              ?FUNCTION_NAME,
-                                                              ?FUNCTION_ARITY},
-                                                      file => ?FILE,
-                                                      line => ?LINE,
-                                                      domain => [ra]})
+            (ra_env:logger_mod()):log(Level, Fmt, Args,
+                                      #{mfa => {?MODULE,
+                                                ?FUNCTION_NAME,
+                                                ?FUNCTION_ARITY},
+                                        file => ?FILE,
+                                        line => ?LINE,
+                                        domain => [ra]})
         catch
             _:_ -> ok
         end,
diff --git a/src/ra_env.erl b/src/ra_env.erl
index 267ea96e..d5264624 100644
--- a/src/ra_env.erl
+++ b/src/ra_env.erl
@@ -9,21 +9,26 @@
 -export([
          data_dir/0,
          server_data_dir/2,
-         configure_logger/1
+         configure_logger/1,
+         logger_mod/0,
+         log/1
          ]).
 
 -export_type([
               ]).
 
+-spec data_dir() -> file:filename_all().
 data_dir() ->
     DataDir = case application:get_env(ra, data_dir) of
-                  {ok, Dir} ->
+                  {ok, Dir} when is_list(Dir) orelse
+                                 is_binary(Dir) ->
                       Dir;
                   undefined ->
                       {ok, Cwd} = file:get_cwd(),
                       Cwd
               end,
     Node = ra_lib:to_list(node()),
+    % eqwalizer:ignore we can't validate the DataDir any further
     filename:join(DataDir, Node).
 
 server_data_dir(System, UId) when is_atom(System) ->
@@ -32,5 +37,17 @@ server_data_dir(System, UId) when is_atom(System) ->
     filename:join(Dir, Me).
 
 %% use this when interacting with Ra from a node without Ra running on it
-configure_logger(Module) ->
+configure_logger(Module) when is_atom(Module) ->
     persistent_term:put('$ra_logger', Module).
+
+-spec logger_mod() -> module().
+logger_mod() ->
+    case persistent_term:get('$ra_logger', undefined) of
+        M when is_atom(M) ->
+            M;
+        undefined ->
+            ?MODULE
+    end.
+
+%% dummy log function
+log(_) -> ok.
diff --git a/src/ra_flru.erl b/src/ra_flru.erl
index 26955a62..ac4c7c66 100644
--- a/src/ra_flru.erl
+++ b/src/ra_flru.erl
@@ -22,12 +22,12 @@
 
 -define(MAX_SIZE, 5).
 
--type kv_item() :: {Key :: term(), Value :: term()}.
+-type kv_item() :: {Key :: dynamic(), Value :: dynamic()}.
 
 -type handler_fun() :: fun((kv_item()) -> ok).
 
 -record(?MODULE, {max_size = ?MAX_SIZE :: non_neg_integer(),
-                  items = [] :: [term()],
+                  items = [] :: [dynamic()],
                   handler = fun (_) -> ok end :: handler_fun()}).
 
 -opaque state() :: #?MODULE{}.
@@ -43,7 +43,7 @@ new(MaxSize, Handler) ->
              max_size = MaxSize}.
 
 -spec fetch(term(), state()) ->
-    {ok, term(), state()} | error.
+    {ok, dynamic(), state()} | error.
 fetch(Key, #?MODULE{items = [{Key, Value} | _]} = State) ->
     %% head optimisation
     {ok, Value, State};
@@ -55,7 +55,7 @@ fetch(Key, #?MODULE{items = Items0} = State0) ->
             error
     end.
 
--spec insert(term(), term(), state()) -> state().
+-spec insert(dynamic(), dynamic(), state()) -> state().
 insert(Key, Value, #?MODULE{items = Items,
                             max_size = M,
                             handler = Handler} = State)
diff --git a/src/ra_lib.erl b/src/ra_lib.erl
index c3efa614..981f6c75 100644
--- a/src/ra_lib.erl
+++ b/src/ra_lib.erl
@@ -53,7 +53,9 @@
          is_any_file/1,
          ensure_dir/1,
          consult/1,
-         cons/2
+         cons/2,
+         max/2,
+         unwrap/1
         ]).
 
 -type file_err() :: file:posix() | badarg | terminated | system_limit.
@@ -122,11 +124,13 @@ to_binary(I) when is_integer(I) ->
 to_binary(L) when is_list(L) ->
     list_to_binary(L).
 
--spec to_string(binary() | string()) -> string().
+-spec to_string(binary() | string() | atom()) -> string().
 to_string(B) when is_binary(B) ->
     binary_to_list(B);
 to_string(L) when is_list(L) ->
-    L.
+    L;
+to_string(A) when is_atom(A) ->
+    atom_to_list(A).
 
 -spec to_atom(atom() | list() | binary()) -> atom().
 to_atom(A) when is_atom(A) ->
@@ -251,13 +255,15 @@ throw_error(Format, Args) ->
 make_uid() ->
     make_uid(<<>>).
 
--spec make_uid(atom() | binary() | string()) -> binary().
+-spec make_uid(atom() | binary() | string() | unicode:chardata()) ->
+    binary().
 make_uid(Prefix0) ->
     ChrsSize = tuple_size(?GENERATED_UID_CHARS),
     F = fun(_, R) ->
                 [element(rand:uniform(ChrsSize), ?GENERATED_UID_CHARS) | R]
         end,
     Prefix = to_binary(Prefix0),
+    % eqwalizer:ignore
     B = list_to_binary(lists:foldl(F, "", lists:seq(1, ?UID_LENGTH))),
     <<Prefix/binary, B/binary>>.
 
@@ -566,6 +572,16 @@ tokens(Str) ->
             {error, Err}
     end.
 
+-spec max(integer(), integer()) -> integer().
+max(A, B) when is_integer(A) andalso
+               is_integer(B) andalso
+               A < B -> B;
+max(A, _) when is_integer(A) -> A.
+
+unwrap(undefined) ->
+    error(unwrap_unexpected_undefined);
+unwrap(T) ->
+    T.
 
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
@@ -642,7 +658,8 @@ lists_detect_sort_test() ->
 partition_parallel_test() ->
     ?assertMatch({error, {partition_parallel_timeout, [], []}},
                  partition_parallel(fun(_) ->
-                                        timer:sleep(infinity)
+                                        timer:sleep(infinity),
+                                        true
                                     end, [1, 2, 3], 1000)),
     ok.
 
diff --git a/src/ra_log.erl b/src/ra_log.erl
index c0760488..3f1d72f9 100644
--- a/src/ra_log.erl
+++ b/src/ra_log.erl
@@ -26,6 +26,7 @@
          partial_read/3,
          execute_read_plan/4,
          read_plan_info/1,
+         check_read_plan_type/1,
          previous_wal_index/1,
          last_index_term/1,
          set_last_index/2,
@@ -71,30 +72,31 @@
 -type ra_meta_key() :: atom().
 -type segment_ref() :: {File :: binary(), ra_range:range()}.
 -type event_body() :: {written, ra_term(), ra_seq:state()} |
-                      {segments, [{ets:tid(), ra:range()}], [segment_ref()]} |
+                      {segments, [{ets:tid(), ra_seq:state()}], [segment_ref()]} |
                       {resend_write, ra_index()} |
                       {snapshot_written, ra_idxterm(),
                        LiveIndexes :: ra_seq:state(),
                        ra_snapshot:kind(),
                        SnapshotSize :: non_neg_integer() | undefined,
                        Duration :: non_neg_integer()} |
-                      {compaction_result, term()} |
+                      {compaction_result, ra_log_segments:compaction_result()} |
                       major_compaction |
                       {down, pid(), term()}.
 
 -type event() :: {ra_log_event, event_body()}.
--type transform_fun() :: fun ((ra_index(), ra_term(), ra_server:command()) -> term()).
+-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().
+    {monitor, process, log | snapshot_writer, pid()}.
+
 %% 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()].
+-type effects() :: [effect() |
+                    ra_snapshot:effect() |
+                    ra_server:effect()].
 
 -record(cfg, {uid :: ra_uid(),
               log_id :: unicode:chardata(),
@@ -126,7 +128,7 @@
         }).
 
 -record(read_plan, {dir :: file:filename_all(),
-                    read :: #{ra_index() := log_entry()},
+                    read :: #{ra_index() => log_entry()},
                     plan :: ra_log_segments:read_plan()}).
 
 -opaque read_plan() :: #read_plan{}.
@@ -145,7 +147,7 @@
                               max_open_segments => non_neg_integer(),
                               snapshot_module => module(),
                               machine => ra_machine:machine(),
-                              counter => counters:counters_ref(),
+                              counter => option(counters:counters_ref()),
                               initial_access_pattern => sequential | random,
                               max_checkpoints => non_neg_integer(),
                               major_compaction_strategy =>
@@ -312,7 +314,8 @@ init(#{uid := UId,
     LastWalIdx = case ra_log_wal:last_writer_seq(Wal, UId) of
                      {ok, undefined} ->
                          -1;
-                     {ok, Idx} ->
+                     {ok, Idx}
+                       when is_integer(Idx)->
                          Idx;
                      {error, wal_down} ->
                          %% TODO: we could enter a condition loop here to
@@ -325,10 +328,11 @@ init(#{uid := UId,
 
     %% recover the pending seq
     MaxConfirmedWrittenIdx = case TruncSegmentRange of
-                                 {_, LastSegIdx} ->
-                                     max(LastWalIdx, LastSegIdx);
+                                 {_, LastSegIdx}
+                                   when is_integer(LastSegIdx) ->
+                                     ra_lib:max(LastWalIdx, LastSegIdx);
                                  _ ->
-                                     max(LastWalIdx, 0)
+                                     ra_lib:max(LastWalIdx, 0)
                              end,
     Pending = ra_seq:floor(MaxConfirmedWrittenIdx + 1, ra_mt:indexes(Mt)),
     ?DEBUG_IF(Pending =/= [],
@@ -383,7 +387,8 @@ init(#{uid := UId,
     LastSegRefIdx = case TruncSegmentRange of
                         undefined ->
                             -1;
-                        {_, L} ->
+                        {_, L}
+                          when is_integer(L) ->
                             L
                     end,
     LastWrittenIdx = lists:max([LastWalIdx, SnapIdx, LastSegRefIdx]),
@@ -397,7 +402,8 @@ init(#{uid := UId,
     LastTerm = ra_lib:default(LastTerm0, -1),
     State4 = State3#?MODULE{last_term = LastTerm,
                             last_written_index_term =
-                                {LastWrittenIdx, LastWrittenTerm}},
+                                {LastWrittenIdx,
+                                 ra_lib:unwrap(LastWrittenTerm)}},
 
     % initialized with a default 0 index 0 term dummy value
     % and an empty meta data map
@@ -452,7 +458,7 @@ commit_tx(#?MODULE{cfg = #cfg{uid = UId,
                                last_wal_write = {Pid, now_ms(), LastIdx},
                                mem_table = Mt}};
         {error, wal_down} ->
-            {Idx, _, _} = hd(Entries),
+            [{Idx, _, _} | _] = Entries,
             Mt = ra_mt:abort(Mt1),
             %% TODO: review this -  still need to return the state here
             {error, wal_down,
@@ -461,7 +467,7 @@ commit_tx(#?MODULE{cfg = #cfg{uid = UId,
                            mem_table = Mt}}
     end;
 commit_tx(#?MODULE{tx = false} = State) ->
-    State.
+    {ok, State}.
 
 -define(IS_NEXT_IDX(Idx, Range),
         Range == undefined orelse
@@ -603,20 +609,25 @@ write_sparse({Idx, Term, Cmd} = Entry, PrevIdx0,
 
 -spec fold(FromIdx :: ra_index(), ToIdx :: ra_index(),
            fun((log_entry(), Acc) -> Acc), Acc, state()) ->
-    {Acc, state()} when Acc :: term().
+    {Acc, state()}.
 fold(From0, To0, Fun, Acc0, State) ->
-    fold(From0, To0, Fun, Acc0, State, error).
-
--spec fold(FromIdx :: ra_index(), ToIdx :: ra_index(),
-           fun((log_entry(), Acc) -> Acc), Acc, state(),
-            MissingKeyStrategy :: error | return) ->
-    {Acc, state()} when Acc :: term().
-fold(From0, To0, Fun, Acc0,
+    fold(Fun, From0, To0, Acc0, State, error).
+
+-spec fold(fun((log_entry(), Acc) -> Acc),
+           FromIdx :: ra_index(),
+           ToIdx :: ra_index(),
+           Acc,
+           state(),
+           MissingKeyStrategy :: error | return) ->
+    {Acc, state()}.
+fold(Fun, From0, To0, Acc0,
      #?MODULE{cfg = Cfg,
               mem_table = Mt,
               range = {StartIdx, EndIdx},
               reader = Reader0} = State, MissingKeyStrat)
-  when To0 >= From0 andalso
+  when is_integer(From0) andalso
+       is_integer(To0) andalso
+       To0 >= From0 andalso
        To0 >= StartIdx ->
 
     %% TODO: move to ra_range function
@@ -773,8 +784,15 @@ read_plan_info(#read_plan{read = Read,
       num_in_segments => NumInSegments,
       num_segments => NumSegments}.
 
+-spec check_read_plan_type(term()) ->
+    {ok, read_plan()} | error.
+check_read_plan_type(#read_plan{} = Plan) ->
+    {ok, Plan};
+check_read_plan_type(_Term) ->
+    error.
+
 
--spec previous_wal_index(state()) -> ra_idxterm() | -1.
+-spec previous_wal_index(state()) -> ra:index() | -1.
 previous_wal_index(#?MODULE{range = Range}) ->
     case Range of
         undefined ->
@@ -845,7 +863,7 @@ set_last_index(Idx, #?MODULE{cfg = Cfg,
     end.
 
 -spec handle_event(event_body(), state()) ->
-    {state(), [effect()]}.
+    {state(), effects()}.
 handle_event({written, Term, WrittenSeq},
              #?MODULE{cfg = Cfg,
                       last_written_index_term = {PrevIdx, _},
@@ -945,7 +963,7 @@ handle_event({segments, TidSeqs, NewSegs},
                 {_, LastSegIdx} = SegRange,
                 {LWTerm, Reader2} = ra_log_segments:fetch_term(LastSegIdx,
                                                                Reader1),
-                {{LastSegIdx, LWTerm}, Reader2};
+                {{LastSegIdx, ra_lib:unwrap(LWTerm)}, Reader2};
             _ ->
                 {LWIT0, Reader1}
         end,
diff --git a/src/ra_log_meta.erl b/src/ra_log_meta.erl
index e44b65fc..03f40d9e 100644
--- a/src/ra_log_meta.erl
+++ b/src/ra_log_meta.erl
@@ -31,7 +31,7 @@
 -define(TIMEOUT, 30000).
 -define(SYNC_INTERVAL, 5000).
 
--record(?MODULE, {ref :: reference(),
+-record(?MODULE, {ref :: dets:tab_name(),
                   table_name :: atom()}).
 
 -opaque state() :: #?MODULE{}.
@@ -41,6 +41,7 @@
 -spec start_link(ra_system:config()) ->
     {ok, pid()} | {error, {already_started, pid()}}.
 start_link(#{names := #{log_meta := Name}} = Cfg) ->
+    % eqwalizer:ignore
     gen_batch_server:start_link({local, Name}, ?MODULE, Cfg, []).
 
 -spec init(ra_system:config()) -> {ok, state()}.
@@ -50,9 +51,11 @@ init(#{name := System,
     process_flag(trap_exit, true),
     ok = ra_lib:make_dir(Dir),
     MetaFile = filename:join(Dir, "meta.dets"),
+    % eqwalizer:ignore a binary filename works even if the type does
     {ok, Ref} = dets:open_file(TblName, [{file, MetaFile},
                                          {auto_save, ?SYNC_INTERVAL}]),
     _ = ets:new(TblName, [named_table, public, {read_concurrency, true}]),
+    % elp:ignore W0060 (bound_var_in_lhs)
     TblName = dets:to_ets(TblName, TblName),
     ?INFO("ra: meta data store initialised for system ~ts. ~b record(s) recovered",
           [System, ets:info(TblName, size)]),
@@ -61,35 +64,22 @@ init(#{name := System,
 
 handle_batch(Commands, #?MODULE{ref = Ref,
                                 table_name = TblName} = State) ->
-    DoInsert =
-        fun (Id, Key, Value, Inserts0) ->
-                case Inserts0 of
-                    #{Id := Data} ->
-                        Inserts0#{Id => update_key(Key, Value, Data)};
-                    _ ->
-                        case ets:lookup(TblName, Id) of
-                            [Data] ->
-                                Inserts0#{Id => update_key(Key, Value, Data)};
-                            [] ->
-                                Data = {Id, undefined, undefined, undefined},
-                                Inserts0#{Id => update_key(Key, Value, Data)}
-                        end
-                end
-        end,
     {Inserts, Replies, ShouldSync} =
         lists:foldl(
           fun ({cast, {store, Id, Key, Value}},
                {Inserts0, Replies, DoSync}) ->
-                  {DoInsert(Id, Key, Value, Inserts0), Replies, DoSync};
+                  {do_insert(Id, Key, Value, Inserts0, TblName), Replies, DoSync};
               ({call, From, {store, Id, Key, Value}},
-               {Inserts0, Replies, _DoSync}) ->
-                  {DoInsert(Id, Key, Value, Inserts0),
+               {Inserts0, Replies, _DoSync})
+                when is_list(Replies) ->
+                  {do_insert(Id, Key, Value, Inserts0, TblName),
                    [{reply, From, ok} | Replies], true};
               ({cast, {delete, Id}},
                {Inserts0, Replies, DoSync}) ->
                   {handle_delete(TblName, Id, Ref, Inserts0), Replies, DoSync};
               ({call, From, {delete, Id}},
-               {Inserts0, Replies, _DoSync}) ->
+               {Inserts0, Replies, _DoSync})
+                when is_list(Replies) ->
                   {handle_delete(TblName, Id, Ref, Inserts0),
                    [{reply, From, ok} | Replies], true}
           end, {#{}, [], false}, Commands),
@@ -123,6 +113,7 @@ store(Name, UId, Key, Value) when is_atom(Name) ->
 %% when it returns the store request has been safely flushed to disk
 -spec store_sync(atom(), ra_uid(), key(), value()) -> ok.
 store_sync(Name, UId, Key, Value) ->
+    % eqwalizer:ignore
     gen_batch_server:call(Name, {store, UId, Key, Value}, ?TIMEOUT).
 
 -spec delete(atom(), ra_uid()) -> ok.
@@ -131,32 +122,40 @@ delete(Name, UId) ->
 
 -spec delete_sync(atom(), ra_uid()) -> ok.
 delete_sync(Name, UId) ->
+    % eqwalizer:ignore
     gen_batch_server:call(Name, {delete, UId}, ?TIMEOUT).
 
 %% READER API
 
--spec fetch(atom(), ra_uid(), key()) -> value() | undefined.
+-spec fetch(atom(), ra_uid(), current_term) -> ra:index();
+           (atom(), ra_uid(), voted_for) -> option(ra:server_id());
+           (atom(), ra_uid(), last_applied) -> ra:index().
 fetch(MetaName, Id, current_term) ->
-    maybe_fetch(MetaName, Id, 2);
+    maybe_fetch(MetaName, Id, 2, 0);
 fetch(MetaName, Id, voted_for) ->
-    maybe_fetch(MetaName, Id, 3);
+    maybe_fetch(MetaName, Id, 3, undefined);
 fetch(MetaName, Id, last_applied) ->
-    maybe_fetch(MetaName, Id, 4).
-
--spec fetch(atom(), ra_uid(), key(), term()) -> value().
+    maybe_fetch(MetaName, Id, 4, 0).
+
+-spec fetch(atom(), ra_uid(), current_term, Def) ->
+    ra:index() | Def when Def :: term();
+           (atom(), ra_uid(), voted_for, Def) ->
+    ra:server_id() | Def when Def :: term();
+           (atom(), ra_uid(), last_applied, Def) ->
+    ra:index() | Def when Def :: term().
 fetch(MetaName, Id, Key, Default) ->
-    case fetch(MetaName, Id, Key) of
+    case fetch(MetaName, Id, Key, undefined) of
         undefined -> Default;
         Value -> Value
     end.
 
 %%% internal
 
-maybe_fetch(MetaName, Id, Pos) ->
-    try ets:lookup_element(MetaName, Id, Pos)
+maybe_fetch(MetaName, Id, Pos, Def) ->
+    try ets:lookup_element(MetaName, Id, Pos, Def)
     catch
         _:badarg ->
-            undefined
+            Def
     end.
 
 handle_delete(TblName, Id, Ref, Inserts) ->
@@ -178,3 +177,18 @@ update_key(voted_for, Value, Data) ->
     setelement(3, Data, Value);
 update_key(last_applied, Value, Data) ->
     setelement(4, Data, Value).
+
+do_insert(Id, Key, Value, Inserts0, TblName) ->
+    case Inserts0 of
+        #{Id := Data} ->
+            Inserts0#{Id => update_key(Key, Value, Data)};
+        _ ->
+            case ets:lookup(TblName, Id) of
+                [Data] ->
+                    Inserts0#{Id => update_key(Key, Value, Data)};
+                [] ->
+                    Data = {Id, undefined, undefined, undefined},
+                    Inserts0#{Id => update_key(Key, Value, Data)}
+            end
+    end.
+
diff --git a/src/ra_log_segment.erl b/src/ra_log_segment.erl
index 4de6084c..e37fafb4 100644
--- a/src/ra_log_segment.erl
+++ b/src/ra_log_segment.erl
@@ -63,7 +63,7 @@
               max_pending = ?SEGMENT_MAX_PENDING :: non_neg_integer(),
               max_size = ?SEGMENT_MAX_SIZE_B :: non_neg_integer(),
               filename :: file:filename_all(),
-              fd :: option(file:io_device()),
+              fd :: file:io_device(),
               index_size :: pos_integer(),
               index_record_size :: pos_integer(),
               access_pattern :: sequential | random,
@@ -251,7 +251,10 @@ process_file(false, Mode, Filename, Fd, Options) ->
 
 -spec append(state(), ra_index(), ra_term(),
              iodata() | {non_neg_integer(), iodata()}) ->
-    {ok, state()} | {error, full | inet:posix()}.
+    {ok, state()} | {error, full |
+                            file:posix() |
+                            badarg |
+                            terminated}.
 append(#state{cfg = #cfg{max_pending = PendingCount},
               pending_count = PendingCount} = State0,
        Index, Term, Data) ->
@@ -313,14 +316,18 @@ sync(State0) ->
             Err
     end.
 
--spec flush(state()) -> {ok, state()} | {error, term()}.
+-spec flush(state()) -> {ok, state()} |
+                        {error, file:posix() |
+                                badarg |
+                                terminated}.
 flush(#state{cfg = #cfg{fd = Fd},
              pending_data = PendData,
              pending_index = PendIndex,
              index_offset = IdxOffs,
              data_offset = DataOffs,
              index_write_offset = IdxWriteOffs,
-             data_write_offset = DataWriteOffs} = State) ->
+             data_write_offset = DataWriteOffs} = State)
+  when Fd =/= undefined ->
     case file:pwrite(Fd, DataWriteOffs, PendData) of
         ok ->
             case file:pwrite(Fd, IdxWriteOffs, PendIndex) of
@@ -342,7 +349,7 @@ flush(#state{cfg = #cfg{fd = Fd},
            ToIdx :: ra_index(),
            fun((binary()) -> term()),
            fun(({ra_index(), ra_term(), term()}, Acc) -> Acc), Acc) ->
-    Acc when Acc :: term().
+    Acc.
 fold(#state{cfg = #cfg{mode = read}} = State,
      FromIdx, ToIdx, Fun, AccFun, Acc) ->
     fold0(State, FromIdx, ToIdx, Fun, AccFun, Acc, error).
@@ -498,7 +505,7 @@ prepare_cache_binary(Fd, Version, RecSize, NumEntries, IndexBin,
 %% Create cache by reading data range from file
 make_cache(Fd, FstPos, FstLength, LastPos, LastLength) ->
     CacheLen = cache_length(FstPos, FstLength, LastPos, LastLength),
-    {ok, CacheData} = file:pread(Fd, FstPos, CacheLen),
+    {ok, CacheData} = file_pread(Fd, FstPos, CacheLen),
     {FstPos, byte_size(CacheData), CacheData}.
 
 %% Binary search with hint optimization for sequential ascending reads
@@ -702,7 +709,7 @@ data_size(#state{data_offset = DataOffset, data_start = DataStart}) ->
 max_count(#state{cfg = #cfg{max_count = Max}}) ->
     Max.
 
--spec filename(state()) -> file:filename().
+-spec filename(state()) -> file:filename_all().
 filename(#state{cfg = #cfg{filename = Fn}}) ->
     Fn.
 
@@ -796,7 +803,8 @@ is_same_as(#state{cfg = #cfg{filename = Fn0}}, Fn) ->
 -spec close(state()) -> ok | {error, term()}.
 close(#state{cfg = #cfg{fd = Fd,
                         mode = append,
-                        file_advise = FileAdvise}} = State0) ->
+                        file_advise = FileAdvise}} = State0)
+      when Fd =/= undefined ->
     case sync(State0) of
         {ok, State} ->
             %% do not report errors from file:advise, it isn't necessary
@@ -847,7 +855,7 @@ copy_with_cache_loop(SrcFd, SrcIndex, DstState, [Idx | Rem] = Indexes, Cache0)
             case prepare_cache(SrcFd, Indexes, SrcIndex) of
                 undefined ->
                     %% Single entry, no consecutive run - read directly
-                    {ok, Data} = file:pread(SrcFd, Pos, Length),
+                    {ok, Data} = file_pread(SrcFd, Pos, Length),
                     {ok, DstState1} = append_raw(DstState, Idx, Term, Length,
                                                  Data, Crc),
                     copy_with_cache_loop(SrcFd, SrcIndex, DstState1, Rem,
@@ -874,7 +882,7 @@ copy_with_cache_loop(_SrcFd, SrcIndex,
 %% This is used during copy operations where the source CRC is known valid.
 -spec append_raw(state(), ra_index(), ra_term(),
                  non_neg_integer(), binary(), integer()) ->
-    {ok, state()} | {error, full | file:posix()}.
+    {ok, state()} | {error, full | file:posix() | badarg | terminated}.
 append_raw(#state{cfg = #cfg{max_pending = PendingCount},
                   pending_count = PendingCount} = State0,
            Index, Term, Length, Data, Crc) ->
@@ -1139,10 +1147,20 @@ read_header(Fd) ->
             Err
     end.
 
+file_pread(Fd, Pos, Length) ->
+    case file:pread(Fd, Pos, Length) of
+        {ok, Data} when is_binary(Data) ->
+            {ok, Data};
+        {ok, _} ->
+            exit(unexpected_data_type);
+        Err ->
+            Err
+    end.
+
 pread(#cfg{access_pattern = random,
            fd = Fd}, Cache, Pos, Length) ->
     %% no cache
-    {ok, Data} = file:pread(Fd, Pos, Length),
+    {ok, Data} = file_pread(Fd, Pos, Length),
     case byte_size(Data)  of
         Length ->
             {ok, Data, Cache};
@@ -1157,7 +1175,7 @@ pread(#cfg{}, {CPos, CLen, Bin} = Cache, Pos, Length)
 pread(#cfg{access_pattern = sequential,
            fd = Fd} = Cfg, undefined, Pos, Length) ->
     CacheLen = max(Length, ?READ_AHEAD_B),
-    {ok, CacheData} = file:pread(Fd, Pos, CacheLen),
+    {ok, CacheData} = file_pread(Fd, Pos, CacheLen),
     case byte_size(CacheData) >= Length  of
         true ->
             pread(Cfg, {Pos, byte_size(CacheData), CacheData}, Pos, Length);
diff --git a/src/ra_log_segment_writer.erl b/src/ra_log_segment_writer.erl
index 7459a372..a38086e0 100644
--- a/src/ra_log_segment_writer.erl
+++ b/src/ra_log_segment_writer.erl
@@ -67,7 +67,7 @@ start_link(#{name := Name} = Config) ->
 
 -spec accept_mem_tables(atom() | pid(),
                         #{ra_uid() => [{ets:tid(), ra_seq:state()}]},
-                        string()) -> ok.
+                        file:filename_all()) -> ok.
 accept_mem_tables(_SegmentWriter, Tables, undefined)
   when map_size(Tables) == 0 ->
     ok;
diff --git a/src/ra_log_segments.erl b/src/ra_log_segments.erl
index d2bb25ac..467e7678 100644
--- a/src/ra_log_segments.erl
+++ b/src/ra_log_segments.erl
@@ -76,6 +76,7 @@
          compacted_segrefs = [] :: [segment_ref()]}).
 
 -opaque state() :: #?STATE{}.
+-opaque compaction_result() :: #compaction_result{}.
 -type read_plan() :: [{BaseName :: file:filename_all(), [ra:index()]}].
 -type read_plan_options() :: #{access_pattern => random | sequential,
                                file_advise => ra_log_segment:posix_file_advise(),
@@ -85,7 +86,8 @@
               state/0,
               read_plan/0,
               read_plan_options/0,
-              major_compaction_strategy/0
+              major_compaction_strategy/0,
+              compaction_result/0
              ]).
 
 %% PUBLIC
@@ -421,16 +423,19 @@ read_plan(#?STATE{cfg = Cfg,
 -spec exec_read_plan(file:filename_all(),
                      read_plan(),
                      undefined | ra_flru:state(),
-                     TransformFun :: fun((ra_index(), ra_term(), binary()) -> term()),
+                     TransformFun :: fun((ra:index(), ra_term(), dynamic()) -> dynamic()),
                      read_plan_options(),
                      #{ra_index() => Command :: term()}) ->
     {#{ra_index() => Command :: term()}, ra_flru:state()}.
 exec_read_plan(Dir, Plan, undefined, TransformFun, Options, Acc0) ->
-    Open = ra_flru:new(1, fun({_, Seg}) -> ra_log_segment:close(Seg) end),
+    Open = ra_flru:new(1, fun({_, Seg}) ->
+                                  _ = ra_log_segment:close(Seg),
+                                  ok
+                          end),
     exec_read_plan(Dir, Plan, Open, TransformFun, Options, Acc0);
 exec_read_plan(Dir, Plan, Open0, TransformFun, Options, Acc0)
   when is_list(Plan) ->
-    Fun = fun (I, T, B, Acc) ->
+    Fun = fun (I, T, B, Acc) when is_binary(B) ->
                   E = TransformFun(I, T, binary_to_term(B)),
                   Acc#{I => E}
           end,
@@ -457,7 +462,7 @@ exec_read_plan(Dir, Plan, Open0, TransformFun, Options, Acc0)
               end
       end, {Acc0, Open0}, Plan).
 
--spec fetch_term(ra_index(), state()) -> {option(ra_index()), state()}.
+-spec fetch_term(ra_index(), state()) -> {option(ra_term()), state()}.
 fetch_term(Idx, #?STATE{cfg = #cfg{} = Cfg} = State0) ->
     incr_counter(Cfg, ?C_RA_LOG_FETCH_TERM, 1),
     segment_term_query(Idx, State0).
diff --git a/src/ra_log_snapshot.erl b/src/ra_log_snapshot.erl
index f46f4491..b2b4e729 100644
--- a/src/ra_log_snapshot.erl
+++ b/src/ra_log_snapshot.erl
@@ -236,7 +236,7 @@ read_meta_internal(Fd) ->
         {ok, <<?MAGIC, ?VERSION:8/unsigned, Crc:32/integer,
                MetaSize:32/unsigned>>} ->
             case file:read(Fd, MetaSize) of
-                {ok, MetaBin} ->
+                {ok, MetaBin} when is_binary(MetaBin) ->
                     {ok, binary_to_term(MetaBin), Crc};
                 Err ->
                     Err
diff --git a/src/ra_log_sync.erl b/src/ra_log_sync.erl
index 75eb41cb..f215d879 100644
--- a/src/ra_log_sync.erl
+++ b/src/ra_log_sync.erl
@@ -58,6 +58,9 @@ sync({pool, BaseName, PoolSize}, SyncFun) when is_function(SyncFun, 0) ->
 init({}) ->
     {ok, #state{}}.
 
+-spec handle_batch([{call, term(), {sync, fun(() -> ok | {error, term()})}}],
+                   #state{}) ->
+    {ok, [], #state{}}.
 handle_batch(Ops, State) ->
     %% Ops arrive in reverse arrival order (most recent first) because
     %% reversed_batch = true. On journaling filesystems (ext4, XFS),
diff --git a/src/ra_log_wal.erl b/src/ra_log_wal.erl
index 8c1bb3e2..61010a1e 100644
--- a/src/ra_log_wal.erl
+++ b/src/ra_log_wal.erl
@@ -191,7 +191,7 @@ write_batch(Wal, WalCommands) when is_atom(Wal) ->
     case whereis(Wal) of
         undefined ->
             {error, wal_down};
-        Pid ->
+        Pid when is_pid(Pid) ->
             write_batch(Pid, WalCommands)
     end.
 
@@ -208,7 +208,7 @@ last_writer_seq(Wal, UId) when is_atom(Wal) ->
     case whereis(Wal) of
         undefined ->
             {error, wal_down};
-        Pid ->
+        Pid when is_pid(Pid) ->
             last_writer_seq(Pid, UId)
     end.
 
@@ -1109,11 +1109,16 @@ recover_entry(Names, UId, {Idx, _, _} = Entry, SmallestIdx,
             {ok, State#recovery{ranges = Ranges,
                                 writers = Writers#{UId => {in_seq, Idx}},
                                 tables = Tables#{UId => Mt1}}};
-        {error, overwriting} ->
+        {error, Reason} when Reason =:= overwriting;
+                             Reason =:= limit_reached ->
             %% create successor memtable
             {ok, Mt1} = ra_log_ets:new_mem_table_please(Names, UId, Mt0),
             {retry, State#recovery{tables = Tables#{UId => Mt1},
-                                   writers = maps:remove(UId, Writers)}}
+                                   writers = maps:remove(UId, Writers)}};
+        {error, gap_detected} ->
+            ?ERROR("WAL recovery gap detected for ~ts: entry ~p, prev_idx ~p",
+                   [UId, Idx, PrevIdx]),
+            exit({gap_detected, UId, Idx, PrevIdx})
     end;
 recover_entry(Names, UId, {Idx, Term, _}, SmallestIdx,
               #recovery{mode = post_boot,
diff --git a/src/ra_lol.erl b/src/ra_lol.erl
index 82cac3a9..865ac308 100644
--- a/src/ra_lol.erl
+++ b/src/ra_lol.erl
@@ -28,7 +28,7 @@
 %% Threshold at or above which we use tuple, below we use simple list
 -define(TUPLE_THRESHOLD, 65).
 
--type gt_fun() :: fun((Item, Item) -> boolean()).
+-type gt_fun() :: fun((term(), term()) -> boolean()).
 
 %% State is either:
 %% - {list, GtFun, List} where List is stored in descending order (newest first)
diff --git a/src/ra_machine.erl b/src/ra_machine.erl
index 76d728b8..6dfc61ff 100644
--- a/src/ra_machine.erl
+++ b/src/ra_machine.erl
@@ -438,13 +438,14 @@ handle_aux(Mod, RaftState, Type, Cmd, Aux, State) ->
     Mod:handle_aux(RaftState, Type, Cmd, Aux, State).
 
 -spec which_aux_fun(module()) ->
-    undefined | {atom(), arity()}.
+    undefined | {handle_aux, arity()}.
 which_aux_fun(Mod) when is_atom(Mod) ->
     case lists:sort([E || {handle_aux, _Arity} = E
-                          <- erlang:apply(Mod,module_info, [exports])]) of
+                          <- Mod:module_info(exports)]) of
         [] ->
             undefined;
-        [AuxFun | _] ->
+        [{handle_aux, Arity} = AuxFun | _]
+          when is_integer(Arity) ->
             %% favour {handle_aux, 5} as this is the newer api
             AuxFun
     end.
diff --git a/src/ra_monitors.erl b/src/ra_monitors.erl
index f6fba5da..6b50f783 100644
--- a/src/ra_monitors.erl
+++ b/src/ra_monitors.erl
@@ -24,7 +24,8 @@
                      node() => #{component() => ok}}.
 
 -export_type([
-              state/0
+              state/0,
+              component/0
              ]).
 
 -spec init() -> state().
diff --git a/src/ra_mt.erl b/src/ra_mt.erl
index 4d6852a6..eff03e5d 100644
--- a/src/ra_mt.erl
+++ b/src/ra_mt.erl
@@ -72,6 +72,7 @@ init(Tid, Mode) ->
                 read_write ->
                     %% Use ets:select for efficient projection - extracts only indexes
                     %% without building intermediate tuples or function closures
+                    % eqwalizer:ignore
                     ra_seq:from_list(
                         ets:select(Tid, [{{'$1', '_', '_'}, [], ['$1']}]))
             end,
@@ -325,8 +326,8 @@ delete({indexes, _Tid, []}) ->
     0;
 delete({indexes, Tid, Seq}) ->
     NumToDelete = ra_seq:length(Seq),
-    Start = ra_seq:first(Seq),
-    End = ra_seq:last(Seq),
+    {Start, End} = ra_seq:range(Seq),
+    % End = ra_seq:last(Seq),
     Limit = ets:info(Tid, size) div 2,
     %% check if there is an entry below the start of the deletion range,
     %% if there is we've missed a segment event at some point and need
@@ -444,7 +445,7 @@ record_flushed(TID = Tid, FlushedSeq,
                         indexes = Seq} = State) ->
     End = ra_seq:last(FlushedSeq),
     case ra_seq:in(End, Seq) of
-        true ->
+        true when End =/= undefined ->
             %% indexes are always written in order so we can delete
             %% the entire sequence preceeding, this will handle the case
             %% where a segments notifications is missed
@@ -460,7 +461,7 @@ record_flushed(TID = Tid, FlushedSeq,
              State#?MODULE{indexes = NewSeq,
                            size = ra_seq:length(NewSeq),
                            prev = Prev}};
-        false ->
+        _ ->
             {undefined, State}
     end;
 record_flushed(_Tid, _FlushedSeq, #?MODULE{prev = undefined} = State) ->
diff --git a/src/ra_range.erl b/src/ra_range.erl
index 041857b6..6902bcd5 100644
--- a/src/ra_range.erl
+++ b/src/ra_range.erl
@@ -145,7 +145,7 @@ subtract({_SubStart, _SubEnd} = SubRange, {Start, End} = Range) ->
     end.
 
 -spec fold(range(), fun((ra:index(), Acc) -> Acc), Acc) ->
-    Acc when Acc :: term().
+    Acc.
 fold(undefined, _Fun, Acc) ->
     Acc;
 fold({S, E}, Fun, Acc) ->
diff --git a/src/ra_seq.erl b/src/ra_seq.erl
index d8a006d7..d7dcf000 100644
--- a/src/ra_seq.erl
+++ b/src/ra_seq.erl
@@ -9,7 +9,8 @@
 %% sequences are ordered high -> low but ranges are ordered
 %% {low, high} so a typical sequence could look like
 %% [55, {20, 52}, 3]
--type state() :: [ra:index() | ra:range()].
+-type non_empty_state() :: [ra:index() | ra:range(), ...].
+-type state() :: [] | non_empty_state().
 
 -record(i, {seq :: state()}).
 -opaque iter() :: #i{}.
@@ -37,7 +38,8 @@
          in/2,
          range/1,
          in_range/2,
-         has_overlap/2
+         has_overlap/2,
+         from_binary/1
         ]).
 
 -spec append(ra:index(), state()) -> state().
@@ -72,7 +74,7 @@ floor(FloorIdxIncl, Seq) when is_list(Seq) ->
     floor0(FloorIdxIncl, Seq, []).
 
 
--spec limit(ra:index(), state()) -> state().
+-spec limit(ra:index() | undefined, state()) -> state().
 limit(CeilIdxIncl, [Last | Rem])
   when is_integer(Last) andalso
        Last > CeilIdxIncl ->
@@ -102,16 +104,16 @@ add([], To) ->
 add(Add, []) ->
     Add;
 add(Add, To) ->
-    Fst = first(Add),
+    Fst = first0(Add),
     fold(fun append/2, limit(Fst - 1, To), Add).
 
 -spec fold(fun ((ra:index(), Acc) -> Acc), Acc, state()) ->
-    Acc when Acc :: term().
+    Acc.
 fold(Fun, Acc0, Seq) ->
     lists:foldr(
       fun ({_, _} = Range, Acc) ->
               ra_range:fold(Range, Fun, Acc);
-          (Idx, Acc) ->
+          (Idx, Acc) when is_integer(Idx) ->
               Fun(Idx, Acc)
       end, Acc0, Seq).
 
@@ -131,23 +133,13 @@ subtract(SeqA, SeqB) ->
 first([]) ->
     undefined;
 first(Seq) ->
-    case lists:last(Seq) of
-        {I, _} ->
-            I;
-        I ->
-            I
-    end.
+    first0(Seq).
 
 -spec last(state()) -> undefined | ra:index().
 last([]) ->
     undefined;
 last(Seq) ->
-    case hd(Seq) of
-        {_, I} ->
-            I;
-        I ->
-            I
-    end.
+    last0(Seq).
 
 -spec remove_prefix(state(), state()) ->
     {ok, state()} | {error, not_prefix}.
@@ -156,7 +148,7 @@ remove_prefix(Prefix, Seq) ->
     S = iterator(Seq),
     drop_prefix(next(P), next(S)).
 
--spec iterator(state()) -> iter() | end_of_seq.
+-spec iterator(state()) -> iter().
 iterator(Seq) when is_list(Seq) ->
     #i{seq = lists:reverse(Seq)}.
 
@@ -225,7 +217,7 @@ in(Idx, [Range | Rem]) ->
 range([]) ->
     undefined;
 range(Seq) ->
-    ra_range:new(first(Seq), last(Seq)).
+    ra_range:new(first0(Seq), last0(Seq)).
 
 
 -spec in_range(ra:range(), state()) ->
@@ -251,8 +243,29 @@ has_overlap(undefined, _Seq) ->
 has_overlap({Start, End}, Seq) ->
     has_overlap0(Start, End, Seq).
 
+-spec from_binary(binary()) ->
+    {ok, state()} | {error, invalid_format}.
+from_binary(Bin) when is_binary(Bin) ->
+    %% TODO: consider option deep check | validate
+    try binary_to_term(Bin) of
+        [] ->
+            {ok, []};
+        [{A, B} | _] = Seq
+          when is_integer(A) andalso
+               is_integer(B)->
+            {ok, Seq};
+        [I | _] = Seq when is_integer(I) ->
+            {ok, Seq};
+        _ ->
+            {error, invalid_format}
+    catch _:_ ->
+              {error, invalid_format}
+    end.
+
+
 %% Internal functions
 
+
 %% Traverse the sequence (ordered high -> low) checking for overlap.
 %% - Skip elements entirely above End
 %% - Return true if any element overlaps [Start, End]
@@ -319,3 +332,17 @@ floor0(FloorIdx, [{_, _} = T | Rem], Acc) ->
     end;
 floor0(_FloorIdx, _Seq, Acc) ->
     lists:reverse(Acc).
+
+last0([{_, I} | _]) when is_integer(I) ->
+            I;
+last0([I | _]) when is_integer(I) ->
+            I.
+
+first0(Seq) ->
+    case lists:last(Seq) of
+        {I, _} when is_integer(I) ->
+            I;
+        I when is_integer(I) ->
+            I
+    end.
+
diff --git a/src/ra_server.erl b/src/ra_server.erl
index 8adb1381..7ebcf040 100644
--- a/src/ra_server.erl
+++ b/src/ra_server.erl
@@ -68,9 +68,9 @@
         ]).
 
 -type ra_await_condition_fun() ::
-    fun((ra_msg(), ra_server_state()) -> {boolean(), ra_server_state()}).
+    fun((ra_msg(), state()) -> {boolean(), state()}).
 
--type ra_server_state() ::
+-type state() ::
     #{cfg := #cfg{},
       leader_id => option(ra_server_id()),
       cluster := ra_cluster(),
@@ -78,7 +78,7 @@
       cluster_index_term := ra_idxterm(),
       previous_cluster => {ra_index(), ra_term(), ra_cluster()},
       current_term := ra_term(),
-      log := term(),
+      log := ra_log:state(),
       voted_for => option(ra_server_id()), % persistent
       votes => non_neg_integer(),
       membership => ra_membership(),
@@ -111,8 +111,6 @@
       snapshot_has_live_indexes => boolean()
      }.
 
--type state() :: ra_server_state().
-
 -type ra_state() :: leader | follower | candidate
                     | pre_vote | await_condition | delete_and_terminate
                     | terminating_leader | terminating_follower | recover
@@ -152,8 +150,11 @@
                   election_timeout |
                   await_condition_timeout |
                   {command, command()} |
+                  %% this is the external "ra_server_proc" format
+                  {command, low, {command_type(), term(), command_reply_mode()}} |
                   {commands, [command()]} |
                   ra_log:event() |
+                  {aux_command, cast | info | {call, from()}, term()} |
                   {consistent_query, from(), ra:query_fun()} |
                   {consistent_aux, from(), AuxCmd :: term()} |
                   #heartbeat_rpc{} |
@@ -161,12 +162,16 @@
                   #info_reply{} |
                   {ra_server_id, #heartbeat_reply{}} |
                   pipeline_rpcs |
-                  {snapshot_retry_timeout, ra_server_id()}.
+                  {snapshot_retry_timeout, ra_server_id()} |
+                  {transfer_leadership, ra:server_id()} |
+                  force_member_change |
+                  try_become_leader.
 
 -type ra_reply_body() :: #append_entries_reply{} |
                          #request_vote_result{} |
                          #install_snapshot_result{} |
-                         #pre_vote_result{}.
+                         #pre_vote_result{} |
+                         term().
 
 -type effect() ::
     ra_machine:effect() |
@@ -185,10 +190,9 @@
      #heartbeat_rpc{} |
      #info_rpc{}} |
     {send_snapshot, To :: ra_server_id(),
-     {Module :: module(), Ref :: term(),
-      LeaderId :: ra_server_id(), Term :: ra_term()}} |
+     {ra_snapshot:state(), LeaderId :: ra_server_id(), Term :: ra_term()}} |
     {next_event, ra_msg()} |
-    {next_event, cast, ra_msg()} |
+    {next_event, cast | info, ra_msg()} |
     {notify, #{pid() => [term()]}} |
     %% used for tracking valid leader messages
     {record_leader_msg, ra_server_id()} |
@@ -297,7 +301,6 @@
 
 -export_type([state/0,
               config/0,
-              ra_server_state/0,
               ra_state/0,
               ra_server_config/0,
               ra_server_info_key/0,
@@ -320,9 +323,10 @@
 -spec name(ClusterName :: ra_cluster_name(), UniqueSuffix::string()) -> atom().
 name(ClusterName, UniqueSuffix) ->
     % elp:ignore W0023 (atoms_exhaustion)
-    list_to_atom("ra_" ++ ClusterName ++ "_server_" ++ UniqueSuffix).
+    list_to_atom("ra_" ++ ra_lib:to_string(ClusterName) ++
+                 "_server_" ++ UniqueSuffix).
 
--spec init(ra_server_config()) -> ra_server_state().
+-spec init(ra_server_config()) -> state().
 init(#{id := Id,
        uid := UId,
        cluster_name := _ClusterName,
@@ -379,9 +383,9 @@ init(#{id := Id,
     end,
 
     MetaName = meta_name(SystemConfig),
-    CurrentTerm = ra_log_meta:fetch(MetaName, UId, current_term, 0),
-    LastApplied = ra_log_meta:fetch(MetaName, UId, last_applied, 0),
-    VotedFor = ra_log_meta:fetch(MetaName, UId, voted_for, undefined),
+    CurrentTerm = ra_log_meta:fetch(MetaName, UId, current_term),
+    LastApplied = ra_log_meta:fetch(MetaName, UId, last_applied),
+    VotedFor = ra_log_meta:fetch(MetaName, UId, voted_for),
 
     LatestMacVer = ra_machine:version(Machine),
     InitialMachineVersion = min(LatestMacVer,
@@ -405,7 +409,7 @@ init(#{id := Id,
         end,
     MacMod = ra_machine:which_module(Machine, EffectiveMacVer),
 
-    CommitIndex = max(LastApplied, SnapshotIdx),
+    CommitIndex = ra_lib:max(LastApplied, SnapshotIdx),
     MinRecoveryCheckpointInterval = maps:get(min_recovery_checkpoint_interval,
                                              Config, 0),
     Cfg = #cfg{id = Id,
@@ -430,7 +434,7 @@ init(#{id := Id,
     put_counter(Cfg, ?C_RA_SVR_METRIC_EFFECTIVE_MACHINE_VERSION, EffectiveMacVer),
 
     NonVoter = get_membership(Cluster0, Id, UId,
-                             maps:get(membership, Config, voter)),
+                              maps:get(membership, Config, voter)),
 
     #{cfg => Cfg,
       leader_id => undefined,
@@ -496,8 +500,8 @@ recover(#{cfg := #cfg{log_id = LogId,
     ?DEBUG("~ts: scanning for cluster changes ~b:~b ",
            [LogId, FromScan, ToScan]),
     %% if we're recovering after a partial sparse write phase this will fail
-    {{LastScannedIdx, State3}, Log1} = ra_log:fold(FromScan, ToScan,
-                                                   fun cluster_scan_fun/2,
+    {{LastScannedIdx, State3}, Log1} = ra_log:fold(fun cluster_scan_fun/2,
+                                                   FromScan, ToScan,
                                                    {CommitIndex, State2}, Log0,
                                                    return),
 
@@ -514,12 +518,12 @@ recover(#{cfg := #cfg{log_id = LogId,
             end,
 
     put_counter(Cfg, ?C_RA_SVR_METRIC_COMMIT_LATENCY, 0),
-    State#{
-           %% reset commit latency as recovery may calculate a very old value
-           commit_latency => 0}.
 
--spec handle_leader(ra_msg(), ra_server_state()) ->
-    {ra_state(), ra_server_state(), effects()}.
+    %% reset commit latency as recovery may calculate a very old value
+    State#{commit_latency => 0}.
+
+-spec handle_leader(ra_msg(), state()) ->
+    {ra_state(), state(), effects()}.
 handle_leader({PeerId, #append_entries_reply{term = Term, success = true,
                                              next_index = NextIdx,
                                              last_index = LastIdx}},
@@ -587,7 +591,8 @@ handle_leader({PeerId, #append_entries_reply{success = false,
                                              last_index = PeerLastIdx,
                                              last_term = PEER_LAST_TERM}},
               #{cfg := #cfg{log_id = LogId} = Cfg,
-                cluster := Nodes, log := Log0} = State0) ->
+                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,
@@ -1024,8 +1029,8 @@ handle_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()}.
+-spec handle_candidate(ra_msg() | election_timeout, state()) ->
+    {ra_state(), state(), effects()}.
 handle_candidate(#request_vote_result{term = Term, vote_granted = true},
                  #{cfg := #cfg{id = Id,
                                log_id = LogId},
@@ -1171,8 +1176,8 @@ 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()}.
+-spec handle_pre_vote(ra_msg(), state()) ->
+    {ra_state(), state(), effects()}.
 handle_pre_vote(#append_entries_rpc{term = Term} = Msg,
                 #{current_term := CurTerm} = State0)
   when Term >= CurTerm ->
@@ -1262,8 +1267,8 @@ handle_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()}.
+-spec handle_follower(ra_msg(), state()) ->
+    {ra_state(), state(), effects()}.
 handle_follower(#append_entries_rpc{term = Term,
                                     leader_id = LeaderId,
                                     leader_commit = LeaderCommit,
@@ -1596,7 +1601,7 @@ handle_follower(#request_vote_result{}, State) ->
 handle_follower(#pre_vote_result{}, State) ->
     %% handle to avoid logging as unhandled
     {follower, State, []};
-handle_follower(#append_entries_reply{}, 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, []};
@@ -1614,7 +1619,7 @@ handle_follower(force_member_change,
                 #{cfg := #cfg{id = Id,
                               uid = Uid,
                               log_id = LogId}} = State0) ->
-    Cluster = #{Id => new_peer_with(#{voter_status => #{uid => Uid, 
+    Cluster = #{Id => new_peer_with(#{voter_status => #{uid => Uid,
                                                         membership => voter}})},
     ?WARN("~ts: Forcing cluster change. New cluster ~tw",
           [LogId, Cluster]),
@@ -1897,8 +1902,8 @@ abort_receive(#{snapshot_phase := _Phase,
                                   current_event_type],
                                  State#{log => Log})).
 
--spec handle_await_condition(ra_msg(), ra_server_state()) ->
-    {ra_state(), ra_server_state(), effects()}.
+-spec handle_await_condition(ra_msg(), state()) ->
+    {ra_state(), state(), effects()}.
 handle_await_condition(#request_vote_rpc{} = Msg, State) ->
     {follower, State, [{next_event, Msg}]};
 handle_await_condition(#pre_vote_rpc{} = PreVote, State) ->
@@ -1940,8 +1945,8 @@ handle_await_condition(Msg, #{condition := #{predicate_fun := Pred} = Cond} = St
             {await_condition, State, []}
     end.
 
--spec process_new_leader_queries(ra_server_state()) ->
-    {ra_server_state(), [from()]}.
+-spec process_new_leader_queries(state()) ->
+    {state(), [from()]}.
 process_new_leader_queries(#{pending_consistent_queries := Pending,
                              queries_waiting_heartbeats := Waiting} = State0) ->
     From0 = [From || {_, From, _, _} <- Pending],
@@ -1952,14 +1957,14 @@ process_new_leader_queries(#{pending_consistent_queries := Pending,
              queries_waiting_heartbeats => queue:new()},
      From0 ++ From1}.
 
--spec tick(ra_server_state()) -> effects().
+-spec tick(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().
+-spec log_tick(state()) -> state().
 log_tick(#{cfg := #cfg{},
            log := Log0} = State) ->
     Now = erlang:system_time(millisecond),
@@ -1967,8 +1972,8 @@ log_tick(#{cfg := #cfg{},
     State#{log => Log}.
 
 
--spec handle_state_enter(ra_state() | eol, ra_state(), ra_server_state()) ->
-    {ra_server_state() | eol, effects()}.
+-spec handle_state_enter(ra_state() | eol, ra_state(), state()) ->
+    {state(), effects()}.
 handle_state_enter(RaftState, OldRaftState,
                    #{cfg := #cfg{effective_machine_module = MacMod},
                      machine_state := MacState} = State) ->
@@ -1976,7 +1981,7 @@ handle_state_enter(RaftState, OldRaftState,
      ra_machine:state_enter(MacMod, RaftState, MacState)}.
 
 
--spec overview(ra_server_state()) -> map().
+-spec overview(state()) -> map().
 overview(#{cfg := #cfg{effective_machine_module = MacMod} = Cfg,
            log := Log,
            machine_state := MacState,
@@ -2011,17 +2016,17 @@ cfg_to_map(Cfg) ->
                          {N + 1, Acc#{F => element(N, Cfg)}}
                  end, {2, #{}}, record_info(fields, cfg))).
 
--spec is_new(ra_server_state()) -> boolean().
+-spec is_new(state()) -> boolean().
 is_new(#{log := Log}) ->
     ra_log:next_index(Log) =:= 1.
 
--spec is_fully_persisted(ra_server_state()) -> boolean().
+-spec is_fully_persisted(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().
+-spec is_fully_replicated(state()) -> boolean().
 is_fully_replicated(#{commit_index := CI} = State) ->
     case maps:values(peers(State)) of
         [] -> true; % there is only one server
@@ -2102,39 +2107,39 @@ wrap_reply(true, Cmd) ->
 wrap_reply(false, Cmd) ->
     Cmd.
 
--spec id(ra_server_state()) -> ra_server_id().
+-spec id(state()) -> ra_server_id().
 id(#{cfg := #cfg{id = Id}}) -> Id.
 
--spec log_id(ra_server_state()) -> unicode:chardata().
+-spec log_id(state()) -> unicode:chardata().
 log_id(#{cfg := #cfg{log_id = LogId}}) -> LogId.
 
--spec uid(ra_server_state()) -> ra_uid().
+-spec uid(state()) -> ra_uid().
 uid(#{cfg := #cfg{uid = UId}}) -> UId.
 
--spec system_config(ra_server_state()) -> ra_system:config().
+-spec system_config(state()) -> ra_system:config().
 system_config(#{cfg := #cfg{system_config = SC}}) -> SC.
 
--spec leader_id(ra_server_state()) -> option(ra_server_id()).
+-spec leader_id(state()) -> option(ra_server_id()).
 leader_id(State) ->
     maps:get(leader_id, State, undefined).
 
--spec clear_leader_id(ra_server_state()) -> ra_server_state().
+-spec clear_leader_id(state()) -> state().
 clear_leader_id(State) ->
     State#{leader_id => undefined}.
 
--spec current_term(ra_server_state()) -> option(ra_term()).
+-spec current_term(state()) -> option(ra_term()).
 current_term(State) ->
     maps:get(current_term, State).
 
--spec machine_version(ra_server_state()) -> non_neg_integer().
+-spec machine_version(state()) -> non_neg_integer().
 machine_version(#{cfg := #cfg{machine_version = MacVer}}) ->
     MacVer.
 
--spec machine(ra_server_state()) -> ra_machine:machine().
+-spec machine(state()) -> ra_machine:machine().
 machine(#{cfg := #cfg{machine = Machine}}) ->
     Machine.
 
--spec machine_query(ra:query_fun(), ra_server_state()) ->
+-spec machine_query(ra:query_fun(), state()) ->
     {ra_idxterm(), term()}.
 machine_query(QueryFun, #{cfg := #cfg{effective_machine_module = MacMod,
                                       effective_machine_version = MacVer},
@@ -2432,8 +2437,8 @@ log_fold_cache(From, _To, _Cache, Acc) ->
 
 % stores the cluster config at an index such that we can later snapshot
 % at this index.
--spec update_release_cursor(ra_index(), term(), map(), ra_server_state()) ->
-    {ra_server_state(), effects()}.
+-spec update_release_cursor(ra_index(), term(), map(), state()) ->
+    {state(), effects()}.
 update_release_cursor(Index, MacState, #{condition := Conds}, State)
   when is_list(Conds) ->
     case check_release_cursor_conditions(Conds, State) of
@@ -2495,8 +2500,8 @@ maybe_emit_pending_release_cursor(#{pending_release_cursor :=
 maybe_emit_pending_release_cursor(State, Effects) ->
     {State, Effects}.
 
--spec checkpoint(ra_index(), term(), ra_server_state()) ->
-      {ra_server_state(), effects()}.
+-spec checkpoint(ra_index(), term(), state()) ->
+      {state(), effects()}.
 checkpoint(Index, MacState,
            #{cfg := #cfg{machine = Machine},
              log := Log0, cluster := Cluster} = State) ->
@@ -2507,8 +2512,8 @@ checkpoint(Index, MacState,
                                        MacState, Log0),
     {State#{log => Log}, Effects}.
 
--spec promote_checkpoint(ra_index(), ra_server_state()) ->
-    {ra_server_state(), effects()}.
+-spec promote_checkpoint(ra_index(), state()) ->
+    {state(), effects()}.
 promote_checkpoint(Index, #{log := Log0} = State) ->
     {Log, Effects} = ra_log:promote_checkpoint(Index, Log0),
     {State#{log => Log}, Effects}.
@@ -2518,7 +2523,7 @@ promote_checkpoint(Index, #{log := Log0} = State) ->
 % 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().
+-spec persist_last_applied(state()) -> state().
 persist_last_applied(#{persisted_last_applied := PLA,
                        last_applied := LA} = State) when LA =< PLA ->
     % if last applied is less than or equal to PL do nothing
@@ -2547,14 +2552,14 @@ persist_last_applied(#{last_applied := LastApplied0,
                     query_index => non_neg_integer(),
                     commit_index_sent => non_neg_integer(),
                     status => ra_peer_status()},
-                  ra_server_state()) -> ra_server_state().
+                  state()) -> state().
 update_peer(PeerId, Update, #{cluster := Peers} = State)
   when is_map(Update) ->
     Peer = maps:merge(maps:get(PeerId, Peers), Update),
     put_peer(PeerId, Peer, State).
 
--spec update_disconnected_peers(node(), nodeup | nodedown, ra_server_state()) ->
-    ra_server_state().
+-spec update_disconnected_peers(node(), nodeup | nodedown, state()) ->
+    state().
 update_disconnected_peers(Node, nodeup, #{cluster := Peers} = State) ->
     State#{cluster => maps:map(
                         fun ({_, PeerNode}, #{status := disconnected} = Peer)
@@ -2602,9 +2607,9 @@ peer_snapshot_process_exited_with_backoff(SnapshotPid, #{cluster := Peers} = Sta
     end.
 
 -spec handle_down(ra_state(),
-                  machine | snapshot_sender | snapshot_writer | aux,
-                  pid(), term(), ra_server_state()) ->
-    {ra_state(), ra_server_state(), effects()}.
+                  machine | snapshot_sender | snapshot_writer | aux | log,
+                  pid(), term(), state()) ->
+    {ra_state(), state(), effects()}.
 handle_down(leader, machine, Pid, Info, State)
   when is_pid(Pid) ->
     % %% commit command to be processed by state machine
@@ -2651,10 +2656,10 @@ handle_down(RaftState, Type, Pid, Info, #{cfg := #cfg{log_id = LogId}} = State)
            [LogId, Type, Pid, Info, 10]),
     {RaftState, State, []}.
 
--spec handle_node_status(ra_state(), machine | aux,
+-spec handle_node_status(ra_state(), ra_monitors:component(),
                          node(), nodeup | nodedown,
-                         term(), ra_server_state()) ->
-    {ra_state(), ra_server_state(), effects()}.
+                         term(), state()) ->
+    {ra_state(), state(), effects()}.
 handle_node_status(leader, machine, Node, Status, _Infos, State)
   when is_atom(Node) ->
     %% commit command to be processed by state machine
@@ -2671,7 +2676,7 @@ handle_node_status(RaftState, Type, Node, Status, _Info,
           [LogId, Type, Node, Status]),
     {RaftState, State, []}.
 
--spec terminate(ra_server_state(), Reason :: {shutdown, delete} | term()) -> ok.
+-spec terminate(state(), Reason :: {shutdown, delete} | term()) -> ok.
 terminate(#{log := Log,
             cfg := #cfg{log_id = LogId}} = _State, {shutdown, delete}) ->
     ?NOTICE("~ts: terminating with reason 'delete'", [LogId]),
@@ -2808,9 +2813,9 @@ maybe_recover_from_recovery_checkpoint(LastApplied, CommitIndex, SnapState,
             {LastApplied, State}
     end.
 
--spec log_fold(ra_server_state(), fun((term(), State) -> State), State) ->
-    {ok, State, ra_server_state()} |
-    {error, term(), ra_server_state()}.
+-spec log_fold(state(), fun((term(), State) -> State), State) ->
+    {ok, State, state()} |
+    {error, term(), state()}.
 log_fold(#{log := Log} = RaState, Fun, State) ->
     Idx = case ra_log:snapshot_index_term(Log) of
               {PrevIdx, _PrevTerm} ->
@@ -2826,9 +2831,9 @@ log_fold(#{log := Log} = RaState, Fun, State) ->
     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()}.
+-spec log_read([ra_index()], state()) ->
+    {ok, [term()], state()} |
+    {error, state()}.
 log_read(Indexes, #{log := Log0} = State) ->
     {Entries, Log} = ra_log:sparse_read(Indexes, Log0),
     {ok, [Data
@@ -2836,7 +2841,7 @@ log_read(Indexes, #{log := Log0} = State) ->
      State#{log => Log}}.
 
 %% reads user commands at the specified index
--spec log_partial_read([ra_index()], ra_server_state()) ->
+-spec log_partial_read([ra_index()], state()) ->
     ra_log:read_plan().
 log_partial_read(Indexes, #{log := Log0}) ->
     ra_log:partial_read(Indexes, Log0,
@@ -3887,7 +3892,7 @@ already_member(State, Effects) ->
 %%% Voter status helpers
 %%% ====================
 
--spec ensure_promotion_target(ra_voter_status(), ra_server_state()) ->
+-spec ensure_promotion_target(ra_voter_status(), state()) ->
     {ok, ra_voter_status()} | {error, term()}.
 ensure_promotion_target(#{membership := promotable, target := _, uid := _} = Status,
                         _) ->
@@ -3920,7 +3925,7 @@ get_membership(Cluster, PeerId, UId, Default) ->
             Default
     end.
 
--spec get_condition_timeout(ra_server_state(), Default :: term()) ->
+-spec get_condition_timeout(state(), Default :: term()) ->
     term() | integer().
 get_condition_timeout(#{condition := #{timeout := #{duration := D}}}, _Def) ->
     D;
@@ -3931,7 +3936,7 @@ get_condition_timeout(_, 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()) ->
+                    state()) ->
     ra_membership().
 get_membership(Cluster, #{cfg := #cfg{id = Id, uid = UId}} = State) ->
     Default = maps:get(membership, State, voter),
@@ -3939,12 +3944,12 @@ get_membership(Cluster, #{cfg := #cfg{id = Id, uid = UId}} = State) ->
 
 %% Get this node's membership.
 %% Defaults to last known-locally value.
--spec get_membership(ra_server_state()) -> ra_membership().
+-spec get_membership(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()) -> 
+-spec maybe_promote_peer(ra_server_id(), state(), effects()) ->
     effects().
 maybe_promote_peer(PeerId, #{cluster := Cluster}, Effects) ->
     case Cluster of
diff --git a/src/ra_server_proc.erl b/src/ra_server_proc.erl
index 391fb03a..a5b62d06 100644
--- a/src/ra_server_proc.erl
+++ b/src/ra_server_proc.erl
@@ -101,6 +101,7 @@
                                    {timeout, ra_server_id()}.
 
 -type ra_cmd_ret() :: ra_leader_call_ret(term()).
+-type ra_cmd_ret(T) :: ra_leader_call_ret(T).
 
 -type gen_statem_start_ret() :: {ok, pid()} | ignore | {error, term()}.
 
@@ -125,6 +126,7 @@
 -export_type([ra_leader_call_ret/1,
               ra_local_call_ret/1,
               ra_cmd_ret/0,
+              ra_cmd_ret/1,
               safe_call_ret/1,
               ra_event_reject_detail/0,
               ra_event/0,
@@ -152,7 +154,7 @@
               }).
 
 -record(state, {conf :: #conf{},
-                server_state :: ra_server:ra_server_state(),
+                server_state :: ra_server:state(),
                 monitors = ra_monitors:init() :: ra_monitors:state(),
                 pending_commands = [] :: [{{pid(), any()}, term()}],
                 leader_monitor :: reference() | undefined,
@@ -178,7 +180,7 @@ start_link(Config = #{id := Id}) ->
     gen_statem:start_link({local, Name}, ?MODULE, Config, []).
 
 -spec command(server_loc(), ra_command(), timeout()) ->
-    ra_cmd_ret().
+    ra_cmd_ret(dynamic()).
 command(ServerLoc, Cmd, Timeout) ->
     leader_call(ServerLoc, {command, normal, Cmd}, Timeout).
 
@@ -194,9 +196,9 @@ cast_command(ServerId, Priority, Cmd) ->
             local | consistent | consistent_aux | leader,
             query_options(),
             timeout()) ->
-    ra_server_proc:ra_leader_call_ret({ra_idxterm(), Reply :: term()})
-    | ra_server_proc:ra_leader_call_ret(Reply :: term())
-    | {ok, {ra_idxterm(), Reply :: term()}, not_known}.
+    ra_server_proc:ra_leader_call_ret({ra_idxterm(), Reply :: dynamic()})
+    | ra_server_proc:ra_leader_call_ret(Reply :: dynamic())
+    | {ok, {ra_idxterm(), Reply :: dynamic()}, not_known}.
 query(ServerLoc, QueryFun, local, Options, Timeout)
   when map_size(Options) =:= 0 ->
     statem_call(ServerLoc, {local_query, QueryFun}, Timeout);
@@ -242,7 +244,7 @@ read_entries({_, Node} = ServerId, Indexes, Flru0, Timeout)
                   members_info |
                   initial_members |
                   machine, timeout()) ->
-    ra_leader_call_ret(term()).
+    ra_leader_call_ret(dynamic()).
 state_query(ServerLoc, Spec, Timeout) ->
     leader_call(ServerLoc, {state_query, Spec}, Timeout).
 
@@ -254,7 +256,7 @@ state_query(ServerLoc, Spec, Timeout) ->
                         members_info |
                         initial_members |
                         machine, timeout()) ->
-    ra_local_call_ret(term()).
+    ra_local_call_ret(dynamic()).
 local_state_query(ServerLoc, Spec, Timeout) ->
     local_call(ServerLoc, {state_query, Spec}, Timeout).
 
@@ -381,12 +383,18 @@ do_init(#{id := Id,
     ReceiveSnapshotTimeout = maps:get(receive_snapshot_timeout, SysConf,
                                       ?DEFAULT_RECEIVE_SNAPSHOT_TIMEOUT),
     % elp:ignore W0011 (application_get_env)
-    AtenPollInt = application:get_env(aten, poll_interval, 1000),
+    AtenPollInt = case application:get_env(aten, poll_interval, undefined) of
+                      undefined ->
+                          1000;
+                      PollInt when is_integer(PollInt) ->
+                          PollInt
+                  end,
     LogId = ra_server:log_id(ServerState),
     %% TODO: full error handling
     WorkerPid = case ra_server_sup:start_ra_worker(ParentPid, Config) of
-                    {ok, P} -> P;
-                    {error, {already_started, P}} ->
+                    {ok, P} when is_pid(P) -> P;
+                    {error, {already_started, P}}
+                      when is_pid(P) ->
                         P
                 end,
     ra_env:configure_logger(logger),
@@ -2245,6 +2253,19 @@ send_pre_snapshot_entries(Id, To, RPC, Indexes, InstallTimeout, Flru0) ->
                                        InstallTimeout, Flru1)
     end.
 
+% -spec receive_read_plan(term()) -> ra_log:read_plan().
+% receive_read_plan(ReqId) ->
+%     {ok, ReadPlan} = case gen_statem:receive_response(ReqId, infinity) of
+%                          {reply, {ok, RP}} ->
+%                              {ok, RP};
+%                          {error, Reason} ->
+%                              error(Reason);
+%                          timeout ->
+%                              error(read_entries_timeout)
+%                      end,
+%     {ok, Plan} = ra_log:check_read_plan_type(ReadPlan),
+%     Plan.
+
 send_pre_snapshot_entries0(Id, To, RPC, SeqIter, CurrentEnts,
                            InstallTimeout, Flru0) ->
     case ra_seq:list_chunk(16, SeqIter) of
@@ -2273,6 +2294,7 @@ send_pre_snapshot_entries0(Id, To, RPC, SeqIter, CurrentEnts,
                                  timeout ->
                                      error(read_entries_timeout)
                              end,
+            % ReadPlan = receive_read_plan(ReqId),
             %% Execute the read plan to get actual entries
             {Reads, Flru1} = ra_log:execute_read_plan(ReadPlan, Flru0,
                                                       fun (Idx, Term, Cmd) ->
@@ -2454,6 +2476,7 @@ maybe_record_cluster_change(#state{conf = #conf{cluster_name = ClusterName},
         map_get(cluster_index_term, ServerStateB) orelse
         LeaderA =/= LeaderB) ->
             MembersB = ra_server:state_query(members, ServerStateB),
+            %% TODO: can we ever record undefined here?
             ok = ra_leaderboard:record(ClusterName, LeaderB, MembersB);
         true ->
             ok
diff --git a/src/ra_snapshot.erl b/src/ra_snapshot.erl
index e0a982dd..68b961fa 100644
--- a/src/ra_snapshot.erl
+++ b/src/ra_snapshot.erl
@@ -1013,7 +1013,7 @@ take_extra_checkpoints(#?MODULE{checkpoints = Checkpoints0,
 %% Data             (binary - term_to_binary of indexes)
 %% @end
 -spec write_indexes(file:filename_all(), ra_seq:state()) ->
-    ok | {error, file:posix()}.
+    ok | {error, ra_lib:file_err()}.
 write_indexes(Dir, Indexes) ->
     File = filename:join(Dir, <<"indexes">>),
     Data = term_to_binary(Indexes),
@@ -1037,21 +1037,14 @@ indexes(Dir) ->
         {ok, <<?IDX_MAGIC, ?IDX_VERSION:8/unsigned, Crc:32/unsigned, Data/binary>>} ->
             case erlang:crc32(Data) of
                 Crc ->
-                    {ok, binary_to_term(Data)};
+                    ra_seq:from_binary(Data);
                 _ ->
                     {error, checksum_error}
             end;
         {ok, <<?IDX_MAGIC, Version:8/unsigned, _Crc:32/unsigned, _Data/binary>>} ->
             {error, {invalid_version, Version}};
-        {ok, Bin} ->
-            %% Backward compatibility: old format without header
-            %% Try to parse as plain term_to_binary data
-            try
-                {ok, binary_to_term(Bin)}
-            catch
-                _:_ ->
-                    {error, invalid_format}
-            end;
+        {ok, _Bin} ->
+            {error, invalid_format};
         {error, enoent} ->
             %% no indexes
             {ok, []};
diff --git a/test/ra_log_wal_SUITE.erl b/test/ra_log_wal_SUITE.erl
index e351384f..43d58cbf 100644
--- a/test/ra_log_wal_SUITE.erl
+++ b/test/ra_log_wal_SUITE.erl
@@ -41,6 +41,7 @@ all_tests() ->
      overwrite_inside,
      recover,
      recover_overwrite,
+     recover_limit_reached,
      recover_with_snapshot_index,
      recover_overwrite_rollover,
      recover_existing_mem_table,
@@ -1061,6 +1062,55 @@ recover_overwrite(Config) ->
     proc_lib:stop(Pid2),
     ok.
 
+recover_limit_reached(Config) ->
+    ok = logger:set_primary_config(level, all),
+    Conf = ?config(wal_conf, Config),
+    {UId, _} = WriterId = ?config(writer_id, Config),
+    Data = <<42:256/unit:8>>,
+    meck:new(ra_log_segment_writer, [passthrough]),
+    meck:expect(ra_log_segment_writer, await, fun(_) -> ok end),
+    Tid = ets:new(?FUNCTION_NAME, []),
+    {ok, Pid} = ra_log_wal:start_link(Conf),
+    %% write some in one term
+    [{ok, _} = ra_log_wal:write(Pid, WriterId, Tid, Idx, 1, Data)
+     || Idx <- lists:seq(1, 10)],
+    _ = await_written(WriterId, 1, [{1, 10}]),
+
+    flush(),
+    ok = proc_lib:stop(ra_log_wal, normal, 5000),
+    
+    meck:new(ra_mt, [passthrough]),
+    meck:expect(ra_mt, insert_sparse,
+                fun({Idx, _, _} = Entry, LastIdx, State) ->
+                        case Idx == 5 andalso get(limit_reached_returned) == undefined of
+                            true ->
+                               put(limit_reached_returned, true),
+                               {error, limit_reached};
+                            false ->
+                               meck:passthrough([Entry, LastIdx, State])
+                        end
+                end),
+
+    {ok, Pid2} = ra_log_wal:start_link(Conf),
+    {ok, Mt} = ra_log_ets:mem_table_please(?config(names, Config), UId),
+
+    ?assertMatch({1, 10}, ra_mt:range(Mt)),
+    MtTid = ra_mt:tid(Mt),
+    PrevMtTid = ra_mt:tid(ra_mt:prev(Mt)),
+    receive
+        {'$gen_cast',
+         {mem_tables, #{UId := [{MtTid, [{5, 10}]},
+                                {PrevMtTid, [{1, 4}]}
+                                ]}, _Wal}} ->
+            ok
+    after 2000 ->
+              flush(),
+              ct:fail("new_mem_tables_timeout")
+    end,
+    meck:unload(),
+    proc_lib:stop(Pid2),
+    ok.
+
 recover_overwrite_rollover(Config) ->
     ok = logger:set_primary_config(level, all),
     Conf = ?config(wal_conf, Config),