Fix snapshot installation CRC failure

kjnilsson · kjnilsson · commit a0c240da923d · 2023-05-09T12:47:20.000+01:00
Due to the unwise use of term_to_binary rather than using the original
binary data in the replication of snapshot states.

This change:

Introduces a new optional ra_snapshot callback: context/0

This is called by the sending Ra leader node to discover context and
capabilities of the receiver. In this case it is used to indicate if
the receiver is capabable of receiving the entire snapshot file.

Receiving the entire file is the updated approach that ensures the
CRC check will be done on the same binary data it was generated from.

If the receiver does not have the context/0 callback or does not indicate
support the old approach of sending the deserialised metat data map and any
data following that is used.

When a snapshot is received from an old node (i.e. _not_ including
the entire file) the receiver will not validate the checksum (as it may
fail due to differences in map serialisation) and instead patch up it's
local file with it's own calculcated checksum.

The scenario where a snapshot taken by a newer version of OTP and is then
sent to a member using the old code cannot be handled and the old node will
fail at snapshot checksum validation.
diff --git a/src/ra_log_snapshot.erl b/src/ra_log_snapshot.erl
@@ -15,11 +15,12 @@
          begin_accept/2,
          accept_chunk/2,
          complete_accept/2,
-         begin_read/1,
+         begin_read/2,
          read_chunk/3,
          recover/1,
          validate/1,
-         read_meta/1
+         read_meta/1,
+         context/0
          ]).
 
 -define(MAGIC, "RASN").
@@ -68,34 +69,51 @@ begin_accept(SnapDir, Meta) ->
                          Data]),
     {ok, {PartialCrc, Fd}}.
 
+accept_chunk(<<?MAGIC, ?VERSION:8/unsigned, Crc:32/integer,
+               Rest/binary>> = Chunk, {_PartialCrc, Fd}) ->
+    % ensure we overwrite the existing header when we are receiving the
+    % full file
+    PartialCrc = erlang:crc32(Rest),
+    {ok, 0} = file:position(Fd, 0),
+    ok = file:write(Fd, Chunk),
+    {ok, {PartialCrc, Crc, Fd}};
 accept_chunk(Chunk, {PartialCrc, Fd}) ->
-    <<Crc:32/integer, Rest/binary>> = Chunk,
-    accept_chunk(Rest, {PartialCrc, Crc, Fd});
+    %% compatibility clause where we did not receive the full file
+    %% do not validate Crc due to OTP 26 map key ordering changes
+    <<_Crc:32/integer, Rest/binary>> = Chunk,
+    accept_chunk(Rest, {PartialCrc, undefined, Fd});
 accept_chunk(Chunk, {PartialCrc0, Crc, Fd}) ->
     ok = file:write(Fd, Chunk),
     PartialCrc = erlang:crc32(PartialCrc0, Chunk),
     {ok, {PartialCrc, Crc, Fd}}.
 
-complete_accept(Chunk, {PartialCrc, Fd}) ->
-    <<Crc:32/integer, Rest/binary>> = Chunk,
-    complete_accept(Rest, {PartialCrc, Crc, Fd});
-complete_accept(Chunk, {PartialCrc0, Crc, Fd}) ->
-    ok = file:write(Fd, Chunk),
-    ok = file:pwrite(Fd, 5, <<Crc:32/integer>>),
-    Crc = erlang:crc32(PartialCrc0, Chunk),
+complete_accept(Chunk, St0) ->
+    {ok, {CalculatedCrc, Crc, Fd}} = accept_chunk(Chunk, St0),
+    CrcToWrite = case Crc of
+                     undefined ->
+                         CalculatedCrc;
+                     _ ->
+                         Crc
+                 end,
+    ok = file:pwrite(Fd, 5, <<CrcToWrite:32/integer>>),
     ok = file:sync(Fd),
     ok = file:close(Fd),
+    CalculatedCrc = CrcToWrite,
     ok.
 
