fix dialyzer errors

Author: andrewjstone

Makefile

@@ -1,7 +1,9 @@
-.PHONY: test
+.PHONY: test deps
 
 rafter:
 	./rebar compile
+deps:
+	./rebar get-deps
 test:
 	./rebar eunit skip_deps=true
 

include/rafter_consensus_fsm.hrl

@@ -13,10 +13,10 @@
     term = 0 :: non_neg_integer(),
     voted_for :: term(),
     commit_index = 0 :: non_neg_integer(),
-    init_config :: undefined | list() | complete,
+    init_config :: undefined | list() | complete | no_client,
 
     %% Used for Election and Heartbeat timeouts
-    timer :: timer:tref(),
+    timer :: reference(),
 
     %% leader state: contains nextIndex for each peer
     followers = dict:new() :: dict(),

rebar.config

@@ -4,6 +4,6 @@
 
 {erl_opts, [debug_info, warnings_as_errors, {parse_transform, lager_transform}]}.
 
-{deps, [{lager, ".*", {git, "git://github.com/basho/lager", {tag, "2.0.0rc2"}}},
+{deps, [{lager, ".*", {git, "git://github.com/basho/lager", {tag, "2.0.1"}}},
         {druuid, ".*", {git, "git://github.com/kellymclaughlin/druuid.git", {tag, "0.2"}}} 
 ]}.

src/rafter.erl

@@ -6,7 +6,7 @@
 
 %% API
 -export([start_node/2, stop_node/1, op/2, read_op/2, set_config/2,
-         get_state/1, get_leader/1, get_entry/2, get_last_entry/1]).
+         get_leader/1, get_entry/2, get_last_entry/1]).
 
 %% Test API
 -export([start_cluster/0, start_test_node/1]).
@@ -33,7 +33,7 @@ set_config(Peer, NewServers) ->
 
 -spec get_leader(peer()) -> peer() | undefined.
 get_leader(Peer) ->
-    rafter_consensus_fsm:leader(Peer).
+    rafter_consensus_fsm:get_leader(Peer).
 
 -spec get_entry(peer(), non_neg_integer()) -> term().
 get_entry(Peer, Index) ->
@@ -48,15 +48,15 @@ get_last_entry(Peer) ->
 %% =============================================
 
 start_cluster() ->
-    application:start(lager),
-    application:start(rafter),
+    ok = application:start(lager),
+    ok = application:start(rafter),
     Opts = #rafter_opts{state_machine=rafter_backend_echo, logdir="./log"},
     Peers = [peer1, peer2, peer3, peer4, peer5],
     [rafter_sup:start_peer(Me, Opts) || Me <- Peers].
 
 start_test_node(Name) ->
-    application:start(lager),
-    application:start(rafter),
+    ok = application:start(lager),
+    ok = application:start(rafter),
     Me = {Name, node()},
     Opts = #rafter_opts{state_machine=rafter_backend_ets, logdir="./data"},
     start_node(Me, Opts).

src/rafter_backend.erl

@@ -1,9 +1,7 @@
 -module(rafter_backend).
 
--export([behaviour_info/1]).
-
-behaviour_info(callbacks) ->
-    [{init, 1}, {read, 2}, {write, 2}];
-behaviour_info(_) ->
-    undefined.
-
+-callback init(Args :: term()) -> State :: term().
+-callback read(Operation :: term(), State :: term()) ->
+    {Value :: term(), State :: term()}.
+-callback write(Operation :: term(), State :: term()) ->
+    {Value :: term(), State :: term()}.

src/rafter_backend_ets.erl

@@ -10,8 +10,8 @@
 init(Peer) ->
     State = #state{peer=Peer},
     NewState = stop(State),
-    ets:new(rafter_backend_ets, [set, named_table, public]),
-    ets:new(rafter_backend_ets_tables, [set, named_table, public]),
+    _Tid = ets:new(rafter_backend_ets, [set, named_table, public]),
+    _Tid = ets:new(rafter_backend_ets_tables, [set, named_table, public]),
     NewState.
 
 stop(State) ->
