Track commit request lifetime across terms

Summary:
Track ongoing commit requests within the RAFT server state itself
using the log index of the log entry assigned to the commit request.
This fixes an issue where replicas that lost leadership would cancel
ongoing commit requests with `{error, not_leader}` immediately
even if the commit could still be processed because the log entry
was not subsequently changed.

Reviewed By: jaher

Differential Revision: D75901229

fbshipit-source-id: ce89d3e27bf90a8c588bf72204ea608a74f128c7

Author: hsun324

include/wa_raft.hrl

@@ -373,7 +373,6 @@
     % Queue options
     queue_name :: atom(),
     queue_counters :: atomics:atomics_ref(),
-    queue_commits :: atom(),
     queue_reads :: atom(),
 
     % Server options
@@ -463,10 +462,13 @@
     %%          that are in queue to be appended and replicated after a short
     %%          wait to see if multiple commits can be handled at once to
     %%          reduce overhead
-    pending = [] :: [wa_raft_acceptor:op()],
+    pending = [] :: [{gen_server:from(), wa_raft_acceptor:op()}],
     %% [Leader] Whether or not a read has been accepted and is waiting for the
     %%          leader to establish a new quorum to be handled.
     pending_read = false :: boolean(),
+    %% [Leader] The queue of accepted commit requests that are waiting to be
+    %%          committed and applied for response to the client.
+    queued = #{} :: #{wa_raft_log:log_index() => gen_server:from()},
     %% [Leader] The index of the next log entry to send in the next heartbeat
     %%          to each peer
     next_indices = #{} :: #{node() => wa_raft_log:log_index()},

include/wa_raft_rpc.hrl

@@ -74,7 +74,7 @@
 
 -define(RAFT_COMMAND(Type, Payload), {command, Type, Payload}).
 
--define(COMMIT_COMMAND(Op),                                     ?RAFT_COMMAND(commit, Op)).
+-define(COMMIT_COMMAND(From, Op),                               ?RAFT_COMMAND(commit, {From, Op})).
 -define(READ_COMMAND(Op),                                       ?RAFT_COMMAND(read, Op)).
 
 -define(STATUS_COMMAND,                                         ?RAFT_COMMAND(status, undefined)).

src/wa_raft_acceptor.erl

@@ -82,11 +82,11 @@
 -type commit_async_request() :: {commit, From :: gen_server:from(), Op :: op()}.
 -type commit_error_type() ::
     not_leader |
-    {duplicate_request, Key :: key()} |
     {commit_queue_full, Key :: key()} |
     {apply_queue_full, Key :: key()} |
     {notify_redirect, Peer :: node()} |
-    commit_stalled.
+    commit_stalled |
+    cancelled.
 -type commit_error() :: {error, commit_error_type()}.
 -type commit_result() :: Result :: dynamic() | Error :: commit_error() | call_error().
 
