Add API to force a server into single member mode

src/ra_log_meta.erl

@@ -94,8 +94,8 @@ handle_batch(Commands, #?MODULE{ref = Ref,
                    [{reply, From, ok} | Replies], true}
           end, {#{}, [], false}, Commands),
     Objects = maps:values(Inserts),
-    ok = dets:insert(TblName, Objects),
     true = ets:insert(TblName, Objects),
+    ok = dets:insert(TblName, Objects),
     case ShouldSync of
         true ->
             ok = dets:sync(TblName);

src/ra_server.erl

@@ -744,6 +744,8 @@ handle_leader({transfer_leadership, ServerId},
 handle_leader({register_external_log_reader, Pid}, #{log := Log0} = State) ->
     {Log, Effs} = ra_log:register_reader(Pid, Log0),
     {leader, State#{log => Log}, Effs};
+handle_leader(force_member_change, State0) ->
+    {follower, State0#{votes => 0}, [{next_event, force_member_change}]};
 handle_leader(Msg, State) ->
     log_unhandled_msg(leader, Msg, State),
     {leader, State, []}.
@@ -853,6 +855,8 @@ handle_candidate(election_timeout, State) ->
 handle_candidate({register_external_log_reader, Pid}, #{log := Log0} = State) ->
     {Log, Effs} = ra_log:register_reader(Pid, Log0),
     {candidate, State#{log => Log}, Effs};
+handle_candidate(force_member_change, State0) ->
+    {follower, State0#{votes => 0}, [{next_event, force_member_change}]};
 handle_candidate(Msg, State) ->
     log_unhandled_msg(candidate, Msg, State),
     {candidate, State, []}.
@@ -931,6 +935,8 @@ handle_pre_vote({ra_log_event, Evt}, State = #{log := Log0}) ->
 handle_pre_vote({register_external_log_reader, Pid}, #{log := Log0} = State) ->
     {Log, Effs} = ra_log:register_reader(Pid, Log0),
     {pre_vote, State#{log => Log}, Effs};
+handle_pre_vote(force_member_change, State0) ->
+    {follower, State0#{votes => 0}, [{next_event, force_member_change}]};
 handle_pre_vote(Msg, State) ->
     log_unhandled_msg(pre_vote, Msg, State),
     {pre_vote, State, []}.
@@ -1064,7 +1070,7 @@ handle_follower(#heartbeat_rpc{query_index = RpcQueryIndex, term = Term,
     Reply = heartbeat_reply(State2),
     {follower, State2, [cast_reply(Id, LeaderId, Reply)]};
 handle_follower(#heartbeat_rpc{leader_id = LeaderId},
-                #{cfg := #cfg{id = Id}} = State)->
+                #{cfg := #cfg{id = Id}} = State) ->
     Reply = heartbeat_reply(State),
     {follower, State, [cast_reply(Id, LeaderId, Reply)]};
 handle_follower({ra_log_event, {written, _} = Evt},
@@ -1187,6 +1193,14 @@ handle_follower(try_become_leader, State) ->
 handle_follower({register_external_log_reader, Pid}, #{log := Log0} = State) ->
     {Log, Effs} = ra_log:register_reader(Pid, Log0),
     {follower, State#{log => Log}, Effs};
+handle_follower(force_member_change,
+                #{cfg := #cfg{id = Id,
+                              log_id = LogId}} = State0) ->
+    Cluster = #{Id => new_peer()},
+    ?WARN("~s: Forcing cluster change. New cluster ~w",
+          [LogId, Cluster]),
+    {ok, _, _, State} = append_cluster_change(Cluster, undefined, no_reply, State0),
+    call_for_election(pre_vote, State, [{reply, ok}]);
 handle_follower(Msg, State) ->
     log_unhandled_msg(follower, Msg, State),
     {follower, State, []}.
@@ -1871,8 +1885,12 @@ log_read(Indexes, #{log := Log0} = State) ->
 %%% Internal functions
 %%%===================================================================
 
+call_for_election(TargetState, State) ->
+    call_for_election(TargetState, State, []).
+
 call_for_election(candidate, #{cfg := #cfg{id = Id, log_id = LogId} = Cfg,
-                               current_term := CurrentTerm} = State0) ->
+                               current_term := CurrentTerm} = State0,
+                 Effects) ->
     ok = incr_counter(Cfg, ?C_RA_SRV_ELECTIONS, 1),
     NewTerm = CurrentTerm + 1,
     ?DEBUG("~s: election called for in term ~b", [LogId, NewTerm]),
@@ -1888,11 +1906,13 @@ call_for_election(candidate, #{cfg := #cfg{id = Id, log_id = LogId} = Cfg,
     VoteForSelf = #request_vote_result{term = NewTerm, vote_granted = true},
     State = update_term_and_voted_for(NewTerm, Id, State0),
     {candidate, State#{leader_id => undefined, votes => 0},
-     [{next_event, cast, VoteForSelf}, {send_vote_requests, Reqs}]};
+     [{next_event, cast, VoteForSelf},
+      {send_vote_requests, Reqs} | Effects]};
 call_for_election(pre_vote, #{cfg := #cfg{id = Id,
                                           log_id = LogId,
                                           machine_version = MacVer} = Cfg,
-                              current_term := Term} = State0) ->
+                              current_term := Term} = State0,
+                 Effects) ->
     ok = incr_counter(Cfg, ?C_RA_SRV_PRE_VOTE_ELECTIONS, 1),
     ?DEBUG("~s: pre_vote election called for in term ~b", [LogId, Term]),
     Token = make_ref(),
@@ -1911,7 +1931,8 @@ call_for_election(pre_vote, #{cfg := #cfg{id = Id,
     State = update_term_and_voted_for(Term, Id, State0),
     {pre_vote, State#{leader_id => undefined, votes => 0,
                       pre_vote_token => Token},
-     [{next_event, cast, VoteForSelf}, {send_vote_requests, Reqs}]}.
+     [{next_event, cast, VoteForSelf},
+      {send_vote_requests, Reqs} | Effects]}.
 
 process_pre_vote(FsmState, #pre_vote_rpc{term = Term, candidate_id = Cand,
                                          version = Version,

src/ra_server_proc.erl

@@ -50,7 +50,8 @@
          trigger_election/2,
          ping/2,
          log_fold/4,
-         transfer_leadership/3
+         transfer_leadership/3,
+         force_shrink_members_to_current_member/1
         ]).
 
 -export([send_rpc/3]).
@@ -210,6 +211,10 @@ trigger_election(ServerId, Timeout) ->
 transfer_leadership(ServerId, TargetServerId, Timeout) ->
     leader_call(ServerId, {transfer_leadership, TargetServerId}, Timeout).
 
+-spec force_shrink_members_to_current_member(ra_server_id()) -> ok.
+force_shrink_members_to_current_member(ServerId) ->
+    gen_statem_safe_call(ServerId, force_member_change, 5000).
+
 -spec ping(ra_server_id(), timeout()) -> safe_call_ret({pong, states()}).
 ping(ServerId, Timeout) ->
     gen_statem_safe_call(ServerId, ping, Timeout).
@@ -434,7 +439,8 @@ leader(info, {node_event, _Node, _Evt}, State) ->
 leader(info, {'DOWN', _MRef, process, Pid, Info}, State0) ->
     handle_process_down(Pid, Info, ?FUNCTION_NAME, State0);
 leader(info, {Status, Node, InfoList}, State0)
-  when Status =:= nodedown orelse Status =:= nodeup ->
+  when Status =:= nodedown orelse
+       Status =:= nodeup ->
     handle_node_status_change(Node, Status, InfoList, ?FUNCTION_NAME, State0);
 leader(info, {update_peer, PeerId, Update}, State0) ->
     State = update_peer(PeerId, Update, State0),

test/ra_2_SUITE.erl

@@ -42,6 +42,7 @@ all_tests() ->
      segment_writer_handles_server_deletion,
      external_reader,
      add_member_without_quorum,
+     force_start_follower_as_single_member,
      initial_members_query
     ].
 
@@ -618,6 +619,47 @@ add_member_without_quorum(Config) ->
     % timer:sleep(5000),
     ok.
 
+force_start_follower_as_single_member(Config) ->
+    ok = logger:set_primary_config(level, all),
+    %% ra:start_server should fail if the node already exists
+    ClusterName = ?config(cluster_name, Config),
+    PrivDir = ?config(priv_dir, Config),
+    ServerId1 = ?config(server_id, Config),
+    ServerId2 = ?config(server_id2, Config),
+    ServerId3 = ?config(server_id3, Config),
+    InitialCluster = [ServerId1, ServerId2, ServerId3],
+    ok = start_cluster(ClusterName, InitialCluster),
+    timer:sleep(100),
+    %% stop majority to simulate permanent outage
+    ok = ra:stop_server(?SYS, ServerId1),
+    ok = ra:stop_server(?SYS, ServerId2),
+
+    timer:sleep(100),
+    %% force the remaining node to change it's membership
+    ok = ra_server_proc:force_shrink_members_to_current_member(ServerId3),
+    {ok, [_], ServerId3} = ra:members(ServerId3),
+    ok = enqueue(ServerId3, msg1),
+
+    %% test that it works after restart
+    ok = ra:stop_server(?SYS, ServerId3),
+    ok = ra:restart_server(?SYS, ServerId3),
+    {ok, [_], ServerId3} = ra:members(ServerId3),
+    ok = enqueue(ServerId3, msg2),
+
+    %% add a member
+    ServerId4 = ?config(server_id4, Config),
+    UId4 = ?config(uid4, Config),
+    Conf4 = conf(ClusterName, UId4, ServerId4, PrivDir, [ServerId3]),
+    {ok, _, _} = ra:add_member(ServerId3, ServerId4),
+    %% the membership has changed but member not running yet
+    {timeout,_} = ra:process_command(ServerId3, {enq, banana}),
+    %% start new member
+    ok = ra:start_server(?SYS, Conf4),
+    {ok, _, ServerId3} = ra:members(ServerId4),
+    ok = enqueue(ServerId3, msg3),
+
+    ok.
+
 initial_members_query(Config) ->
     ok = logger:set_primary_config(level, all),
     %% ra:start_server should fail if the node already exists

test/ra_server_SUITE.erl

@@ -2147,9 +2147,9 @@ leader_heartbeat_reply_node_size_5(_Config) ->
     HeartbeatReply = #heartbeat_reply{term = Term,
                                       query_index = QueryIndex},
     WaitingQuery = queue:in({QueryIndex, QueryRef1}, queue:new()),
-    State0 = set_peer_query_index(BaseState#{query_index => QueryIndex + 1,
+    State0 = set_peer_query_index(BaseState#{query_index => QueryIndex,
                                              queries_waiting_heartbeats => WaitingQuery},
-                                  Id, QueryIndex + 1),
+                                  Id, QueryIndex),
 
     {leader, State, []}
         = ra_server:handle_leader({ReplyingPeerId, HeartbeatReply}, State0),