Fix repair scenarios where no record on term server

Author: Peter Tihanyi

src/pes_registrar.erl

@@ -200,6 +200,14 @@ handle_event(state_timeout, heartbeat, monitoring, #state{replies = Replies,
       %?TRACE("cannot_renew timeout ~p", [Answer], State#state.id),
       {stop, {cannot_renew_registration, {timeout, Answer}}, State}
   end;
+handle_event(info, #promise_reply{result = {nack, {Server, not_found}}} = Reply, monitoring,
+             #state{id = Id, term = Term, pid = Pid, last_timestamp = Now} = State) ->
+  % we are in monitoring phase so we can do repair when term not found on server
+  NewResult = case pes_promise:await(repair(Server, Id, not_found, Term, {Pid, self(), Now})) of
+                ack -> ack;
+                _ -> nack
+              end,
+  handle_event(info, Reply#promise_reply{result = NewResult}, monitoring, State);
 handle_event(info, #promise_reply{result = {nack, {Server, OldTerm}}} = Reply, monitoring,
              #state{id = Id, term = Term, pid = Pid, last_timestamp = Now} = State) ->
   % we are in monitoring phase so we can do repair because we surely have the majority,
@@ -244,6 +252,10 @@ handle_event(_EventType, _EventContext, handoff, _State) ->
   % queue all the stuff until handoff is not ready
   {keep_state_and_data, [postpone]};
 
+% transfer is only allowed in monitoring or registered state
+handle_event({call, From}, {update, _}, StateName, _State)
+  when StateName =/= monitoring andalso StateName =/= registered ->
+  {keep_state_and_data, [{reply, From, {error, not_in_proper_state}}]};
 % we need to update the guarded pid
 % @TODO unfortunately if the registration not succeed that the old reg could not be restored
 handle_event({call, From}, {update, NewPid}, _StateName, State) when node(NewPid) =:= node() ->
@@ -252,8 +264,7 @@ handle_event({call, From}, {update, NewPid}, _StateName, State) when node(NewPid
   % If it goes down and the pid is not matched in the state basically we just ignores it.
   erlang:monitor(process, NewPid),
   {next_state, commit, State#state{pid = NewPid, caller = From}};
-handle_event({call, From}, {update, NewPid}, StateName,
-              #state{id = Id, term = Term} = State) ->
+handle_event({call, From}, {update, NewPid}, StateName, #state{id = Id} = State) ->
   % things gets complicated we need too transfer the guard process to the target node
   Now = pes_time:now(),
   Nodes = pes_cluster:nodes(),
@@ -263,9 +274,8 @@ handle_event({call, From}, {update, NewPid}, StateName,
   ),
   TargetNode = node(NewPid),
   {ok, NewGuard} = rpc:call(TargetNode, gen_statem, start, [?MODULE, {handoff, NewState}, []]),
-  CurrentTerm = encapsulate_term(Term),
   NewValue = {NewPid, NewGuard, Now},
-  Promises = [repair(Server, Id, CurrentTerm, NewState#state.term, NewValue) || Server <- Nodes],
+  Promises = [force_repair(Server, Id, NewState#state.term, NewValue) || Server <- Nodes],
   lists:foreach(fun(Promise) -> pes_promise:await(Promise, ?DEFAULT_TIMEOUT) end, Promises),
   ok = gen_statem:call(NewGuard, {handoff_ready, StateName}),
   gen_statem:reply(From, registered),
@@ -437,6 +447,9 @@ commit(Node, Id, Term, Value) ->
 repair(Node, Id, OldTerm, NewTerm, Value) ->
   pes_server_sup:repair(Node, Id, OldTerm, encapsulate_term(NewTerm), Value).
 
+force_repair(Node, Id, NewTerm, Value) ->
+  pes_server_sup:force_repair(Node, Id, encapsulate_term(NewTerm), Value).
+
 encapsulate_term(Term) ->
   {Term, self()}.
 

src/pes_server.erl

@@ -21,7 +21,7 @@
 -compile({no_auto_import, [register/2, is_process_alive/1, send/2]}).
 
 %% API
--export([prepare/3, commit/4, read/2, repair/5]).
+-export([prepare/3, commit/4, read/2, repair/5, force_repair/4]).
 
 -export([start_link/1, init/1, loop/1]).
 -export([system_continue/3, system_terminate/4, system_get_state/1, system_code_change/4]).
@@ -64,11 +64,16 @@ read({Server, Node}, Id) when Node =:= node() ->
 read(Node, Id) ->
   async(Node, {read, Id}).
 
--spec repair(target(), id(), consensus_term_proposal(), consensus_term_proposal(), value()) ->
+-spec repair(target(), id(), consensus_term_proposal() | not_found, consensus_term_proposal(), value()) ->
   pes_promise:promise().
 repair(Node, Id, CurrentTerm, NewTerm, Value) ->
   async(Node, {repair, Id, CurrentTerm, NewTerm, Value}).
 
+-spec force_repair(target(), id(), consensus_term_proposal(), value()) ->
+  pes_promise:promise().
+force_repair(Node, Id, NewTerm, Value) ->
+  async(Node, {repair, Id, NewTerm, Value}).
+
 -spec async(target(), term()) -> pes_promise:promise().
 async({Server, Node}, Command) ->
   case pes_cluster:is_node_alive(Node) of
@@ -143,9 +148,17 @@ handle_command({commit, Id, {Term, Server}, Value}, #state{data_storage_ref = DS
 % Commit can reply with nack and the actual term, and server data.
 % To ensure in the mean time no other registration attempt were made,
 % we need to send back those values.
-handle_command({repair, Id, _Term, {NewTermId, _} = NewTerm, Value},
-                #state{data_storage_ref = DSR,
-                       term_storage_ref = TSR}) ->
+handle_command({repair, Id, Term, NewTerm, Value}, #state{term_storage_ref = TSR} = State) ->
+  case ets:lookup(TSR, Id) of
+    [] -> % if the db is empty than we are good to rewrite stuff
+      handle_command({repair, Id, NewTerm, Value}, State);
+    [{_Id, StoredTerm}] when StoredTerm =:= Term ->
+      handle_command({repair, Id, NewTerm, Value}, State);
+    _ ->
+      nack
+  end;
+handle_command({repair, Id, {NewTermId, _} = NewTerm, Value},
+               #state{data_storage_ref = DSR, term_storage_ref = TSR}) ->
   pes_stat:count([server, repair]),
   true = ets:insert(TSR, {Id, NewTerm}),
   true = ets:insert(DSR, {Id, {NewTermId, Value}, pes_time:now()}),

src/pes_server_sup.erl

@@ -10,7 +10,7 @@
 
 % @TODO different nodes can have different pes_server shard count, and it does not work!!! :/
 
--export([start_link/1, prepare/3, commit/4, read/2, repair/5]).
+-export([start_link/1, prepare/3, commit/4, read/2, repair/5, force_repair/4]).
 
 -export([servers/0]).
 
@@ -35,11 +35,18 @@ read(Node, Id) ->
 -spec repair(node(), Id, Term, NewTerm, term()) ->
   pes_promise:promise() when
   Id :: pes_server:id(),
-  Term :: pes_server:consensus_term_proposal(),
+  Term :: pes_server:consensus_term_proposal() | not_found,
   NewTerm :: {pes_server:consensus_term(), pid()}.
 repair(Node, Id, OldTerm, NewTerm, Value) ->
   pes_server:repair({hash(Id), Node}, Id, OldTerm, NewTerm, Value).
 
+-spec force_repair(node(), Id, NewTerm, term()) ->
+  pes_promise:promise() when
+  Id :: pes_server:id(),
+  NewTerm :: {pes_server:consensus_term(), pid()}.
+force_repair(Node, Id, NewTerm, Value) ->
+  pes_server:force_repair({hash(Id), Node}, Id, NewTerm, Value).
+
 -spec start_link(pos_integer()) -> {ok, pid()}.
 start_link(ServerCount) ->
   supervisor:start_link({local, ?SERVER}, ?MODULE, ServerCount).