wip

kjnilsson · kjnilsson · commit a905053c9865 · 2025-03-17T18:09:42.000Z
diff --git a/src/ra_log_wal.erl b/src/ra_log_wal.erl
@@ -58,7 +58,7 @@
 -record(batch_writer, {smallest_live_idx :: ra_index(),
                        tid :: ets:tid(),
                        uid :: term(),
-                       range :: ra:range(),
+                       seq :: ra_seq:state(),
                        term :: ra_term(),
                        old :: undefined | #batch_writer{}
                       }).
@@ -491,39 +491,39 @@ write_data({UId, Pid} = Id, MtTid, Idx, Term, Data0, Trunc, SmallestIndex,
     end.
 
 
-handle_msg({append, {UId, Pid} = Id, MtTid, PrevIdx0, Idx, Term, Entry},
+handle_msg({append, {UId, Pid} = Id, MtTid, ExpectedPrevIdx, Idx, Term, Entry},
            #state{conf = Conf,
                   writers = Writers} = State0) ->
     SmallestIdx = smallest_live_index(Conf, UId),
     %% detect if truncating flag should be set
     Trunc = Idx == SmallestIdx,
 
-    case maps:find(UId, Writers) of
+    case maps:get(UId, Writers, undefined) of
         _ when Idx < SmallestIdx ->
             %% the smallest live index for the last snapshot is higher than
             %% this index, just drop it
             LastIdx = SmallestIdx - 1,
             State0#state{writers = Writers#{UId => {in_seq, LastIdx}}};
-        {ok, {_, PrevIdx}}
-          when PrevIdx0 =< PrevIdx orelse
+        {_, PrevIdx}
+          when ExpectedPrevIdx =< PrevIdx orelse
                Trunc ->
             %% if the passed in previous index is less than the last written
             %% index (gap detection) _or_ it is a truncation
             %% then we can proceed and write the entry
             write_data(Id, MtTid, Idx, Term, Entry, Trunc, SmallestIdx, State0);
-        error ->
+        undefined ->
             %% no state for the UId is known so go ahead and write
             write_data(Id, MtTid, Idx, Term, Entry, false, SmallestIdx, State0);
-        {ok, {out_of_seq, _}} ->
+        {out_of_seq, _} ->
             % writer is out of seq simply ignore drop the write
             % TODO: capture metric for dropped writes?
             State0;
-        {ok, {in_seq, PrevIdx}} ->
+        {in_seq, PrevIdx} ->
             % writer was in seq but has sent an out of seq entry
             % notify writer
             ?DEBUG("WAL in ~ts: requesting resend from `~w`, "
-                   "last idx ~b idx received ~b",
-                   [Conf#conf.system, UId, PrevIdx, Idx]),
+                   "last idx ~b idx received (~b,~b)",
+                   [Conf#conf.system, UId, PrevIdx, ExpectedPrevIdx, Idx]),
             Pid ! {ra_log_event, {resend_write, PrevIdx + 1}},
             State0#state{writers = Writers#{UId => {out_of_seq, PrevIdx}}}
     end;
@@ -538,32 +538,28 @@ incr_batch(#batch{num_writes = Writes,
                   waiting = Waiting0,
                   pending = Pend} = Batch,
            UId, Pid, MT_TID = MtTid,
-           Idx, TERM = Term, Data, SmallestIdx) ->
+           Idx, TERM = Term, Data, SmallestLiveIdx) ->
     Waiting = case Waiting0 of
                   #{Pid := #batch_writer{term = TERM,
                                          tid = MT_TID,
-                                         range = Range0
-                                        } = W} ->
+                                         seq = Seq0} = W} ->
                       %% The Tid and term is the same so add to current batch_writer
-                      Range = ra_range:extend(Idx, ra_range:truncate(SmallestIdx - 1,
-                                                                     Range0)),
+                      Range = ra_seq:append(Idx, Seq0),
                       %% TODO: range nees to become a ra_seq so that we can
                       %% capture sparse writes correctly
