Fix find_fragment timestamp predicate and add property-based tests

`find_fragment` used `Ts >= FTs` to select the manifest fragment for a
timestamp subscription. This predicate finds the last fragment whose
first timestamp is at or before `Ts`. When `Ts` falls between two
fragments — after fragment N's `LTs` but before fragment N+1's `FTs` —
the predicate incorrectly returns fragment N instead of N+1, causing the
reader to start from the wrong position.

The correct predicate is `Ts >= LTs`: find the first fragment whose last
timestamp is greater than or equal to `Ts`. This correctly identifies
the fragment whose range contains `Ts`, or the earliest fragment after
`Ts` if `Ts` falls in a gap between fragments.

The index position calculation for the timestamp case uses
`min(Idx0, NumEntries - 1)` rather than `saturating_decr` because the
partition point already identifies the target fragment directly.

The `read_from_remote_tier_by_timestamp` integration test is also fixed:
`Timestamp3` must be captured before publishing offset 2 so that
`find_index_position` returns offset 2 rather than offset 3.

- Fix `init_offset_reader/2` calling `osiris_log` directly for the local
  path; route through `init_local_reader/2` instead
- Fix `range_spec_to_location_number/2`: suffix range used addition
  instead of subtraction; open-ended byte range returned `infinity`
  instead of the remaining file size
- Preserve `?C_OSIRIS_LOG_FIRST_OFFSET` after local segment deletion;
  only update the shared atomic used for local/remote tier routing

- Upload CT logs as an artifact on test failure
- Reduce `?MAX_SEGMENT_SIZE_BYTES` to 1 MiB in test builds to trigger
  segment rolls without large payloads

- Add integration tests for reading stream data from the remote tier by
  offset and timestamp
- Convert `unit_SUITE` to use PropEr; add property-based tests for
  `range_spec_to_location_number/2`
- Add property-based tests for `find_fragment` timestamp behaviour,
  covering timestamps within fragment ranges, in gaps between fragments,
  and after all fragments
- Add property-based tests for `rabbitmq_stream_s3_array`:
  `partition_point`, `binary_search_by`, `rfind`, and `fold`
- Add property-based tests for `find_index_position` offset and
  timestamp specs
- Add diagnostic logging to `read_from_remote_tier_by_timestamp` to
  surface manifest entry timestamps and captured timestamps in CI output
- Add `get_range_by_reference/1` and `get_manifest_by_reference/1` test
  helpers to `rabbitmq_stream_s3_server`
- Support container credentials in the integration test group

Author: lukebakken

.github/workflows/build-test.yaml

@@ -49,5 +49,11 @@ jobs:
         run: make -C ${{ github.workspace }}/rabbitmq-server/deps/rabbitmq_stream_s3
       - name: tests
         run: make -C ${{ github.workspace }}/rabbitmq-server/deps/rabbitmq_stream_s3 tests
+      - name: maybe upload CT logs
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: ct-logs-${{ matrix.rmq-version }}-otp-${{ matrix.otp-version }}
+          path: ${{ github.workspace }}/rabbitmq-server/logs/
       - name: dialyze
         run: make -C ${{ github.workspace }}/rabbitmq-server/deps/rabbitmq_stream_s3 dialyze

Makefile

@@ -3,7 +3,7 @@ PROJECT_DESCRIPTION = RabbitMQ S3 plugin
 PROJECT_MOD = rabbitmq_stream_s3_app
 
 DEPS = rabbit_common rabbit osiris khepri gun
-TEST_DEPS = rabbitmq_ct_helpers rabbitmq_ct_client_helpers
+TEST_DEPS = rabbitmq_ct_helpers rabbitmq_ct_client_helpers rabbitmq_stream proper
 LOCAL_DEPS = xmerl
 
 DEP_EARLY_PLUGINS = rabbit_common/mk/rabbitmq-early-plugin.mk

docs/TODO.md