-begin_read(Dir) ->
+begin_read(Dir, Context) ->
     File = filename(Dir),
     case file:open(File, [read, binary, raw]) of
         {ok, Fd} ->
             case read_meta_internal(Fd) of
+                {ok, Meta, _Crc}
+                  when map_get(can_accept_full_file, Context) ->
+                    {ok, Eof} = file:position(Fd, eof),
+                    {ok, Meta, {0, Eof, Fd}};
                 {ok, Meta, Crc} ->
-                    {ok, DataStart} = file:position(Fd, cur),
+                    {ok, Cur} = file:position(Fd, cur),
                     {ok, Eof} = file:position(Fd, eof),
-                    {ok, Meta, {Crc, {DataStart, Eof, Fd}}};
+                    {ok, Meta, {Crc, {Cur, Eof, Fd}}};
                 {error, _} = Err ->
                     _ = file:close(Fd),
                     Err
@@ -105,6 +123,7 @@ begin_read(Dir) ->
     end.
 
 read_chunk({Crc, ReadState}, Size, Dir) when is_integer(Crc) ->
+    %% this the compatibility read mode for old snapshot receivers
     case read_chunk(ReadState, Size - 4, Dir) of
         {ok, Data, ReadState1} ->
             {ok, <<Crc:32/integer, Data/binary>>, ReadState1};
@@ -156,25 +175,28 @@ validate(Dir) ->
 %% entire binary body. NB: this does not do checksum validation.
 -spec read_meta(file:filename()) ->
     {ok, meta()} | {error, invalid_format |
-                          {invalid_version, integer()} |
-                          checksum_error |
-                          file_err()}.
+                    {invalid_version, integer()} |
+                    checksum_error |
+                    file_err()}.
 read_meta(Dir) ->
     File = filename(Dir),
     case file:open(File, [read, binary, raw]) of
         {ok, Fd} ->
             case read_meta_internal(Fd) of
-                {ok, Meta, _} ->
+                {ok, Meta, _Crc} ->
                     _ = file:close(Fd),
                     {ok, Meta};
-                {error, _} = Err ->
+                Err ->
                     _ = file:close(Fd),
                     Err
-            end;
-        Err ->
-            Err
+            end
     end.
 
+-spec context() -> map().
+context() ->
+    #{can_accept_full_file => true}.
+
+
 %% Internal
 
 read_meta_internal(Fd) ->
diff --git a/src/ra_server_proc.erl b/src/ra_server_proc.erl
@@ -1565,11 +1565,12 @@ fold_log(From, Fun, Term, State) ->
 
 send_snapshots(Me, Id, Term, {_, ToNode} = To, ChunkSize,
                InstallTimeout, SnapState, Machine) ->