@@ -46,7 +46,7 @@ read({list_keys, Table}, State) ->
 
 write({new, Name}, State) ->
     Val = try
-              ets:new((Name), [ordered_set, named_table, public]),
+              _Tid = ets:new((Name), [ordered_set, named_table, public]),
               ets:insert(rafter_backend_ets_tables, {Name}),
               {ok, Name}
           catch _:E ->

src/rafter_config.erl

@@ -10,15 +10,15 @@
 %% API
 %%====================================================================
 
--spec quorum_min(term(), #config{}, dict()) -> non_neg_integer().
+-spec quorum_min(term(), #config{} | [], dict()) -> non_neg_integer().
 quorum_min(_Me, #config{state=blank}, _) ->
     0;
 quorum_min(Me, #config{state=stable, oldservers=OldServers}, Responses) ->
     quorum_min(Me, OldServers, Responses);
 quorum_min(Me, #config{state=staging, oldservers=OldServers}, Responses) ->
     quorum_min(Me, OldServers, Responses);
-quorum_min(Me, #config{state=transitional, 
-                   oldservers=Old, 
+quorum_min(Me, #config{state=transitional,
+                   oldservers=Old,
                    newservers=New}, Responses) ->
     min(quorum_min(Me, Old, Responses), quorum_min(Me, New, Responses));
 
@@ -51,13 +51,13 @@ quorum(Me, #config{state=transitional, oldservers=Old, newservers=New}, Response
 %% this case.
 quorum(Me, Servers, Responses) ->
     TrueResponses = [R || {Peer, R} <- dict:to_list(Responses), R =:= true,
-                                        lists:member(Peer, Servers)], 
+                                        lists:member(Peer, Servers)],
     case lists:member(Me, Servers) of
         true ->
             length(TrueResponses) + 1 > length(Servers)/2;
         false ->
             %% We are about to commit a new configuration that doesn't contain
-            %% the local leader. We must therefore have responses from a 
+            %% the local leader. We must therefore have responses from a
             %% majority of the other servers to have a quorum.
             length(TrueResponses) > length(Servers)/2
     end.
@@ -101,17 +101,17 @@ reconfig(#config{state=stable}=Config, Servers) ->
 -spec allow_config(#config{}, list()) -> boolean().
 allow_config(#config{state=blank}, _NewServers) ->
     true;
-allow_config(#config{state=stable, oldservers=OldServers}, NewServers) 
+allow_config(#config{state=stable, oldservers=OldServers}, NewServers)
     when NewServers =/= OldServers ->
     true;
-allow_config(#config{oldservers=OldServers}, NewServers) 
-    when NewServers =:= OldServers -> 
+allow_config(#config{oldservers=OldServers}, NewServers)
+    when NewServers =:= OldServers ->
     {error, not_modified};
 allow_config(_Config, _NewServers) ->
     {error, config_in_progress}.
 
 %%====================================================================
-%% Internal Functions 
+%% Internal Functions
 %%====================================================================
 
 -spec index(atom() | {atom(), atom()}, dict()) -> non_neg_integer().

src/rafter_consensus_fsm.erl

@@ -12,7 +12,7 @@
 -define(HEARTBEAT_TIMEOUT, 25).
 
 %% API
--export([start/0, stop/1, start/1, start_link/3, read_op/2, op/2,
+-export([start_link/3, stop/1, get_leader/1, read_op/2, op/2,
          set_config/2, send/2, send_sync/2]).
 
 %% gen_fsm callbacks
@@ -25,19 +25,11 @@
 %% Testing outputs
 -export([set_term/2, candidate_log_up_to_date/4]).
 
-%% This function is simply for testing a single peer with erlang transport
-start() ->
-    start(peer1).
-
 stop(Pid) ->
     gen_fsm:send_all_state_event(Pid, stop).
 
-start(Me) ->
-    gen_fsm:start({local, Me}, ?MODULE, [Me], []).
-
 start_link(NameAtom, Me, Opts) ->
     gen_fsm:start_link({local, NameAtom}, ?MODULE, [Me, Opts], []).
-    %%gen_fsm:start_link({local, NameAtom}, ?MODULE, [Me, Opts], [{debug, [trace]}]).
 
 op(Peer, Command) ->
     gen_fsm:sync_send_event(Peer, {op, Command}).
@@ -48,13 +40,16 @@ read_op(Peer, Command) ->
 set_config(Peer, Config) ->
     gen_fsm:sync_send_event(Peer, {set_config, Config}).
 
+get_leader(Pid) ->
+    gen_fsm:sync_send_all_state_event(Pid, get_leader).
+
 -spec send(atom(), #vote{} | #append_entries_rpy{}) -> ok.
 send(To, Msg) ->
     %% Catch badarg error thrown if name is unregistered
     catch gen_fsm:send_event(To, Msg).
 
 -spec send_sync(atom(), #request_vote{} | #append_entries{}) ->
-    {ok, #vote{}} | {ok, #append_entries_rpy{}}.
+    #vote{} | #append_entries_rpy{} | timeout.
 send_sync(To, Msg) ->
     Timeout=100,
     gen_fsm:sync_send_event(To, Msg, Timeout).
@@ -64,7 +59,6 @@ send_sync(To, Msg) ->
 %%=============================================================================
 
 init([Me, #rafter_opts{state_machine=StateMachine}]) ->
-    random:seed(),
     Timer = gen_fsm:send_event_after(election_timeout(), timeout),
     #meta{voted_for=VotedFor, term=Term} = rafter_log:get_metadata(Me),
     BackendState = StateMachine:init(Me),
@@ -95,6 +89,8 @@ handle_event(stop, _, State) ->
 handle_event(_Event, _StateName, State) ->
     {stop, {error, badmsg}, State}.
 
+handle_sync_event(get_leader, _, StateName, State=#state{leader=Leader}) ->
+    {reply, Leader, StateName, State};
 handle_sync_event(_Event, _From, _StateName, State) ->
     {stop, badmsg, State}.
 
@@ -174,7 +170,7 @@ follower(#append_entries{term=Term, from=From, prev_log_index=PrevLogIndex,
     State3 = reset_timer(election_timeout(), State2),
     case consistency_check(AppendEntries, State3) of
         false ->
-            lager:info("~p ~p ~n", [AppendEntries, State3]),
+            ok = lager:info("~p ~p ~n", [AppendEntries, State3]),
             {reply, Rpy, follower, State3};
         true ->
             {ok, CurrentIndex} = rafter_log:check_and_append(Me,
@@ -811,7 +807,7 @@ send_entry(Peer, Index, #state{me=Me,
 
 send_append_entries(#state{followers=Followers, send_clock=SendClock}=State) ->
     NewState = State#state{send_clock=SendClock+1},
-    [send_entry(Peer, Index, NewState) ||
+    _ = [send_entry(Peer, Index, NewState) ||
         {Peer, Index} <- dict:to_list(Followers)],
     NewState.
 
@@ -843,7 +839,7 @@ become_candidate(#state{term=CurrentTerm, me=Me}=State0) ->
                          responses=dict:new(),
                          leader=undefined},
     State3 = set_metadata(Me, State2),
-    request_votes(State3),
+    _ = request_votes(State3),
     State3.
 
 become_leader(#state{me=Me, term=Term, init_config=InitConfig}=State) ->

src/rafter_consensus_sup.erl

@@ -8,7 +8,6 @@
 %% Supervisor callbacks
 -export([init/1]).
 
--spec start_link(atom() | {atom(), atom()}, atom()) -> ok.
 start_link(Me, Opts) when is_atom(Me) ->
     SupName = name(Me, "sup"),
     start_link(Me, SupName, Me, Opts);
@@ -29,7 +28,7 @@ init([NameAtom, Me, Opts]) ->
     {ok, {{one_for_all, 5, 10}, [LogServer, ConsensusFsm]}}.
 
 %% ===================================================================
-%% Private Functions 
+%% Private Functions
 %% ===================================================================
 name(Name, Extension) ->
     list_to_atom(atom_to_list(Name) ++ "_" ++ Extension).

src/rafter_log.erl

@@ -258,13 +258,13 @@ maybe_append(Index, Loc, [#rafter_entry{term=Term}=Entry | Entries],
                 #rafter_entry{index=Index, term=Term} ->
                     maybe_append(Index+1, NewLocation, Entries, State);
                 #rafter_entry{index=Index, term=_} ->
-                    truncate(File, Loc),
+                    ok = truncate(File, Loc),
                     State1 = State#state{write_location=Loc},
                     State2 = write_entry(Entry, State1),
                     maybe_append(Index + 1, eof, Entries, State2)
             end;
         eof ->
-            truncate(File, Loc),
+            ok = truncate(File, Loc),
             State1 = State#state{write_location=Loc},
             State2 = write_entry(Entry, State1),
             maybe_append(Index+1, eof, Entries, State2)
@@ -353,7 +353,7 @@ read_metadata(Filename, FileSize) ->
     end.
 
 truncate(File, Pos) ->
-    file:position(File, Pos),
+    {ok, _} = file:position(File, Pos),
     file:truncate(File).
 
 maybe_truncate(File, TruncateAt, FileSize) ->
@@ -373,7 +373,7 @@ repair_file(File, Size) ->
             {ok, ConfigStart, Term, Index, TruncateAt};
         not_found ->
             io:format("NOT FOUND: Size = ~p~n", [Size]),
-            truncate(File, 0),
+            ok = truncate(File, 0),
             empty_file
     end.
 

src/rafter_requester.erl

@@ -14,8 +14,8 @@ send(To, #append_entries{from=From}=Msg) ->
 send(To, From, Msg) ->
     spawn(fun() ->
               case rafter_consensus_fsm:send_sync(To, Msg) of
-                  Rpy when is_record(Rpy, vote) orelse 
-                           is_record(Rpy, append_entries_rpy) -> 
+                  Rpy when is_record(Rpy, vote) orelse
+                           is_record(Rpy, append_entries_rpy) ->
                       rafter_consensus_fsm:send(From, Rpy);
                   E ->
                       lager:error("Error sending ~p to To ~p: ~p", [Msg, To, E])

src/rafter_sup.erl

@@ -2,6 +2,8 @@
 
 -behaviour(supervisor).
 
+-include("rafter_opts.hrl").
+
 %% API
 -export([start_link/0, start_peer/2, stop_peer/1]).
 
@@ -15,11 +17,12 @@
 start_link() ->
     supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
-%% @doc Start an individual peer. In production, this will only be called once 
-%% on each machine with node/local name semantics. 
-%% For testing, this allows us to start 5 nodes in one erlang VM and 
+%% @doc Start an individual peer. In production, this will only be called once
+%% on each machine with node/local name semantics.
+%% For testing, this allows us to start 5 nodes in one erlang VM and
 %% communicate with local names.
--spec start_peer(atom() | {atom(), atom()}, atom()) -> ok.
+-spec start_peer(atom() | {atom(), atom()}, #rafter_opts{}) ->
+    {'error',_} | {'ok','undefined' | pid()} | {'ok','undefined' | pid(),_}.
 start_peer(Me, Opts) when is_atom(Me) ->
     SupName = consensus_sup(Me),
     start_child(SupName, Me, Opts);
@@ -45,7 +48,7 @@ init([]) ->
     {ok, { {one_for_one, 5, 10}, []} }.
 
 %% ===================================================================
-%% Private Functions 
+%% Private Functions
 %% ===================================================================
 consensus_sup(Peer) ->
     list_to_atom(atom_to_list(Peer) ++ "_consensus_sup").
@@ -57,7 +60,7 @@ start_child(SupName, Me, Opts) ->
     supervisor:start_child(?MODULE, ConsensusSup).
 
 stop_child(ChildId) ->
-    supervisor:terminate_child(?MODULE, ChildId),
+    ok = supervisor:terminate_child(?MODULE, ChildId),
     %% A big meh to OTP supervisor child handling. Just delete so start works
     %% again.
     supervisor:delete_child(?MODULE, ChildId).