@@ -0,0 +1,20 @@
+# TODO
+
+## `stream_test_utils` bug: `receive_stream_commands/3` discards accumulated data on retry
+
+`stream_test_utils:receive_stream_commands/3` loops up to 10 times calling
+`gen_tcp:recv/3`, accumulating incoming data into the `rabbit_stream_core`
+state. When the loop is exhausted without a complete frame, it returns the
+atom `empty` — but discards the updated core state containing the partially
+accumulated frame data.
+
+Any caller that retries by calling `receive_stream_commands` again with the
+original core state will re-read from an empty buffer, losing the bytes
+already received. For large chunks (e.g. a 2 MB message), this means the
+frame can never be assembled.
+
+The fix is to either return the updated core state alongside `empty`, or for
+callers to implement their own recv loop that preserves the accumulated state
+across iterations. `s3_streams_SUITE` works around this with a local
+`receive_deliver/3` helper that calls `rabbit_stream_core:next_command/1` and
+`rabbit_stream_core:incoming_data/2` directly.

include/rabbitmq_stream_s3.hrl

@@ -258,8 +258,12 @@
 -define(MAX_FRAGMENT_SIZE_B, 67_108_864).
 %% %% 1 GiB (2^30 B)
 %% -define(MAX_SEGMENT_SIZE_BYTES, 1_073_741_824).
-%% 1/G GiB (2^29 B)
+%% 1/2 GiB (2^29 B), reduced to 1 MiB in tests to trigger segment rolls.
+-ifdef(TEST).
+-define(MAX_SEGMENT_SIZE_BYTES, 1_048_576).
+-else.
 -define(MAX_SEGMENT_SIZE_BYTES, 536_870_912).
+-endif.
 
 -type byte_offset() :: non_neg_integer().
 -type checksum() :: non_neg_integer().

src/rabbitmq_stream_s3_api_fs.erl

@@ -30,6 +30,10 @@ associated file in that folder.
     set_data_dir/1
 ]).
 
+-ifdef(TEST).
+-export([range_spec_to_location_number/2]).
+-endif.
+
 -behaviour(rabbitmq_stream_s3_api).
 
 -type connection() :: rabbitmq_stream_s3_api:connection().
@@ -238,16 +242,16 @@ with_timeout(Timeout, Fun) ->
 range_spec_to_location_number(FileSize, SuffixRange) when
     is_integer(SuffixRange), SuffixRange < 0
 ->
-    Location = FileSize - SuffixRange,
+    Location = FileSize + SuffixRange,
     {Location, -SuffixRange};
 range_spec_to_location_number(FileSize, SuffixRange) when is_integer(SuffixRange) ->
     Location = 0,
     Number = min(SuffixRange, FileSize),
     {Location, Number};
-range_spec_to_location_number(_FileSize, {StartByte, EndByte}) ->
+range_spec_to_location_number(FileSize, {StartByte, EndByte}) ->
     Number =
         case EndByte of
-            undefined -> infinity;
+            undefined -> FileSize - StartByte;
             _ -> EndByte - StartByte + 1
         end,
     {StartByte, Number}.

src/rabbitmq_stream_s3_log_reader.erl

@@ -82,6 +82,10 @@ tier.
     code_change/3
 ]).
 
+-ifdef(TEST).
+-export([find_fragment/3, find_index_position/2]).
+-endif.
+
 %%%===================================================================
 %%% osiris_log_reader callbacks
 %%%===================================================================
@@ -101,7 +105,7 @@ init_offset_reader(OffsetSpec, Config) ->
         {ok, Fragment, Position, ChunkId} ->
             init_remote_reader(Fragment, Position, ChunkId, Config);
         {local, LocalSpec} ->
-            osiris_log:init_offset_reader(LocalSpec, Config);
+            init_local_reader(LocalSpec, Config);
         {error, _} = Err ->
             Err
     end.
@@ -674,14 +678,19 @@ find_fragment(Entries, Spec, GetGroup) ->
             {offset, Offset} ->
                 fun(?ENTRY(O, _FTs, _LTs, _K, _)) -> Offset >= O end;
             {timestamp, Ts} ->
-                fun(?ENTRY(_O, FTs, _LTs, _K, _)) -> Ts >= FTs end
+                fun(?ENTRY(_O, _FTs, LTs, _K, _)) -> Ts >= LTs end
         end,
     Idx0 = rabbitmq_stream_s3_array:partition_point(
         PartitionPredicate,
         ?ENTRY_B,
         Entries
     ),
