Merge pull request #154 from rabbitmq/next-with-filter-fix

Calculate the correct chunk size when filters are used.

Author: kjnilsson

src/osiris_log.erl

@@ -1922,7 +1922,8 @@ build_segment_info(SegFile, LastChunkPos, IdxFile) ->
            FirstChId:64/unsigned,
            _FirstCrc:32/integer,
            FirstSize:32/unsigned,
-           FirstTSize:32/unsigned,
+           FirstFSize:8/unsigned,
+           FirstTSize:24/unsigned,
            _/binary>>} ->
             case file:pread(Fd, LastChunkPos, ?HEADER_SIZE_B) of
                 {ok,
@@ -1939,33 +1940,34 @@ build_segment_info(SegFile, LastChunkPos, IdxFile) ->
                    LastTSize:32/unsigned,
                    LastFSize:8/unsigned,
                    _Reserved:24>>} ->
-                    Size = LastChunkPos + ?HEADER_SIZE_B + LastFSize + LastSize + LastTSize,
+                    LastChunkSize = LastFSize + LastSize + LastTSize,
+                    Size = LastChunkPos + ?HEADER_SIZE_B + LastChunkSize,
                     %% TODO: this file:position/2 all has no actual function and
                     %% is only used to emit a debug log. Remove?
                     {ok, Eof} = file:position(Fd, eof),
                     ?DEBUG_IF("~s: segment ~ts has trailing data ~w ~w",
                               [?MODULE, filename:basename(SegFile),
                                Size, Eof], Size =/= Eof),
                     _ = file:close(Fd),
+                    FstChInfo = #chunk_info{epoch = FirstEpoch,
+                                            timestamp = FirstTs,
+                                            id = FirstChId,
+                                            num = FirstNumRecords,
+                                            type = FirstChType,
+                                            size = FirstFSize + FirstSize + FirstTSize,
+                                            pos = ?LOG_HEADER_SIZE},
+                    LastChInfo = #chunk_info{epoch = LastEpoch,
+                                             timestamp = LastTs,
+                                             id = LastChId,
+                                             num = LastNumRecords,
+                                             type = LastChType,
+                                             size = LastChunkSize,
+                                             pos = LastChunkPos},
                     {ok, #seg_info{file = SegFile,
                                    index = IdxFile,
                                    size = Size,
-                                   first =
-                                   #chunk_info{epoch = FirstEpoch,
-                                               timestamp = FirstTs,
-                                               id = FirstChId,
-                                               num = FirstNumRecords,
-                                               type = FirstChType,
-                                               size = FirstSize + FirstTSize,
-                                               pos = ?LOG_HEADER_SIZE},
-                                   last =
-                                   #chunk_info{epoch = LastEpoch,
-                                               timestamp = LastTs,
-                                               id = LastChId,
-                                               num = LastNumRecords,
-                                               type = LastChType,
-                                               size = LastSize + LastTSize,
-                                               pos = LastChunkPos}}};
+                                   first = FstChInfo,
+                                   last = LastChInfo}};
                 _ ->
                     % last chunk is corrupted - try the previous one
                     _ = file:close(Fd),

test/osiris_log_SUITE.erl

@@ -32,6 +32,7 @@ all_tests() ->
      init_recover_with_writers,
      init_with_lower_epoch,
      write_batch,
+     write_with_filter_attach_next,
      write_batch_with_filter,
      write_batch_with_filters_variable_size,
      subbatch,
@@ -207,6 +208,25 @@ write_batch(Config) ->
     ?assertEqual(1, osiris_log:next_offset(S1)),
     ok.
 
+write_with_filter_attach_next(Config) ->
+    %% bug fix where the chunk_info size didn't include the filter size which
+    %% cause the next attach strategy to point to an invalid location in the
+    %% segment when filters were used.
+    Conf0 = ?config(osiris_conf, Config),
+    S0 = osiris_log:init(Conf0#{filter_size => 32}),
+    ?assertEqual(0, osiris_log:next_offset(S0)),
+    %% write an entry with a filter value and an entry without a filter value
+    {_, S1} = write_committed([<<"ho">>, {<<"banana">>, <<"hi">>}], S0),
+    Shared = osiris_log:get_shared(S1),
+    Conf = Conf0#{shared => Shared},
+    {ok, R0} = osiris_log:init_offset_reader(next, Conf),
+    {end_of_stream, R1} = osiris_log:read_chunk_parsed(R0),
+    %% then write a chunk without any filtred entries at all
+    {_, _S2} = write_committed([<<"hum">>], S1),
+    ?assertMatch({[{2, <<"hum">>}], _R1},
+                 osiris_log:read_chunk_parsed(R1)),
+    ok.
+
 write_batch_with_filter(Config) ->
     Conf0 = ?config(osiris_conf, Config),
     S0 = osiris_log:init(Conf0#{filter_size => 32}),