Fix find_fragment timestamp predicate and add property-based tests

.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

README.md

@@ -1,6 +1,15 @@
 ## rabbitmq-stream-s3
 
-This is an ongoing project for integrating RabbitMQ streams with S3 storage.
+RabbitMQ Streams stores data in an append-only log on local disk, managed by the [osiris](https://github.com/rabbitmq/osiris) library. Each stream is a sequence of segments; old segments are deleted by retention policies. Once a segment is deleted locally, any consumer that hasn't read it yet loses access to that data permanently.
+
+`rabbitmq_stream_s3` adds a **remote tier** by hooking into osiris via two plugin behaviours:
+
+- **`osiris_log_manifest`** (write path) — receives events for every chunk written and every segment roll. It tracks fragment metadata in memory and notifies the upload server when a fragment is ready. Uploads happen asynchronously in the background and do not block writes. The server ensures a fragment is fully uploaded before the corresponding local segment is eligible for deletion.
+- **`osiris_log_reader`** (read path) — when a consumer requests an offset or timestamp that no longer exists locally, instead of failing, the broker resolves the manifest to find the right S3 object and byte position, then streams the data directly from S3 to the consumer using HTTP range requests.
+
+Streams can retain data indefinitely in S3 at low cost, while local disk only holds recent data. Consumers see a single continuous stream regardless of whether the data is local or remote.
+
+For a detailed description of the design, see [docs/README.md](docs/README.md).
 
 ## Project Maturity
 
@@ -13,44 +22,42 @@ This project currently requires specific development branches of the `rabbitmq-s
 ### rabbitmq-server
 Branch: [`streams-tiered-storage`](https://github.com/amazon-mq/upstream-to-rabbitmq-server/tree/streams-tiered-storage)
 
-Contains changes needed for S3 integration.
-
 ### osiris
 Branch: [`tiered-storage-abstractions`](https://github.com/amazon-mq/upstream-to-osiris/tree/tiered-storage-abstractions)
 
-Contains the abstraction layer in Osiris. See [Tiered Storage Support for RabbitMQ Streams](https://github.com/rabbitmq/osiris/issues/184)
+See [Tiered Storage Support for RabbitMQ Streams](https://github.com/rabbitmq/osiris/issues/184).
 
 ## Build
 
-1. **Clone the RabbitMQ server repository**
+1. Clone the RabbitMQ server repository:
 ```
 git clone https://github.com/amazon-mq/upstream-to-rabbitmq-server.git
-cd rabbitmq-server
+cd upstream-to-rabbitmq-server
 ```
-2. **Switch to the required branch**
+2. Switch to the required branch:
 ```
 git checkout streams-tiered-storage
 ```
-3. **This build will include the stream-s3 plugin and the correct osiris branch so just build!**
+3. Build:
 ```
 make
 ```
 
-For more information on how to build and develop plugins in RabbitMQ, see [plugin-development](https://www.rabbitmq.com/plugin-development)
+For more information on building and developing RabbitMQ plugins, see [plugin-development](https://www.rabbitmq.com/plugin-development).
 
 ## Configure
 
-### Osiris
-Osiris needs to be configured to use our s3 log_reader and log_manifest:
+See [docs/configuration.md](docs/configuration.md) for all configuration options.
+
+The minimum required configuration is:
 
 ```
 streams.log_reader = rabbitmq_stream_s3_log_reader
 streams.log_manifest = rabbitmq_stream_s3_log_manifest
-```
 
-### AWS Credentials
-
-See [rabbitmq_aws](https://github.com/rabbitmq/rabbitmq-server/blob/su_aws/replace_httpc_with_gun_fmt/deps/rabbitmq_aws/README.md#configuration)
+s3.bucket = my-rabbitmq-streams-bucket
+s3.region = us-east-1
+```
 
 ## Security
 

docs/README.md

@@ -53,15 +53,15 @@ data/
 
 Each stream uses a directory under the `stream/` directory in the format `{vhost}_{stream-name}_{created-timestamp}`. Messages published to streams are stored in segment files. Segment files are named with the offset of the first entry in the segment. Messages are batched together by the writer process into chunks and appended to the latest segment file. Once the segment file exceeds a max size or a max number of chunks, the segment is closed and a new one is opened. Every segment has a corresponding index file with the same offset. The index file contains a small record for each chunk in the segment with metadata like the offset, timestamp, and byte offset of the chunk within the segment file. The default max size of a segment file is 500 MB. The size of the index depends on the number of chunks in the segment file. Publishing at high throughput results in smaller index files since there are fewer, larger chunks.
 
-Segment and index files are identical between cluster members of a stream. All records are send to a writer member which writes the records as chunks. Any number of replica members then replicate the chunks from the writer's log. When a majority of members have written a chunk, publishers receive confirms that their messages have been written and the messages may then be read by consumers. Cluster membership, epoch numbers and writer/replica roles are decided by a Raft system.
+Segment and index files are identical between cluster members of a stream. All records are sent to a writer member which writes the records as chunks. Any number of replica members then replicate the chunks from the writer's log. When a majority of members have written a chunk, publishers receive confirms that their messages have been written and the messages may then be read by consumers. Cluster membership, epoch numbers and writer/replica roles are decided by a Raft system.
 
 ### Remote-tier storage
 
 `rabbitmq-stream-s3` uploads committed stream data to the remote tier aggressively to make local-tier data redundant quickly. In order to upload more aggressively, `rabbitmq-stream-s3` uses a different data representation in the remote tier.
 
 ### S3 bucket layout
 
-`rabbitmq-stream-s3` uses one S3 bucket per cluster. The remote tier uses one [prefix](https://docs.aws.amazon.com/AmazonS3/latest/userguide/using-prefixes.html) per stream similar to the local tier's use of directories. Under this prefix there prefixes `data` to store stream data like segment and index file contents, and `metadata` to store tracking information used for consumers and retention.
+`rabbitmq-stream-s3` uses one S3 bucket per cluster. The remote tier uses one [prefix](https://docs.aws.amazon.com/AmazonS3/latest/userguide/using-prefixes.html) per stream similar to the local tier's use of directories. Under this prefix there are prefixes `data` to store stream data like segment and index file contents, and `metadata` to store tracking information used for consumers and retention.
 
 <!-- NOTE: it would be easy to also support multi-tenant buckets by adding config for a prefix to use within a bucket. We wouldn't use this feature ourselves though. -->
 
@@ -77,7 +77,7 @@ rabbitmq/
 
 ### Fragments
 
-The data representation between the local and remote tiers are separate. The local tier contains segment and index files while the remote tier contains smaller objects called _fragments_ which concatenate smaller sections of segment and index data together. Where a segment file typically reaches 500 MB, fragments store a smaller section of a segment around 64 MB of chunk-aligned segment data and their accompanying records from the index file. NOTE: the 64 MB figure will be tuned in testing. We expect a size in the range of 10 MB - 128 MB to be ideal.
+The data representation between the local and remote tiers are separate. The local tier contains segment and index files while the remote tier contains smaller objects called _fragments_ which concatenate smaller sections of segment and index data together. Where a segment file typically reaches 500 MB, fragments store a smaller section of a segment around 64 MiB of chunk-aligned segment data and their accompanying records from the index file. NOTE: the 64 MiB figure will be tuned in testing. We expect a size in the range of 10 MiB - 128 MiB to be ideal.
 
 ![Fragment layout](./Fragment.svg)
 
@@ -120,7 +120,7 @@ Also see the [manifest documentation](./manifest.md#operations) for details abou
 
 #### Stream deletion
 
-`rabbitmq-stream-s3` leverages [the data stored in Khepri](#concurrency-control) to automatically kick off tasks to delete stream data from the remote tier. The `rabbitmq_stream_s3_db` module covers the plugin's interactions with Khepri. When storing data in Khepri, `rabbitmq_stream_s3_db` sets a Khepri _keep-while condition_ that ties together the lifetime of the entry for the stream ID with the stream queue's metadata. When the stream queue is deleted from the metadata store, the plugin's data for the stream ID is automatically deleted as well. `rabbitmq_stream_s3_db` also sets a Khepri _trigger_ on the stream ID path of the tree which watches for deletions of those tree nodes and executes a _stored procedure_. The stored procedure kicks off a task to perform the deletion of remote tier objects.
+`rabbitmq-stream-s3` leverages [the data stored in Khepri](./manifest.md#concurrency-control) to automatically kick off tasks to delete stream data from the remote tier. The `rabbitmq_stream_s3_db` module covers the plugin's interactions with Khepri. When storing data in Khepri, `rabbitmq_stream_s3_db` sets a Khepri _keep-while condition_ that ties together the lifetime of the entry for the stream ID with the stream queue's metadata. When the stream queue is deleted from the metadata store, the plugin's data for the stream ID is automatically deleted as well. `rabbitmq_stream_s3_db` also sets a Khepri _trigger_ on the stream ID path of the tree which watches for deletions of those tree nodes and executes a _stored procedure_. The stored procedure kicks off a task to perform the deletion of remote tier objects.
 
 For more details about keep-while conditions, triggers and stored procedures, see the [Khepri overview documentation](https://hexdocs.pm/khepri/).
 

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.

docs/configuration.md

@@ -0,0 +1,53 @@
+# Configuration
+
+## Enabling the plugin
+
+Add the following to `rabbitmq.conf` to route stream reads and writes through the plugin:
+
+```
+streams.log_reader = rabbitmq_stream_s3_log_reader
+streams.log_manifest = rabbitmq_stream_s3_log_manifest
+```
+
+## S3 settings
+
+### `s3.bucket`
+
+The S3 bucket to use for remote tier storage. Required.
+
+```
+s3.bucket = my-rabbitmq-streams-bucket
+```
+
+### `s3.region`
+
+The AWS region of the S3 bucket. If not set, the plugin attempts to determine the region automatically via the EC2 instance metadata service.
+
+```
+s3.region = us-east-1
+```
+
+### `s3.region_endpoints.$region`
+
+Overrides the endpoint hostname for a given region. Useful for S3-compatible storage or VPC endpoints.
+
+```
+s3.region_endpoints.us-east-1 = s3.us-east-1.amazonaws.com
+```
+
+## Credentials
+
+The plugin resolves AWS credentials in the following order:
+
+1. Static credentials from `rabbitmq.conf` (`s3.access_key_id` and `s3.secret_key`)
+2. Container credentials endpoint (if the `AWS_CONTAINER_CREDENTIALS_FULL_URI` environment variable is set)
+3. EC2 instance metadata service (IMDSv2)
+
+### `s3.access_key_id` and `s3.secret_key`
+
+Static AWS credentials. Not recommended for production; prefer IAM roles.
+
+```
+s3.access_key_id = AKIAIOSFODNN7EXAMPLE
+s3.secret_key = wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
+```

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_manifest.erl

@@ -226,6 +226,18 @@ recover_fragments(IdxFile, Acc) ->
         IdxArray
     ).
 
+recover_fragments(
+    _Threshold0,
+    _SegmentFile,
+    SegmentOffset,
+    SeqNo0,
+    _NumChunks0,
+    Fragments0,
+    <<>>
+) ->
+    %% The index file exists but contains no records yet (e.g. the stream was
+    %% created but no segment has been flushed). Treat as an empty fragment.
+    {#fragment{segment_offset = SegmentOffset, seq_no = SeqNo0}, Fragments0};
 recover_fragments(
     Threshold0,
     SegmentFile,

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

@@ -452,9 +452,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)}],