Merge pull request #101 from rabbitmq/init-first-offset-atomic

Ensure a writer|replica start initialises first offset

Author: acogoluegnes

src/osiris.erl

@@ -25,7 +25,8 @@
          start_writer/1,
          start_replica/2,
          delete_cluster/1,
-         configure_logger/1]).
+         configure_logger/1,
+         get_stats/1]).
 
 %% holds static or rarely changing fields
 -record(cfg, {}).
@@ -273,3 +274,12 @@ start_replicas(Config, [Node | Nodes], ReplicaPids) ->
 configure_logger(Module) ->
     persistent_term:put('$osiris_logger', Module).
 
+-spec get_stats(pid()) -> #{committed_chunk_id => integer(),
+                            first_chunk_id => integer()}.
+get_stats(Pid)
+    when node(Pid) =:= node() ->
+    {ok, #{offset_ref := ORef}} = gen:call(Pid, '$gen_call', get_reader_context),
+    #{committed_chunk_id => atomics:get(ORef, 1),
+      first_chunk_id => atomics:get(ORef, 2)};
+get_stats(Pid) when is_pid(Pid) ->
+    erpc:call(node(Pid), ?MODULE, ?FUNCTION_NAME, [Pid]).

src/osiris_log.erl

@@ -486,6 +486,7 @@ init(#{dir := Dir,
                              _ ->
                                  0
                          end,
+            FirstOffsetFun(NextOffset - 1),
             open_new_segment(#?MODULE{cfg = Cfg,
                                       mode =
                                           #write{type = WriterType,
@@ -517,6 +518,7 @@ init(#{dir := Dir,
             counters:put(Cnt, ?C_FIRST_TIMESTAMP, FstTs),
             counters:put(Cnt, ?C_OFFSET, LastChId + LastNum - 1),
             counters:put(Cnt, ?C_SEGMENTS, NumSegments),
+            FirstOffsetFun(FstChId),
             ?DEBUG("~s:~s/~b: ~s next offset ~b first offset ~b",
                    [?MODULE,
                     ?FUNCTION_NAME,
@@ -551,6 +553,7 @@ init(#{dir := Dir,
             %% here too?
             {ok, _} = file:position(SegFd, eof),
             {ok, _} = file:position(IdxFd, eof),
+            FirstOffsetFun(-1),
             #?MODULE{cfg = Cfg,
                      mode =
                          #write{type = WriterType,
@@ -2112,7 +2115,7 @@ write_chunk(Chunk,
             Timestamp,
             Epoch,
             NumRecords,
-            #?MODULE{cfg = #cfg{counter = CntRef},
+            #?MODULE{cfg = #cfg{counter = CntRef} = Cfg,
                      fd = Fd,
                      index_fd = IdxFd,
                      mode =
@@ -2144,6 +2147,7 @@ write_chunk(Chunk,
             %% update counters
             counters:put(CntRef, ?C_OFFSET, NextOffset - 1),
             counters:add(CntRef, ?C_CHUNKS, 1),
+            maybe_set_first_offset(Next, Cfg),
             State#?MODULE{mode =
                           Write#write{tail_info =
                                       {NextOffset,
@@ -2152,6 +2156,12 @@ write_chunk(Chunk,
                                                       SegSizeChunks + 1}}}
     end.
 
+
+maybe_set_first_offset(0, #cfg{first_offset_fun = Fun}) ->
+    Fun(0);
+maybe_set_first_offset(_, _Cfg) ->
+    ok.
+
 max_segment_size_reached(
   #?MODULE{mode = #write{segment_size = {CurrentSizeBytes,
                                          CurrentSizeChunks}},

src/osiris_writer.erl

@@ -161,12 +161,13 @@ handle_continue(#{name := Name,
     process_flag(trap_exit, true),
     process_flag(message_queue_data, off_heap),
     ORef = atomics:new(2, [{signed, true}]),
+    atomics:put(ORef, 2, -1),
     CntName = {?MODULE, ExtRef},
     Log = osiris_log:init(Config#{dir => Dir,
                                   first_offset_fun =>
-                                  fun (Fst) ->
-                                          atomics:put(ORef, 2, Fst)
-                                  end,
+                                      fun (Fst) ->
+                                              atomics:put(ORef, 2, Fst)
+                                      end,
                                   counter_spec =>
                                       {CntName, ?ADD_COUNTER_FIELDS}}),
     Trk = osiris_log:recover_tracking(Log),

test/osiris_SUITE.erl

@@ -621,7 +621,9 @@ cluster_restart(Config) ->
           epoch => 1,
           replica_nodes => Replicas,
           leader_node => LeaderNode},
-    {ok, #{leader_pid := Leader} = Conf} = osiris:start_cluster(Conf0),
+    {ok, #{leader_pid := Leader,
+           replica_pids := [Replica1, Replica2]} = Conf} =
+        osiris:start_cluster(Conf0),
     WriterId = <<"wid1">>,
     ok = osiris:write(Leader, WriterId, 42, <<"before-restart">>),
     receive
@@ -632,12 +634,36 @@ cluster_restart(Config) ->
         exit(osiris_written_timeout)
     end,
 
+    timer:sleep(1),
+    RangesBeforeRecovery =
+    [begin
+         #{committed_chunk_id := E,
+           first_chunk_id := S} = osiris:get_stats(P),
+         {S, E}
+     end || P <- [Leader, Replica1, Replica2]],
+
+    ?assertEqual([{0,0}, {0,0}, {0,0}], RangesBeforeRecovery),
+
     osiris:stop_cluster(Conf),
-    {ok, #{leader_pid := Leader1}} =
+    {ok, #{leader_pid := Leader1,
+           replica_pids := [ReplicaPid1, ReplicaPid2]}} =
         osiris:start_cluster(Conf0#{epoch => 2}),
     %% give leader some time to discover the committed offset
     timer:sleep(1000),
-    ok = validate_log(Leader1, [{0, <<"before-restart">>}]),
+
+    %% get all log ranges after recovery for each member and validate they are all
+    %% the same
+    RangesAfterRecovery =
+    [begin
+         #{committed_chunk_id := E,
+           first_chunk_id := S} = osiris:get_stats(P),
+         {S, E}
+     end || P <- [Leader1, ReplicaPid1, ReplicaPid2]],
+    ?assertEqual([{0,0}, {0,0}, {0,0}], RangesAfterRecovery),
+    ct:pal("~p ~p", [RangesBeforeRecovery, RangesAfterRecovery]),
+    ok = validate_log_offset_reader(Leader1, [{0, <<"before-restart">>}]),
+    ok = validate_log_offset_reader(ReplicaPid1, [{0, <<"before-restart">>}]),
+    ok = validate_log_offset_reader(ReplicaPid2, [{0, <<"before-restart">>}]),
 
     ok = osiris:write(Leader1, WriterId, 43, <<"after-restart">>),
     receive
@@ -648,9 +674,11 @@ cluster_restart(Config) ->
         exit(osiris_written_timeout)
     end,
 
-    ok =
-        validate_log(Leader1,
-                     [{0, <<"before-restart">>}, {1, <<"after-restart">>}]),
+    ExpectedLog = [{0, <<"before-restart">>}, {1, <<"after-restart">>}],
+
+    ok = validate_log_offset_reader(Leader1, ExpectedLog),
+    ok = validate_log_offset_reader(ReplicaPid1, ExpectedLog),
+    ok = validate_log_offset_reader(ReplicaPid2, ExpectedLog),
     [stop_peer(Ref) || {Ref, _} <- PeerStates],
     ok.
 
@@ -1835,7 +1863,8 @@ search_paths() ->
 validate_log(Leader, Exp) when is_pid(Leader) ->
     case node(Leader) == node() of
         true ->
-            {ok, Log0} = osiris_writer:init_data_reader(Leader, {0, empty}, #{counter_spec => {'test', []}}),
+            {ok, Log0} = osiris_writer:init_data_reader(Leader, {0, empty},
+                                                        #{counter_spec => {'test', []}}),
             validate_log(Log0, Exp);
         false ->
             ok = erpc:call(node(Leader), ?MODULE, ?FUNCTION_NAME, [Leader, Exp])
@@ -1851,6 +1880,25 @@ validate_log(Log0, Expected) ->
             validate_log(Log, Expected -- Entries)
     end.
 
+validate_log_offset_reader(Leader, Exp) when is_pid(Leader) ->
+    case node(Leader) == node() of
+        true ->
+            {ok, Log0} = osiris:init_reader(Leader, first, {'test', []}),
+            validate_log_offset_reader(Log0, Exp);
+        false ->
+            ok = erpc:call(node(Leader), ?MODULE, ?FUNCTION_NAME, [Leader, Exp])
+    end;
+validate_log_offset_reader(Log, []) ->
+    ok = osiris_log:close(Log),
+    ok;
+validate_log_offset_reader(Log0, Expected) ->
+    case osiris_log:read_chunk_parsed(Log0) of
+        {end_of_stream, _} ->
+            ct:fail("validate log failed, rem: ~p", [Expected]);
+        {Entries, Log} ->
+            validate_log_offset_reader(Log, Expected -- Entries)
+    end.
+
 print_counters() ->
     [begin
          ct:pal("~w counters ~p",