-                      Waiting0#{Pid => W#batch_writer{range = Range,
-                                                      smallest_live_idx = SmallestIdx,
-                                                      term = Term
-                                                     }};
+                      Waiting0#{Pid => W#batch_writer{seq = Range,
+                                                      smallest_live_idx = SmallestLiveIdx,
+                                                      term = Term}};
                   _ ->
                       %% The tid is different, open a new batch writer for the
                       %% new tid and term
                       PrevBatchWriter = maps:get(Pid, Waiting0, undefined),
-                      Writer = #batch_writer{smallest_live_idx = SmallestIdx,
+                      Writer = #batch_writer{smallest_live_idx = SmallestLiveIdx,
                                              tid = MtTid,
-                                             range = ra_range:new(Idx),
+                                             seq = [Idx],
                                              uid = UId,
                                              term = Term,
-                                             old = PrevBatchWriter
-                                            },
+                                             old = PrevBatchWriter},
                       Waiting0#{Pid => Writer}
               end,
 
@@ -699,7 +695,7 @@ complete_batch(#state{batch = #batch{waiting = Waiting,
 complete_batch_writer(Pid, #batch_writer{smallest_live_idx = SmallestIdx,
                                          tid = MtTid,
                                          uid = UId,
-                                         range = Range,
+                                         seq = Range,
                                          term = Term,
                                          old = undefined}, Ranges) ->
     Pid ! {ra_log_event, {written, Term, Range}},
@@ -972,23 +968,24 @@ should_roll_wal(#state{conf = #conf{max_entries = MaxEntries},
 smallest_live_index(#conf{ra_log_snapshot_state_tid = Tid}, ServerUId) ->
     ra_log_snapshot_state:smallest(Tid, ServerUId).
 
-update_ranges(Ranges, UId, MtTid, SmallestIdx, {Start, _} = AddRange) ->
+update_ranges(Ranges, UId, MtTid, SmallestIdx, AddSeq) ->
     case Ranges of
-        #{UId := [{MtTid, Range0} | Rem]} ->
+        #{UId := [{MtTid, Seq0} | Rem]} ->
             %% SmallestIdx might have moved to we truncate the old range first
             %% before extending
-            Range1 = ra_range:truncate(SmallestIdx - 1, Range0),
+            Seq1 = ra_seq:floor(SmallestIdx, Seq0),
             %% limit the old range by the add end start as in some resend
             %% cases we may have got back before the prior range.
-            Range = ra_range:add(AddRange, ra_range:limit(Start, Range1)),
-            Ranges#{UId => [{MtTid, Range} | Rem]};
-        #{UId := [{OldMtTid, OldMtRange} | Rem]} ->
+            Seq = ra_seq:add(AddSeq, Seq1),
+            Ranges#{UId => [{MtTid, Seq} | Rem]};
+        #{UId := [{OldMtTid, OldMtSeq} | Rem]} ->
             %% new Tid, need to add a new range record for this
-            Ranges#{UId => [{MtTid, AddRange},
-                            ra_range:truncate(SmallestIdx - 1, {OldMtTid, OldMtRange})
+            Ranges#{UId => [{MtTid, AddSeq},
+                            {OldMtTid,
+                             ra_seq:floor(SmallestIdx, OldMtSeq)}
                             | Rem]};
         _ ->
