put back get_terms/fold_terms

Reviewed By: hsun324

Differential Revision: D72419076

fbshipit-source-id: 7e84f4f8dcae6216808c94456b40cd13998aba6a

Author: Sarah Hassan

src/wa_raft_log.erl

@@ -30,11 +30,13 @@
     fold/5,
     fold/6,
 
+    fold_terms/5,
 
     term/2,
     get/2,
     get/3,
     get/4,
+    get_terms/3,
 
     config/1
 ]).
@@ -434,6 +436,45 @@ fold_impl(Log, First, Last, SizeLimit, Func, AccIn) ->
             {error, Reason}
     end.
 
+%% Folds over the terms in the log view of raw entries from the log provider
+%% between the provided first and last log indices (inclusive).
+%% If there exists a log term between the provided first and last indices then
+%% the accumulator function will be called on at least that term.
+%% This API provides no validation of the log indices and term passed by the
+%% provider to the callback function.
+-spec fold_terms(LogOrView :: log() | view(),
+                 First :: log_index(),
+                 Last :: log_index(),
+                 Func :: fun((Index :: log_index(), Term :: log_term(), Acc) -> Acc),
+                 Acc) ->
+    {ok, Acc} | wa_raft:error().
+fold_terms(#log_view{log = Log, first = LogFirst, last = LogLast}, First, Last, Func, Acc) ->
+    fold_terms_impl(Log, max(First, LogFirst), min(Last, LogLast), Func, Acc);
+fold_terms(Log, First, Last, Func, Acc) ->
+    Provider = provider(Log),
+    LogFirst = Provider:first_index(Log),
+    LogLast = Provider:last_index(Log),
+    fold_terms_impl(Log, max(First, LogFirst), min(Last, LogLast), Func, Acc).
+
+-spec fold_terms_impl(
+    Log :: log(),
+    First :: log_index(),
+    Last :: log_index(),
+    Func :: fun((Index :: log_index(), Term :: log_term(), Acc) -> Acc),
+    Acc :: term()
+) -> {ok, Acc} | wa_raft:error().
+fold_terms_impl(Log, First, Last, Func, AccIn) ->
+    ?RAFT_COUNT('raft.log.fold_terms'),
+    ?RAFT_COUNTV('raft.log.fold_terms.total', Last - First + 1),
+    Provider = provider(Log),
+    case Provider:fold_terms(Log, First, Last, Func, AccIn) of
+        {ok, AccOut} ->
+            {ok, AccOut};
+        {error, Reason} ->
+            ?RAFT_COUNT('raft.log.fold_terms.error'),
+            {error, Reason}
+    end.
+
 %% Gets the term of entry at the provided log index. When using a log view
 %% this function may return 'not_found' even if the underlying log entry still
 %% exists if the entry is outside of the log view.
@@ -492,6 +533,25 @@ get(LogOrView, First, CountLimit, SizeLimit) ->
             {error, corruption}
     end.
 
+-spec get_terms(LogOrView :: log() | view(), First :: log_index(), Limit :: non_neg_integer()) ->
+    {ok, Terms :: [wa_raft_log:log_term()]} | wa_raft:error().
+get_terms(LogOrView, First, Limit) ->
+    try
+        fold_terms(LogOrView, First, First + Limit - 1,
+            fun
+                (Index, Term, {Index, Acc})            -> {Index + 1, [Term | Acc]};
+                (_Index, _Term, {ExpectedIndex, _Acc}) -> throw({missing, ExpectedIndex})
+            end, {First, []})
+    of
+        {ok, {_, TermsRev}} -> {ok, lists:reverse(TermsRev)};
+        {error, Reason} -> {error, Reason}
+    catch
+        throw:{missing, Index} ->
+            ?LOG_WARNING("[~p] detected log is missing index ~p during get of ~p ~~ ~p",
+                [log_name(LogOrView), Index, First, First + Limit - 1], #{domain => [whatsapp, wa_raft]}),
+            {error, corruption}
+    end.
+
 -spec config(LogOrView :: log() | view()) -> {ok, Index :: log_index(), Config :: wa_raft_server:config() | undefined} | not_found.
 config(#log_view{config_index = undefined}) ->
     not_found;