-    Idx = saturating_decr(Idx0),
+    NumEntries = byte_size(Entries) div ?ENTRY_B,
+    Idx =
+        case Spec of
+            {offset, _} -> saturating_decr(Idx0);
+            {timestamp, _} -> min(Idx0, NumEntries - 1)
+        end,
     case rabbitmq_stream_s3_array:at(Idx, ?ENTRY_B, Entries) of
         ?FRAGMENT(EntryOffset, _FTs, _LTs, _Sq, _Sz, _) ->
             EntryOffset;
@@ -766,39 +775,14 @@ find_fragment_test() ->
     Size = 200,
     FragmentEntries = <<
         ?FRAGMENT(
-            %% Fragments every 20 offsets, 0..=2000
             (N * 20),
-            %% Timestamps between 2000ms ago and `Ts`
             (Ts - 2000 + N * 20),
             (Ts - 2000 + (N + 1) * 20),
             0,
             Size
         )
      || N <- lists:seq(0, 100)
     >>,
-
-    FindFragment1 = fun(Spec) ->
-        find_fragment(
-            FragmentEntries,
-            Spec,
-            %% There are only fragments in this manifest.
-            fun(_, _, _) -> erlang:error(unimplemented) end
-        )
-    end,
-    %% Offsets:
-    ?assertEqual(0, FindFragment1({offset, 0})),
-    ?assertEqual(40, FindFragment1({offset, 50})),
-    ?assertEqual(100, FindFragment1({offset, 100})),
-    ?assertEqual(2_000, FindFragment1({offset, 10_000})),
-    %% Timestamps:
-    ?assertEqual(0, FindFragment1({timestamp, Ts - 2000})),
-    %% When placed between two chunks timestamp search prefers the later chunk.
-    %% But when searching in fragments we want to prefer the earlier fragment
-    %% since it most likely contains the target timestamp.
-    ?assertEqual(40, FindFragment1({timestamp, Ts - 2000 + 50})),
-    ?assertEqual(100, FindFragment1({timestamp, Ts - 2000 + 100})),
-    ?assertEqual(2_000, FindFragment1({timestamp, Ts - 2000 + 10_000})),
-
     %% Factor out those fragments into a group.
     NextFragmentEntries = <<
         ?FRAGMENT((N * 20), (Ts - 2000 + N * 20), (Ts - 2000 + (N + 1) * 20), 0, Size)
@@ -829,7 +813,8 @@ find_fragment_test() ->
     ?assertEqual(0, FindFragment2({offset, 0})),
     ?assertEqual(40, FindFragment2({offset, 50})),
     ?assertEqual(40, FindFragment2({timestamp, Ts - 2000 + 50})),
-
+    %% Offset and timestamp assertions on flat fragment lists are omitted here;
+    %% they are covered by the property-based tests in unit_SUITE.
     ok.
 
 find_index_position_test() ->

src/rabbitmq_stream_s3_server.erl

@@ -64,6 +64,10 @@
     delete_stream/1
 ]).
 