+    Context = ra_snapshot:context(SnapState, ToNode),
     {ok, #{machine_version := SnapMacVer} = Meta, ReadState} =
-        ra_snapshot:begin_read(SnapState),
+        ra_snapshot:begin_read(SnapState, Context),
 
     %% only send the snapshot if the target server can accept it
-    TheirMacVer = rpc:call(ToNode, ra_machine, version, [Machine]),
+    TheirMacVer = erpc:call(ToNode, ra_machine, version, [Machine]),
 
     case SnapMacVer > TheirMacVer of
         true ->
diff --git a/src/ra_snapshot.erl b/src/ra_snapshot.erl
@@ -16,7 +16,7 @@
 -export([
          recover/1,
          read_meta/2,
-         begin_read/1,
+         begin_read/2,
          read_chunk/3,
          delete/2,
 
@@ -36,6 +36,8 @@
          accept_chunk/4,
          abort_accept/1,
 
+         context/2,
+
          handle_down/3,
          current_snapshot_dir/1
         ]).
@@ -68,6 +70,8 @@
 
 -export_type([state/0]).
 
+-optional_callbacks([context/0]).
+
 %% Side effect function
 %% Turn the current state into immutable reference.
 -callback prepare(Index :: ra_index(),
@@ -87,7 +91,9 @@
 
 %% Read the snapshot metadata and initialise a read state used in read_chunk/1
 %% The read state should contain all the information required to read a chunk
--callback begin_read(Location :: file:filename()) ->
+%% The Context is the map returned by the context/0 callback
+%% This can be used to inform the sender of receive capabilities.
+-callback begin_read(Location :: file:filename(), Context :: map()) ->
     {ok, Meta :: meta(), ReadState :: term()}
     | {error, term()}.
 
@@ -131,6 +137,8 @@
             file_err() |
             term()}.
 
+-callback context() -> map().
+
 -spec init(ra_uid(), module(), file:filename()) ->
     state().
 init(UId, Mod, File) ->
@@ -310,6 +318,18 @@ abort_accept(#?MODULE{accepting = #accept{idxterm = {Idx, Term}},
     ok = delete(Dir, {Idx, Term}),
     State#?MODULE{accepting = undefined}.
 
+%% get the snapshot capabilities context of a remote node
+-spec context(state(), node()) -> map().
+context(#?MODULE{module = Mod}, Node) ->
+    try erpc:call(Node, Mod, ?FUNCTION_NAME, []) of
+        Result ->
+            Result
+    catch
+        error:{exception, undef, _} ->
+            #{}
+    end.
+
+
 
 -spec handle_down(pid(), Info :: term(), state()) ->
     state().
@@ -335,14 +355,15 @@ delete(Dir, {Idx, Term}) ->
     ok = ra_lib:recursive_delete(SnapDir),
     ok.
 
--spec begin_read(State :: state()) ->
+-spec begin_read(State :: state(), Context :: map()) ->
     {ok, Meta :: meta(), ReadState} |
     {error, term()} when ReadState :: term().
 begin_read(#?MODULE{module = Mod,
                     directory = Dir,
-                    current = {Idx, Term}}) ->
+                    current = {Idx, Term}},
+          Context) when is_map(Context) ->
     Location = make_snapshot_dir(Dir, Idx, Term),
