Merge pull request #15 from amazon-mq/md/dialyze

Author: the-mikedavis

.github/workflows/build-test.yaml

@@ -41,5 +41,13 @@ jobs:
       - uses: actions/checkout@v5
         with:
           path: rabbitmq-server/deps/rabbitmq_stream_s3
-      - run: make -C ${{ github.workspace }}/rabbitmq-server/deps/rabbitmq_stream_s3
-      - run: make -C ${{ github.workspace }}/rabbitmq-server/deps/rabbitmq_stream_s3 tests
+      - uses: actions/cache@v4
+        with:
+          path: ${{ github.workspace }}/rabbitmq-server/deps/rabbitmq_stream_s3/.rabbitmq_stream_s3.plt
+          key: plt-${{ matrix.rmq-version }}-otp-${{ matrix.otp-version }}-elixir-${{ matrix.elixir-version }}
+      - name: make
+        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: dialyze
+        run: make -C ${{ github.workspace }}/rabbitmq-server/deps/rabbitmq_stream_s3 dialyze

Makefile

@@ -2,11 +2,13 @@ PROJECT = rabbitmq_stream_s3
 PROJECT_DESCRIPTION = RabbitMQ S3 plugin
 PROJECT_MOD = rabbitmq_stream_s3_app
 
-DEPS = rabbit_common rabbit osiris
+DEPS = rabbit_common rabbit osiris rabbitmq_aws
 TEST_DEPS = rabbitmq_ct_helpers rabbitmq_ct_client_helpers
 
 DEP_EARLY_PLUGINS = rabbit_common/mk/rabbitmq-early-plugin.mk
 DEP_PLUGINS = rabbit_common/mk/rabbitmq-plugin.mk
 
+PLT_APPS += ssl
+
 include ../../rabbitmq-components.mk
 include ../../erlang.mk

src/rabbitmq_stream_s3_api.erl

@@ -23,40 +23,41 @@
 -include_lib("rabbit_common/include/rabbit.hrl").
 -include_lib("kernel/include/logger.hrl").
 
--define(DEFAULT_OPTS, []).
-
--spec put_object(term(), binary(), binary(), binary()) ->
+-spec put_object(rabbitmq_aws:connection_handle(), binary(), binary(), iodata()) ->
     ok | {error, term()}.
 put_object(Handle, Bucket, Key, Object) ->
     put_object(Handle, Bucket, Key, Object, []).
 
--spec put_object(term(), binary(), binary(), binary(), list()) ->
+-spec put_object(rabbitmq_aws:connection_handle(), binary(), binary(), iodata(), list()) ->
     ok | {error, term()}.
-put_object(Handle, Bucket, Key, Object, Opts) ->
+put_object(Handle, Bucket, Key, Object, Opts) when is_list(Opts) ->
     Path = object_path(Bucket, Key),
-    Headers =
+    {Headers, Opts1} =
         case proplists:get_value(crc32, Opts, []) of
             [] ->
-                [];
+                {[], Opts};
             Checksum when is_integer(Checksum) ->
                 C = base64:encode_to_string(<<Checksum:32/unsigned>>),
-                [{"x-amz-checksum-crc32", C}]
+                O = proplists:delete(crc32, Opts),
+                {[{"x-amz-checksum-crc32", C}], O}
         end,
-    case rabbitmq_aws:put(Handle, Path, Object, Headers, [?DEFAULT_OPTS | Opts]) of
+    case rabbitmq_aws:put(Handle, Path, Object, Headers, Opts1) of
         {ok, {_Headers, <<>>}} ->
             ok;
         Error ->
             {error, Error}
     end.
 
--spec get_object(term(), binary(), binary()) ->
+-spec get_object(rabbitmq_aws:connection_handle(), binary(), binary()) ->
     {ok, binary()} | {error, term()}.
 get_object(Handle, Bucket, Key) ->
     get_object(Handle, Bucket, Key, []).
 
+-spec get_object(rabbitmq_aws:connection_handle(), binary(), binary(), list()) ->
+    {ok, binary()} | {error, term()}.
 get_object(Handle, Bucket, Key, Opts) ->
     Path = object_path(Bucket, Key),
-    case rabbitmq_aws:get(Handle, Path, [], [?DEFAULT_OPTS | Opts]) of
+    case rabbitmq_aws:get(Handle, Path, [], Opts) of
         {ok, {_Headers, Body}} ->
             {ok, Body};
         {error, "Not Found", _} ->