+-ifdef(TEST).
+-export([get_range_by_reference/1, get_manifest_by_reference/1]).
+-endif.
+
 %% Useful for other modules
 -export([
     get_fragment_trailer/1,
@@ -224,6 +228,29 @@ handle_call(#get_manifest{stream = StreamId}, From, State) ->
     %% readers to find anything within branches.
     Event = #manifest_requested{stream = StreamId, requester = From},
     {noreply, evolve_event(Event, State)};
+handle_call(
+    {get_range_by_reference, Reference},
+    _From,
+    #?MODULE{references = References} = State
+) ->
+    Range =
+        case References of
+            #{Reference := StreamId} -> get_range(StreamId);
+            _ -> empty
+        end,
+    {reply, Range, State};
+handle_call(
+    {get_manifest_by_reference, Reference},
+    From,
+    #?MODULE{references = References} = State
+) ->
+    case References of
+        #{Reference := StreamId} ->
+            Event = #manifest_requested{stream = StreamId, requester = From},
+            {noreply, evolve_event(Event, State)};
+        _ ->
+            {reply, undefined, State}
+    end;
 handle_call(Request, From, State) ->
     ?LOG_INFO(?MODULE_STRING " received unexpected call from ~p: ~W", [From, Request, 10]),
     {noreply, State}.
@@ -452,9 +479,20 @@ apply_effect(
             is_integer(FstTs)
         ->
             counters:add(counter(), ?C_LOCAL_TIER_RETENTION_EVALUATIONS, 1),
+            %% Update the shared atomic so that the log reader routes offsets
+            %% below FstOff to the remote tier. Do NOT update
+            %% ?C_OSIRIS_LOG_FIRST_OFFSET here: that counter reflects the
+            %% first offset of the entire stream (local + remote) and is set
+            %% by #set_range{} to the remote tier's first offset. Overwriting
+            %% it with the local tier's first offset would cause stream metrics
+            %% to hide the existence of data in the remote tier.
             osiris_log_shared:set_first_chunk_id(Shared, FstOff),
             counters:put(Cnt, ?C_OSIRIS_LOG_SEGMENTS, NumSegLeft);
-        (_) ->
+        (Result) ->
+            ?LOG_DEBUG(
+                "trigger_retention EvalFun: unexpected result ~p for stream '~ts', skipping set_first_chunk_id",
+                [Result, StreamId]
+            ),
             ok
     end,
     Spec = [{'fun', local_retention_fun(StreamId)}],
@@ -1117,6 +1155,14 @@ counter() ->
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
 
+-spec get_range_by_reference(stream_reference()) -> rabbitmq_stream_s3:range().
+get_range_by_reference(Reference) ->
+    gen_server:call(?SERVER, {get_range_by_reference, Reference}, infinity).
+
+-spec get_manifest_by_reference(stream_reference()) -> #manifest{} | undefined.
+get_manifest_by_reference(Reference) ->
+    gen_server:call(?SERVER, {get_manifest_by_reference, Reference}, infinity).
+
 eval_local_retention_test() ->
     IdxFiles = [
         <<"/data/00000000000000000000.index">>,

test/rabbitmq_stream_s3_api_aws_SUITE.erl

@@ -78,6 +78,7 @@ end_per_suite(Config) ->
 
 init_per_group(integration, Config) ->
     Cfg = {
+        os:getenv("AWS_CONTAINER_CREDENTIALS_FULL_URI"),
         os:getenv("AWS_ACCESS_KEY_ID"),
         os:getenv("AWS_SECRET_ACCESS_KEY"),
         os:getenv("AWS_SESSION_TOKEN"),
@@ -88,18 +89,24 @@ init_per_group(integration, Config) ->
         {skip,
             "AWS access credentials are not set. Skipping this group is OK! See the moduledoc for more information."},
     case Cfg of
-        {false, _, _, _, _} ->
+        {AwsCredentialsUri, _, _, _, _, Bucket0} when
+            AwsCredentialsUri =/= false andalso Bucket0 =/= false
+        ->
+            {ok, _} = application:ensure_all_started(gun),
+            ok = application:set_env(rabbitmq_stream_s3, bucket, list_to_binary(Bucket0)),
+            Config;
+        {_, false, _, _, _, _} ->
             Skip;
-        {_, false, _, _, _} ->
+        {_, _, false, _, _, _} ->
             Skip;
-        {_, _, false, _, _} ->
+        {_, _, _, false, _, _} ->
             Skip;
-        {_, _, _, false, _} ->
+        {_, _, _, _, false, _} ->
             Skip;
-        {_, _, _, _, false} ->
+        {_, _, _, _, _, false} ->
             Skip;
-        {AccessKey, SecretKey, SecurityToken, Region, Bucket} ->
-            application:ensure_all_started(gun),
+        {_, AccessKey, SecretKey, SecurityToken, Region, Bucket1} ->
+            {ok, _} = application:ensure_all_started(gun),
             ok = application:set_env(rabbitmq_stream_s3, aws_access_key, list_to_binary(AccessKey)),
             ok = application:set_env(rabbitmq_stream_s3, aws_secret_key, list_to_binary(SecretKey)),
             ok = application:set_env(
@@ -108,7 +115,7 @@ init_per_group(integration, Config) ->
                 list_to_binary(SecurityToken)
             ),
             ok = application:set_env(rabbitmq_stream_s3, aws_region, list_to_binary(Region)),
-            ok = application:set_env(rabbitmq_stream_s3, bucket, list_to_binary(Bucket)),
+            ok = application:set_env(rabbitmq_stream_s3, bucket, list_to_binary(Bucket1)),
             Config
     end.