Merge pull request #187 from rabbitmq/accept-chunk-with-filter-fix

Fix bug in replica chunk acceptance when filters are used.

Author: michaelklishin

src/osiris_log.erl

@@ -778,13 +778,12 @@ accept_chunk([<<?MAGIC:4/unsigned,
                 _TrailerSize:32/unsigned,
                 FilterSize:8/unsigned,
                 _Reserved:24,
-                _Filter:FilterSize/binary,
                 Data/binary>>
               | DataParts] =
                  Chunk,
              #?MODULE{cfg = #cfg{}, mode = #write{tail_info = {Next, _}}} =
                  State0) ->
-    DataAndTrailer = [Data | DataParts],
+    DataAndTrailer = skip(FilterSize, [Data | DataParts]),
     validate_crc(Next, Crc, part(DataSize, DataAndTrailer)),
     %% assertion
     % true = iolist_size(DataAndTrailer) == (DataSize + TrailerSize),
@@ -2788,6 +2787,17 @@ part(Len, [B | L]) when Len > 0 ->
             [binary:part(B, {0, Len})]
     end.
 
+skip(0, L) ->
+    L;
+skip(Len, [B | L]) when Len > 0 ->
+    S = byte_size(B),
+    case Len < S of
+        true ->
+            [binary:part(B, {S, Len - S}) | L];
+        false ->
+            skip(Len - S, L)
+    end.
+
 -spec recover_tracking(state()) ->
     osiris_tracking:state().
 recover_tracking(#?MODULE{cfg = #cfg{directory = Dir,
@@ -3232,6 +3242,7 @@ list_dir(Dir) ->
     end.
 
 -ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
 
 
 part_test() ->
@@ -3240,4 +3251,12 @@ part_test() ->
     [<<"AB">>, <<"CDEF">>] = part(6, [<<"AB">>, <<"CDEF">>]),
     ok.
 
+skip_test() ->
+    [<<"EF">>] = skip(4, [<<"ABCDEF">>]),
+    [<<"B">>, <<"CDEF">>] = skip(1, [<<"AB">>, <<"CDEF">>]),
+    [<<"CDEF">>] = skip(2, [<<"AB">>, <<"CDEF">>]),
+    [<<"DEF">>] = skip(3, [<<"AB">>, <<"CDEF">>]),
+    [<<"AB">>, <<"CDEF">>] = skip(0, [<<"AB">>, <<"CDEF">>]),
+    ok.
+
 -endif.

test/osiris_log_SUITE.erl

@@ -72,6 +72,7 @@ all_tests() ->
      % truncate_multi_segment,
      accept_chunk,
      accept_chunk_inital_offset,
+     accept_chunk_iolist_header_in_first_element,
      init_acceptor_truncates_tail,
      accept_chunk_truncates_tail,
      accept_chunk_does_not_truncate_tail_in_same_epoch,
@@ -1086,6 +1087,19 @@ accept_chunk_inital_offset(Config) ->
 
     ok.
 
+accept_chunk_iolist_header_in_first_element(Config) ->
+    ok = logger:set_primary_config(level, all),
+    Conf = ?config(osiris_conf, Config),
+
+
+    <<Head:48/binary, Res/binary>> = fake_chunk([{<<"filter">>, <<"blob1">>}],
+                                                ?LINE, 1, 100),
+    Ch1 = [Head, Res],
+    F0 = osiris_log:init(Conf#{initial_offset => 100}, acceptor),
+    %% this should not crash
+    _F1 = osiris_log:accept_chunk(Ch1, F0),
+    ok.
+
 accept_chunk(Config) ->
     ok = logger:set_primary_config(level, all),
     Conf = ?config(osiris_conf, Config),