Fix bug in replica chunk acceptance when filters are used.

If osiris_replica:parse_chunk ended up parsing an chunk iolist where
the first element in the list is exactly 48 bytes (the size of the
osiris chunk header) it would end up crashing the call to
osiris_log:accept_chunk as this assumed the filter data, when included,
would always be in the first list element.

Author: kjnilsson

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),