@@ -244,24 +244,19 @@ commit_impl(From, {Key, _} = Op, #state{name = Name, server = Server, queues = Q
     try
         ?LOG_DEBUG("Acceptor[~0p] starts to handle commit of ~0P from ~0p.",
             [Name, Op, 30, From], #{domain => [whatsapp, wa_raft]}),
-        case wa_raft_queue:commit(Queues, Key, From) of
-            duplicate ->
-                ?LOG_WARNING("Acceptor[~0p] is rejecting commit request from ~0p because it has duplicate key ~0p.",
-                    [Name, From, Key], #{domain => [whatsapp, wa_raft]}),
-                ?RAFT_COUNT('raft.acceptor.error.duplicate_commit'),
-                {error, {duplicate_request, Key}};
+        case wa_raft_queue:commit_started(Queues) of
             commit_queue_full ->
-                ?LOG_WARNING("Acceptor[~0p] is rejecting commit request from ~0p with key ~0p because the commit queue is full.",
-                    [Name, From, Key], #{domain => [whatsapp, wa_raft]}),
+                ?LOG_WARNING("Acceptor[~0p] is rejecting commit request from ~0p because the commit queue is full.",
+                    [Name, From], #{domain => [whatsapp, wa_raft]}),
                 ?RAFT_COUNT('raft.acceptor.error.commit_queue_full'),
                 {error, {commit_queue_full, Key}};
             apply_queue_full ->
-                ?LOG_WARNING("Acceptor[~0p] is rejecting commit request from ~0p with key ~0p because the apply queue is full.",
-                    [Name, From, Key], #{domain => [whatsapp, wa_raft]}),
+                ?LOG_WARNING("Acceptor[~0p] is rejecting commit request from ~0p because the apply queue is full.",
+                    [Name, From], #{domain => [whatsapp, wa_raft]}),
                 ?RAFT_COUNT('raft.acceptor.error.apply_queue_full'),
                 {error, {apply_queue_full, Key}};
             ok ->
-                wa_raft_server:commit(Server, Op),
+                wa_raft_server:commit(Server, From, Op),
                 continue
         end
     after

src/wa_raft_part_sup.erl

@@ -155,7 +155,6 @@ normalize_spec(Application, #{table := Table, partition := Partition} = Spec) ->
         log_catchup_name = wa_raft_log_catchup:default_name(Table, Partition),
         queue_name = wa_raft_queue:default_name(Table, Partition),
         queue_counters = wa_raft_queue:default_counters(),
-        queue_commits = wa_raft_queue:default_commit_queue_name(Table, Partition),
         queue_reads = wa_raft_queue:default_read_queue_name(Table, Partition),
         server_name = ServerName,
         storage_name = StorageName,

src/wa_raft_queue.erl

@@ -29,17 +29,15 @@
 -export([
     default_name/2,
     default_counters/0,
-    default_commit_queue_name/2,
     default_read_queue_name/2,
     registered_name/2
 ]).
 
 %% PENDING COMMIT QUEUE API
 -export([
-    commit/3,
-    fulfill_commit/3,
-    fulfill_incomplete_commit/3,
-    fulfill_all_commits/2
+    commit_started/1,
+    commit_cancelled/3,
+    commit_completed/3
 ]).
 
 %% PENDING READ API
@@ -108,7 +106,6 @@
 -record(queues, {
     application :: atom(),
     counters :: atomics:atomics_ref(),
-    commits :: atom(),
     reads :: atom()
 }).
 -opaque queues() :: #queues{}.
@@ -122,7 +119,6 @@ queues(Options) ->
     #queues{
         application = Options#raft_options.application,
         counters = Options#raft_options.queue_counters,
-        commits = Options#raft_options.queue_commits,
         reads = Options#raft_options.queue_reads
     }.
 
@@ -204,12 +200,6 @@ default_name(Table, Partition) ->
 default_counters() ->
     atomics:new(?RAFT_NUMBER_OF_QUEUE_SIZE_COUNTERS, []).
 
-%% Get the default name for the RAFT commit queue ETS table associated with the
-%% provided RAFT partition.
--spec default_commit_queue_name(Table :: wa_raft:table(), Partition :: wa_raft:partition()) -> Name :: atom().
-default_commit_queue_name(Table, Partition) ->
-    binary_to_atom(<<"raft_commit_queue_", (atom_to_binary(Table))/bytes, "_", (integer_to_binary(Partition))/bytes>>).
-
 %% Get the default name for the RAFT read queue ETS table associated with the
 %% provided RAFT partition.
 -spec default_read_queue_name(Table :: wa_raft:table(), Partition :: wa_raft:partition()) -> Name :: atom().
@@ -229,9 +219,8 @@ registered_name(Table, Partition) ->
 %% PENDING COMMIT QUEUE API
 %%-------------------------------------------------------------------
 
--spec commit(Queues :: queues(), Key :: wa_raft_acceptor:key(), From :: gen_server:from()) ->
-    ok | apply_queue_full | commit_queue_full | duplicate.
-commit(#queues{counters = Counters, commits = Commits} = Queues, Reference, From) ->
+-spec commit_started(Queues :: queues()) -> ok | apply_queue_full | commit_queue_full.
+commit_started(#queues{counters = Counters} = Queues) ->
     case commit_queue_full(Queues) of
         true ->
             commit_queue_full;
@@ -240,47 +229,24 @@ commit(#queues{counters = Counters, commits = Commits} = Queues, Reference, From
                 true ->
                     apply_queue_full;
                 false ->
-                    case ets:insert_new(Commits, {Reference, From}) of
-                        true ->
-                            PendingCommits = atomics:add_get(Counters, ?RAFT_COMMIT_QUEUE_SIZE_COUNTER, 1),
-                            ?RAFT_GATHER('raft.acceptor.commit.request.pending', PendingCommits),
-                            ok;
-                        false ->
-                            duplicate
-                    end
+                    PendingCommits = atomics:add_get(Counters, ?RAFT_COMMIT_QUEUE_SIZE_COUNTER, 1),
+                    ?RAFT_GATHER('raft.acceptor.commit.request.pending', PendingCommits),
+                    ok
             end
     end.
 
-% Fulfill a pending commit with the result of the application of the command contained
-% within the commit.
--spec fulfill_commit(Queues :: queues(), term(), dynamic()) -> ok | not_found.
-fulfill_commit(#queues{counters = Counters, commits = Commits}, Reference, Reply) ->
-    case ets:take(Commits, Reference) of
-        [{Reference, From}] ->
-            atomics:sub(Counters, ?RAFT_COMMIT_QUEUE_SIZE_COUNTER, 1),
-            gen_server:reply(From, Reply);
-        [] ->
-            not_found
-    end.
 
-% Fulfill a pending commit with an error that indicates that the commit was not completed.
--spec fulfill_incomplete_commit(Queues :: queues(), term(), wa_raft_acceptor:commit_error()) -> ok | not_found.
-fulfill_incomplete_commit(Queues, Reference, Error) ->
-    fulfill_commit(Queues, Reference, Error).
+-spec commit_cancelled(Queues :: queues(), From :: gen_server:from(), Reason :: wa_raft_acceptor:commit_error() | undefined) -> ok.
+commit_cancelled(#queues{counters = Counters}, From, Reason) ->
+    atomics:sub(Counters, ?RAFT_COMMIT_QUEUE_SIZE_COUNTER, 1),
+    Reason =/= undefined andalso gen_server:reply(From, Reason),
+    ok.
 
-% Fulfill a pending commit with an error that indicates that the commit was not completed.
--spec fulfill_all_commits(Queues :: queues(), wa_raft_acceptor:commit_error()) -> ok.
-fulfill_all_commits(#queues{counters = Counters, commits = Commits}, Reply) ->
-    lists:foreach(
-        fun ({Reference, _}) ->
-            case ets:take(Commits, Reference) of
-                [{Reference, From}] ->
-                    atomics:sub(Counters, ?RAFT_COMMIT_QUEUE_SIZE_COUNTER, 1),
-                    gen_server:reply(From, Reply);
-                [] ->
-                    ok
-            end
-        end, ets:tab2list(Commits)).
+-spec commit_completed(Queues :: queues(), From :: gen_server:from(), Reply :: term()) -> ok.
+commit_completed(#queues{counters = Counters}, From, Reply) ->
+    atomics:sub(Counters, ?RAFT_COMMIT_QUEUE_SIZE_COUNTER, 1),
+    gen_server:reply(From, Reply),
+    ok.
 
 %%-------------------------------------------------------------------
 %% PENDING READ QUEUE API
@@ -398,23 +364,21 @@ init(
         partition = Partition,
         queue_name = Name,
         queue_counters = Counters,
-        queue_commits = CommitsName,
         queue_reads = ReadsName
     }
 ) ->
     process_flag(trap_exit, true),
 
-    ?LOG_NOTICE("Queue[~p] starting for partition ~0p/~0p with read queue ~0p and commit queue ~0p",
-        [Name, Table, Partition, ReadsName, CommitsName], #{domain => [whatsapp, wa_raft]}),
+    ?LOG_NOTICE("Queue[~p] starting for partition ~0p/~0p with read queue ~0p",
+        [Name, Table, Partition, ReadsName], #{domain => [whatsapp, wa_raft]}),
 
     % The queue process is the first process in the supervision for a single
     % RAFT partition. The supervisor is configured to restart all processes if
     % even a single process fails. Since the queue process is starting up, all
     % queues tracked should be empty so reset all counters.
     [atomics:put(Counters, Index, 0) || Index <- lists:seq(1, ?RAFT_NUMBER_OF_QUEUE_SIZE_COUNTERS)],
 
-    % Create ETS tables for pending commits and reads.
-    CommitsName = ets:new(CommitsName, [set | ?RAFT_QUEUE_TABLE_OPTIONS]),
+    % Create ETS table for pending reads.
     ReadsName = ets:new(ReadsName, [ordered_set | ?RAFT_QUEUE_TABLE_OPTIONS]),
 
     {ok, #state{name = Name}}.