@@ -65,27 +66,33 @@ get_object(Handle, Bucket, Key, Opts) ->
             {error, Error}
     end.
 
--spec get_object_attributes(term(), binary(), binary()) ->
-    {ok, proplists:proplist()} | {error, term()}.
+-spec get_object_attributes(rabbitmq_aws:connection_handle(), binary(), binary()) ->
+    {ok, [{binary(), binary()}]} | {error, term()}.
 get_object_attributes(Handle, Bucket, Key) ->
     get_object_attributes(Handle, Bucket, Key, []).
 get_object_attributes(Handle, Bucket, Key, Opts) ->
     get_object_attributes(Handle, Bucket, Key, [], Opts).
 
--spec get_object_attributes(term(), binary(), binary(), [string()], [term()]) ->
-    {ok, proplists:proplist()} | {error, term()}.
+-spec get_object_attributes(
+    rabbitmq_aws:connection_handle(),
+    binary(),
+    binary(),
+    [binary()],
+    [term()]
+) ->
+    {ok, [{binary(), binary()}]} | {error, term()}.
 get_object_attributes(Handle, Bucket, Key, Attributes, Opts) ->
     Path = object_path(Bucket, Key),
-    case rabbitmq_aws:request(Handle, head, Path, <<"">>, [], [?DEFAULT_OPTS | Opts]) of
+    case rabbitmq_aws:request(Handle, head, Path, <<"">>, [], Opts) of
         {ok, {Headers, _Body}} ->
-            {ok, parse_head_response_headers(Headers, Attributes)};
+            {ok, filter_attributes(Headers, Attributes)};
         {error, "Not Found", _} ->
             {error, not_found};
         Error ->
             {error, Error}
     end.
 
--spec get_object_with_range(term(), binary(), binary(), range_spec()) ->
+-spec get_object_with_range(rabbitmq_aws:connection_handle(), binary(), binary(), range_spec()) ->
     {ok, binary()} | {error, term()}.
 get_object_with_range(Handle, Bucket, Key, RangeSpec) ->
     get_object_with_range(Handle, Bucket, Key, RangeSpec, []).
@@ -94,7 +101,7 @@ get_object_with_range(Handle, Bucket, Key, RangeSpec0, Opts) ->
     Path = object_path(Bucket, Key),
     RangeValue = range_specifier(RangeSpec0),
     Headers = [{"Range", lists:flatten(["bytes=" | RangeValue])}],
-    case rabbitmq_aws:get(Handle, Path, Headers, [?DEFAULT_OPTS | Opts]) of
+    case rabbitmq_aws:get(Handle, Path, Headers, Opts) of
         {ok, {_Headers, Body}} ->
             {ok, Body};
         {error, "Not Found", _} ->
@@ -112,14 +119,21 @@ range_specifier(SuffixLen) when is_integer(SuffixLen) andalso SuffixLen < 0 ->
     %% ~b will format the '-' for us.
     io_lib:format("~b", [SuffixLen]).
 
--spec get_object_size(term(), binary(), binary()) ->
-    integer() | {error, term()}.
+-spec get_object_size(rabbitmq_aws:connection_handle(), binary(), binary()) ->
+    {ok, integer()} | {error, term()}.
 get_object_size(Handle, Bucket, Key) ->
     get_object_size(Handle, Bucket, Key, []).
 
+-spec get_object_size(rabbitmq_aws:connection_handle(), binary(), binary(), [term()]) ->
+    {ok, integer()} | {error, term()}.
 get_object_size(Handle, Bucket, Key, Opts) ->
-    {ok, [{_, Size}]} = get_object_attributes(Handle, Bucket, Key, [<<"content-length">>], Opts),
-    binary_to_integer(Size).
+    case get_object_attributes(Handle, Bucket, Key, [<<"content-length">>], Opts) of
+        {ok, Attributes} ->
+            [{<<"content-length">>, Size}] = Attributes,
+            {ok, binary_to_integer(Size)};
+        Error ->
+            Error
+    end.
 
 object_path(Bucket, Key) ->
     BucketStr = ensure_string(Bucket),
@@ -131,9 +145,8 @@ ensure_string(Binary) when is_binary(Binary) ->
 ensure_string(List) when is_list(List) ->
     List.
 
-parse_head_response_headers(Headers, Attributes) ->
-    filter_attributes(Headers, Attributes).
-
+-spec filter_attributes([{binary(), binary()}], [binary()]) ->
+    [{binary(), binary()}].
 filter_attributes(Headers, []) ->
     Headers;
 filter_attributes(Headers, Attributes) ->