-    Mod:begin_read(Location).
+    Mod:begin_read(Location, Context).
 
 
 -spec read_chunk(ReadState, ChunkSizeBytes :: non_neg_integer(),
diff --git a/test/ra_SUITE.erl b/test/ra_SUITE.erl
@@ -674,6 +674,18 @@ server_catches_up(Config) ->
     terminate_cluster([N3 | InitialNodes]).
 
 snapshot_installation(Config) ->
+    ra_env:configure_logger(logger),
+    ok = logger:set_primary_config(level, debug),
+
+    LogFile = filename:join(?config(priv_dir, Config), "ra.log"),
+    SaslFile = filename:join(?config(priv_dir, Config), "ra_sasl.log"),
+    logger:add_handler(ra_handler, logger_std_h,
+                       #{config => #{file => LogFile}}),
+    application:load(sasl),
+    application:set_env(sasl, sasl_error_logger, {file, SaslFile}),
+    application:stop(sasl),
+    application:start(sasl),
+    _ = error_logger:tty(false),
     N1 = nth_server_name(Config, 1),
     N2 = nth_server_name(Config, 2),
     N3 = nth_server_name(Config, 3),
@@ -710,9 +722,9 @@ snapshot_installation(Config) ->
                      length(filelib:wildcard(
                               filename:join([Dir, "snapshots", "*"]))) > 0
              end,
-    ?assert(try_n_times(fun () -> TryFun(N2Dir) end, 20)),
-    ?assert(try_n_times(fun () -> TryFun(N3Dir) end, 20)),
-    ?assert(try_n_times(fun () -> TryFun(N1Dir) end, 20)),
+    ?assert(try_n_times(fun () -> TryFun(N2Dir) end, 100)),
+    ?assert(try_n_times(fun () -> TryFun(N3Dir) end, 100)),
+    ?assert(try_n_times(fun () -> TryFun(N1Dir) end, 100)),
 
     % then do some more
     [begin
diff --git a/test/ra_log_2_SUITE.erl b/test/ra_log_2_SUITE.erl
@@ -678,7 +678,8 @@ snapshot_written_after_installation(Config) ->
                          end,
 
     Meta = meta(15, 2, [?N1]),
-    Chunk = create_snapshot_chunk(Config, Meta),
+    Context = #{},
+    Chunk = create_snapshot_chunk(Config, Meta, Context),
     SnapState0 = ra_log:snapshot_state(Log2),
     {ok, SnapState1} = ra_snapshot:begin_accept(Meta, SnapState0),
     {ok, SnapState} = ra_snapshot:accept_chunk(Chunk, 1, last, SnapState1),
@@ -726,7 +727,7 @@ snapshot_installation(Config) ->
 
     %% create snapshot chunk
     Meta = meta(15, 2, [?N1]),
-    Chunk = create_snapshot_chunk(Config, Meta),
+    Chunk = create_snapshot_chunk(Config, Meta, #{}),
     SnapState0 = ra_log:snapshot_state(Log2),
     {ok, SnapState1} = ra_snapshot:begin_accept(Meta, SnapState0),
     {ok, SnapState} = ra_snapshot:accept_chunk(Chunk, 1, last, SnapState1),
@@ -775,7 +776,7 @@ append_after_snapshot_installation(Config) ->
                              end),
     %% do snapshot
     Meta = meta(15, 2, [?N1]),
-    Chunk = create_snapshot_chunk(Config, Meta),
+    Chunk = create_snapshot_chunk(Config, Meta, #{}),
     SnapState0 = ra_log:snapshot_state(Log1),
     {ok, SnapState1} = ra_snapshot:begin_accept(Meta, SnapState0),
     {ok, SnapState} = ra_snapshot:accept_chunk(Chunk, 1, last, SnapState1),
@@ -806,7 +807,7 @@ written_event_after_snapshot_installation(Config) ->
     SnapIdx = 10,
     %% do snapshot in
     Meta = meta(SnapIdx, 2, [?N1]),
-    Chunk = create_snapshot_chunk(Config, Meta),
+    Chunk = create_snapshot_chunk(Config, Meta, #{}),
     SnapState0 = ra_log:snapshot_state(Log1),
     {ok, SnapState1} = ra_snapshot:begin_accept(Meta, SnapState0),
     {ok, SnapState} = ra_snapshot:accept_chunk(Chunk, 1, last, SnapState1),
@@ -1274,7 +1275,7 @@ meta(Idx, Term, Cluster) ->
       cluster => Cluster,
       machine_version => 1}.
 
-create_snapshot_chunk(Config, #{index := Idx} =  Meta) ->
+create_snapshot_chunk(Config, #{index := Idx} = Meta, Context) ->
     OthDir = filename:join(?config(priv_dir, Config), "snapshot_installation"),
     ok = ra_lib:make_dir(OthDir),
     Sn0 = ra_snapshot:init(<<"someotheruid_adsfasdf">>, ra_log_snapshot,
@@ -1288,7 +1289,7 @@ create_snapshot_chunk(Config, #{index := Idx} =  Meta) ->
         after 1000 ->
                   exit(snapshot_timeout)
         end,
-    {ok, Meta, ChunkSt} = ra_snapshot:begin_read(Sn2),
+    {ok, Meta, ChunkSt} = ra_snapshot:begin_read(Sn2, Context),
     {ok, Chunk, _} = ra_snapshot:read_chunk(ChunkSt, 1000000000, Sn2),
     Chunk.
 
diff --git a/test/ra_log_snapshot_SUITE.erl b/test/ra_log_snapshot_SUITE.erl
diff --git a/test/ra_snapshot_SUITE.erl b/test/ra_snapshot_SUITE.erl