-            Ranges#{UId => [{MtTid, AddRange}]}
+            Ranges#{UId => [{MtTid, AddSeq}]}
     end.
 
 recover_entry(Names, UId, {Idx, _, _} = Entry, SmallestIdx,
@@ -1005,7 +1002,7 @@ recover_entry(Names, UId, {Idx, _, _} = Entry, SmallestIdx,
     case ra_mt:insert(Entry, Mt0) of
         {ok, Mt1} ->
             Ranges = update_ranges(Ranges0, UId, ra_mt:tid(Mt1),
-                                   SmallestIdx, ra_range:new(Idx)),
+                                   SmallestIdx, [Idx]),
             {ok, State#recovery{ranges = Ranges,
                                 writers = Writers#{UId => {in_seq, Idx}},
                                 tables = Tables#{UId => Mt1}}};
@@ -1037,7 +1034,7 @@ recover_entry(Names, UId, {Idx, Term, _}, SmallestIdx,
                                 tables = Tables#{UId => Mt0}}};
         Tid ->
             Ranges = update_ranges(Ranges0, UId, Tid,
-                                   SmallestIdx, ra_range:new(Idx)),
+                                   SmallestIdx, [Idx]),
             {ok, State#recovery{ranges = Ranges,
                                 writers = Writers#{UId => {in_seq, Idx}},
                                 tables = Tables#{UId => Mt0}}}
diff --git a/src/ra_range.erl b/src/ra_range.erl
@@ -25,6 +25,13 @@
 
 -export_type([range/0]).
 
+-define(IS_RANGE(R), ((is_tuple(R) andalso
+                       tuple_size(R) == 2 andalso
+                       is_integer(element(1, R)) andalso
+                       is_integer(element(2, R))) orelse
+                      R == undefined)).
+
+
 -spec new(ra:index()) -> range().
 new(Start) when is_integer(Start) ->
     {Start, Start}.
@@ -71,14 +78,17 @@ limit(CeilExcl, Range)
 -spec truncate(ra:index(), range()) -> range().
 truncate(UpToIncl, {_Start, End})
   when is_integer(UpToIncl) andalso
+       is_integer(End) andalso
        UpToIncl >= End ->
     undefined;
 truncate(UpToIncl, {Start, End})
   when is_integer(UpToIncl) andalso
+       is_integer(Start) andalso
        UpToIncl >= Start ->
     {UpToIncl + 1, End};
 truncate(UpToIncl, Range)
-  when is_integer(UpToIncl) ->
+  when is_integer(UpToIncl) andalso
+       ?IS_RANGE(Range) ->
     Range.
 
 size(undefined) ->
diff --git a/src/ra_seq.erl b/src/ra_seq.erl
@@ -1,9 +1,126 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2017-2025 Broadcom. All Rights Reserved. The term Broadcom refers to Broadcom Inc. and/or its subsidiaries.
 -module(ra_seq).
 
 %% open type
+%% sequences are ordered high -> low but ranges are ordered
+%% {low, high} so a typical sequence could look like
+%% [55, {20, 52}, 3]
 -type state() :: [ra:index() | ra:range()].
 
 -export_type([state/0]).
 
 
+-export([
+         append/2,
+         from_list/1,
+         floor/2,
+         limit/2,
+         add/2,
+         fold/3
+        ]).
 
+-spec append(ra:index(), state()) -> state().
+append(Idx, [IdxN1, IdxN2 | Rem])
+  when Idx == IdxN1 + 1 andalso
+       Idx == IdxN2 + 2 ->
+    %% we can compact into a range
+    [{IdxN2, Idx} | Rem];
+append(Idx, [{IdxN, IdxN1} | Rem])
+  when Idx == IdxN1 + 1 ->
+    %% Extend the raage
+    [{IdxN, Idx} | Rem];
+append(Idx, [])
+  when is_integer(Idx) ->
+    [Idx];
+append(Idx, [Prev | _] = Seq)
+  when is_integer(Idx) andalso
+       ((is_tuple(Prev) andalso
+         Idx > element(2, Prev)) orelse
+        Idx > Prev) ->
+    [Idx | Seq].
+
+-spec from_list([ra:index()]) -> state().
+from_list(L) ->
+    lists:foldl(fun append/2, [], lists:sort(L)).
+
+-spec floor(ra:index(), state()) -> state().
+floor(FloorIdxIncl, Seq) ->
+    %% TODO: assert appendable
+    %% for now assume appendable
+    floor0(FloorIdxIncl, Seq, []).
+
+
+limit(CeilIdx, [Last | Rem])
+  when is_integer(Last) andalso
+       Last > CeilIdx ->
+    limit(CeilIdx, Rem);
+limit(CeilIdx, [{_, _} = T | Rem]) ->
+    case ra_range:limit(CeilIdx + 1, T) of
+        undefined ->
+            limit(CeilIdx, Rem);
+        {I, I} ->
+            [I | Rem];
+        {I, I2} when I == I2 - 1 ->
+            [I2, I | Rem];
+        NewRange ->
+            [NewRange | Rem]
+    end;
+limit(_CeilIdx, Seq) ->
+    Seq.
+
+-spec add(state(), state()) -> state().
+add([], Seq2) ->
+    Seq2;
+add(Seq1, Seq2) ->
+    Fst = case lists:last(Seq1) of
+              {I, _} -> I;
+              I -> I
+          end,
+    fold(fun append/2, limit(Fst - 1, Seq2), Seq1).
+
+
+-spec fold(fun ((ra:index(), Acc) -> Acc), Acc, state())
+-> Acc when Acc :: term().
+fold(Fun, Acc0, Seq) ->
+    %% TODO: factor out the lists:seq/2
+    lists:foldr(
+      fun ({S, E}, Acc) ->
+              lists:foldl(Fun, Acc, lists:seq(S, E));
+          (Idx, Acc) ->
+              Fun(Idx, Acc)
+      end, Acc0, Seq).
+
+%% internal functions
+
+floor0(FloorIdx, [Last | Rem], Acc)
+  when is_integer(Last) andalso
+       Last >= FloorIdx ->
+    floor0(FloorIdx, Rem, [Last | Acc]);
+floor0(FloorIdx, [{_, _} = T | Rem], Acc) ->
+    case ra_range:truncate(FloorIdx - 1, T) of
+        undefined ->
+            lists:reverse(Acc);
+        {I, I} ->
+            floor0(FloorIdx, Rem, [I | Acc]);
+        {I, I2} when I == I2 - 1 ->
+            floor0(FloorIdx, Rem, [I, I2 | Acc]);
+        NewRange ->
+            floor0(FloorIdx, Rem, [NewRange | Acc])
+    end;
+floor0(_FloorIdx, _Seq, Acc) ->
+    lists:reverse(Acc).
+
+% first_index(Seq) ->
+%     last_index(lists:reverse(Seq)).
+
+% last_index([{_, I} | _]) ->
+%     I;
+% last_index([I | _])
+%   when is_integer(I) ->
+%     I;
+% last_index([]) ->
+%     undefined.
diff --git a/test/ra_log_wal_SUITE.erl b/test/ra_log_wal_SUITE.erl
@@ -159,12 +159,14 @@ sparse_writes(Config) ->
     {ok, Pid} = ra_log_wal:start_link(Conf),
     {ok, _} = ra_log_wal:write(Pid, WriterId, Tid, 11, 12, 1, "value"),
     ok = await_written(WriterId, 1, {12, 12}),
-    debugger:start(),
-    int:i(ra_log_wal),
-    int:break(ra_log_wal, 975),
+    % debugger:start(),
+    % int:i(ra_log_wal),
+    % int:break(ra_log_wal, 975),
     timer:sleep(1000),
+    %% write  a "sparse write" at index 15 but reference 12 as the last
+    %% one
     {ok, _} = ra_log_wal:write(Pid, WriterId, Tid, 12, 15, 1, "value2"),
-    timer:sleep(200000),
+    % timer:sleep(200000),
     ok = await_written(WriterId, 1, {15, 15}),
     ra_log_wal:force_roll_over(Pid),
     receive
diff --git a/test/ra_seq_SUITE.erl b/test/ra_seq_SUITE.erl
