Add property-based tests for find_index_position

Tests two properties covering offset and timestamp specs across randomly
generated index data with strictly increasing offsets and timestamps.
Query values cover before the first entry, within range, and beyond the
last entry.

- Offset spec: returns the last chunk whose offset is at or before the
  query, or the first chunk if the query is before all chunks
- Timestamp spec: returns the first chunk whose timestamp is at or after
  the query, or the last chunk if the query is after all chunks

Also exports `find_index_position/2` under `-ifdef(TEST)`.

Author: lukebakken

src/rabbitmq_stream_s3_log_reader.erl

@@ -83,7 +83,7 @@ tier.
 ]).
 
 -ifdef(TEST).
--export([find_fragment/3]).
+-export([find_fragment/3, find_index_position/2]).
 -endif.
 
 %%%===================================================================

test/unit_SUITE.erl

@@ -18,7 +18,9 @@ all() ->
         array_partition_point,
         array_binary_search_by,
         array_rfind,
-        array_fold
+        array_fold,
+        find_index_position_offset,
+        find_index_position_timestamp
     ].
 
 init_per_suite(Config) ->
@@ -265,8 +267,8 @@ array_rfind(_Config) ->
     ).
 
 %% rfind returns the rightmost index satisfying the predicate, or undefined.
-%%   - undefined  => no entry satisfies the predicate
-%%   - {ok, Idx}  => entry at Idx satisfies predicate, no entry at Idx+1..N-1 does
+%%   - undefined => no entry satisfies the predicate
+%%   - Idx       => entry at Idx satisfies predicate, no entry at Idx+1..N-1 does
 prop_array_rfind() ->
     ?FORALL(
         {Values, Threshold},
@@ -330,3 +332,88 @@ prop_array_fold() ->
 
 ints_to_array(Ints) ->
     iolist_to_binary([<<V:64/unsigned>> || V <- Ints]).
+
+%% -------------------------------------------------------------------------
+%% find_index_position properties
+%% -------------------------------------------------------------------------
+
+find_index_position_offset(_Config) ->
+    rabbit_ct_proper_helpers:run_proper(
+        fun prop_find_index_position_offset/0, [], 500
+    ).
+
+%% Offset spec: returns the last chunk whose offset =< O.
+%% If O is before all chunks, returns the first chunk.
+prop_find_index_position_offset() ->
+    ?FORALL(
+        {Chunks, QueryOffset},
+        gen_index_data(),
+        begin
+            IndexData = chunks_to_index(Chunks),
+            {ChunkId, _Ts, _Pos} = rabbitmq_stream_s3_log_reader:find_index_position(
+                IndexData, {offset, QueryOffset}
+            ),
+            Offsets = [O || {O, _Ts} <- Chunks],
+            %% Expected: last offset =< QueryOffset, or first offset if all are greater.
+            Expected =
+                case lists:takewhile(fun(O) -> O =< QueryOffset end, Offsets) of
+                    [] -> hd(Offsets);
+                    Matching -> lists:last(Matching)
+                end,
+            ChunkId =:= Expected
+        end
+    ).
+
+find_index_position_timestamp(_Config) ->
+    rabbit_ct_proper_helpers:run_proper(
+        fun prop_find_index_position_timestamp/0, [], 500
+    ).
+
+%% Timestamp spec: returns the first chunk whose timestamp >= Ts.
+%% If Ts is after all chunks, returns the last chunk.
+prop_find_index_position_timestamp() ->
+    ?FORALL(
+        {Chunks, QueryTs},
+        gen_index_data(),
+        begin
+            IndexData = chunks_to_index(Chunks),
+            {ChunkId, _Ts, _Pos} = rabbitmq_stream_s3_log_reader:find_index_position(
+                IndexData, {timestamp, QueryTs}
+            ),
+            %% Expected: first chunk whose timestamp >= QueryTs, or last chunk.
+            Expected =
+                case lists:dropwhile(fun({_O, Ts}) -> Ts < QueryTs end, Chunks) of
+                    [] -> element(1, lists:last(Chunks));
+                    [{O, _Ts} | _] -> O
+                end,
+            ChunkId =:= Expected
+        end
+    ).
+
+%% Generates a non-empty list of {Offset, Timestamp} pairs with strictly
+%% increasing offsets and timestamps, plus a random query value in a range
+%% that covers before the first entry, within the range, and beyond the last.
+gen_index_data() ->
+    ?LET(
+        {N, Steps},
+        {integer(1, 20), non_empty(list(integer(1, 100)))},
+        begin
+            StepsN = lists:sublist(Steps ++ lists:duplicate(N, 1), N),
+            {Chunks, LastOffset, LastTs} = lists:foldl(
+                fun(Step, {Acc, NextOffset, NextTs}) ->
+                    {[{NextOffset, NextTs} | Acc], NextOffset + Step, NextTs + Step}
+                end,
+                {[], 1, 1_000_000},
+                StepsN
+            ),
+            SortedChunks = lists:reverse(Chunks),
+            %% Query range covers before first entry (0), within range, and beyond last.
+            ?LET(Query, integer(0, max(LastOffset, LastTs) + 200), {SortedChunks, Query})
+        end
+    ).
+
+chunks_to_index(Chunks) ->
+    iolist_to_binary([
+        ?INDEX_RECORD(O, Ts, Pos)
+     || {Pos, {O, Ts}} <- lists:zip(lists:seq(0, length(Chunks) - 1), Chunks)
+    ]).