src/rabbitmq_stream_s3_log_reader.erl

@@ -36,10 +36,10 @@
 }).
 
 -record(state, {
-    connection :: pid(),
+    connection :: rabbitmq_aws:connection_handle(),
     buffer = <<>> :: binary(),
-    offset_start :: byte_offset(),
-    offset_end :: byte_offset(),
+    offset_start :: byte_offset() | undefined,
+    offset_end :: byte_offset() | undefined,
     read_size :: pos_integer(),
     bucket :: binary(),
     object :: binary(),
@@ -242,11 +242,11 @@ chunk_iterator(#?MODULE{mode = Local0} = State0, Credit, PrevIter) ->
 iterator_next(#remote_iterator{next_offset = NextOffset0, data = Data0} = Iter0) ->
     case Data0 of
         ?REC_MATCH_SIMPLE(Len, Rem0) ->
-            <<Record:Len/binary, Rem>> = Rem0,
+            <<Record:Len/binary, Rem/binary>> = Rem0,
             Iter = Iter0#remote_iterator{next_offset = NextOffset0 + 1, data = Rem},
             {{NextOffset0, Record}, Iter};
         ?REC_MATCH_SUBBATCH(CompType, NumRecs, UncompressedLen, Len, Rem0) ->
-            <<BatchData:Len/binary, Rem>> = Rem0,
+            <<BatchData:Len/binary, Rem/binary>> = Rem0,
             Record = {batch, NumRecs, CompType, UncompressedLen, BatchData},
             Iter = Iter0#remote_iterator{next_offset = NextOffset0 + NumRecs, data = Rem},
             {{NextOffset0, Record}, Iter};
@@ -266,7 +266,7 @@ start_link(Reader, Bucket, Key) ->
 init({Reader, Bucket, Key}) ->
     erlang:monitor(process, Reader),
     {ok, Connection} = rabbitmq_aws:open_connection("s3", []),
-    Size = rabbitmq_stream_s3_api:get_object_size(
+    {ok, Size} = rabbitmq_stream_s3_api:get_object_size(
         Connection,
         Bucket,
         Key,
@@ -314,7 +314,7 @@ terminate(_Reason, #state{connection = Connection}) ->
 
 format_status(#{state := #state{buffer = Buffer} = State0} = Status0) ->
     %% Avoid formatting the buffer - it can be large.
-    Size = lists:flatten(io_lib:format("~b bytes", [byte_size(Buffer)])),
+    Size = iolist_to_binary(io_lib:format("~b bytes", [byte_size(Buffer)])),
     Status0#{state := State0#state{buffer = Size}}.
 
 code_change(_, _, State) ->
@@ -472,9 +472,7 @@ read_header2(
                         next_offset = NextChId0 + NumRecords,
                         position = NextPosition
                     },
-                    read_header(Remote);
-                {retry_with, Filter} ->
-                    read_header(Remote0#remote{filter = Filter})
+                    read_header(Remote)
             end;
         false ->
             %% skip and recurse
@@ -560,9 +558,15 @@ convert_remote_to_local(#?MODULE{
     init_local_reader(first, Config).
 
 %% TODO: make this generic for timestamps too.
--spec find_fragment_for_offset(osiris:offset(), #manifest{}, file:filename_all()) ->
+-spec find_fragment_for_offset(
+    osiris:offset(),
+    #manifest{} | rabbitmq_stream_s3_binary_array:array(),
+    file:filename_all()
+) ->
     {ok, ChunkId :: osiris:offset(), byte_offset(), Fragment :: osiris:offset()}.
 find_fragment_for_offset(Offset, #manifest{entries = Entries}, Dir) ->
+    find_fragment_for_offset(Offset, Entries, Dir);
+find_fragment_for_offset(Offset, Entries, Dir) ->
     RootIdx0 =
         rabbitmq_stream_s3_binary_array:partition_point(
             fun(?ENTRY(O, _T, _K, _S, _N, _)) -> Offset > O end,
@@ -619,7 +623,7 @@ find_fragment_for_offset(Offset, #manifest{entries = Entries}, Dir) ->
                 _:70,
                 GroupEntries/binary
             >> = get_group(Dir, Kind, EntryOffset),
-            find_fragment_for_offset(Dir, Offset, GroupEntries)
+            find_fragment_for_offset(Offset, GroupEntries, Dir)
     end.
 
 saturating_decr(0) -> 0;