Add resolve_stream_offset stream protocol command

acogoluegnes · acogoluegnes · commit 765adb49e9a8 · 2026-02-06T15:26:08.000+01:00
This command resolves an offset specification (e.g. 'next' or a
timestamp) to a numeric offset.
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_reader.erl b/deps/rabbitmq_stream/src/rabbit_stream_reader.erl
@@ -2123,6 +2123,49 @@ handle_frame_post_auth(Transport,
                                   {query_offset, ResponseCode, Offset}}),
     send(Transport, S, Frame),
     {Connection1, State};
+handle_frame_post_auth(Transport,
+                       #stream_connection{socket = S,
+                                          virtual_host = VirtualHost,
+                                          user = User} =
+                           Connection0,
+                       State,
+                       {request, CorrelationId,
+                        {resolve_offset_spec, Stream, OffsetSpec, Properties}}) ->
+    {ResponseCode, Offset, Connection1} =
+        case rabbit_stream_utils:check_read_permitted(#resource{name = Stream,
+                                                                kind = queue,
+                                                                virtual_host =
+                                                                    VirtualHost},
+                                                      User, #{})
+        of
+            ok ->
+                case rabbit_stream_manager:lookup_member(VirtualHost, Stream) of
+                    {error, not_found} ->
+                        increase_protocol_counter(?STREAM_DOES_NOT_EXIST),
+                        {?RESPONSE_CODE_STREAM_DOES_NOT_EXIST, 0, Connection0};
+                    {error, not_available} ->
+                        increase_protocol_counter(?STREAM_NOT_AVAILABLE),
+                        {?RESPONSE_CODE_STREAM_NOT_AVAILABLE, 0, Connection0};
+                    {ok, MemberPid} ->
+                        Opts0 = #{chunk_selector => get_chunk_selector(Properties)},
+                        Opts1 = maps:merge(Opts0,
+                                           rabbit_stream_utils:filter_spec(Properties)),
+                        case resolve_offset_spec(MemberPid, OffsetSpec, Opts1) of
+                            {ok, ResolvedOffset} ->
+                                {?RESPONSE_CODE_OK, ResolvedOffset, Connection0};
+                            {error, _} ->
+                                {?RESPONSE_CODE_NO_OFFSET, 0, Connection0}
+                        end
+                end;
+            error ->
+                increase_protocol_counter(?ACCESS_REFUSED),
+                {?RESPONSE_CODE_ACCESS_REFUSED, 0, Connection0}
+        end,
+    Frame =
+        rabbit_stream_core:frame({response, CorrelationId,
+                                  {resolve_offset_spec, ResponseCode, Offset}}),
+    send(Transport, S, Frame),
+    {Connection1, State};
 handle_frame_post_auth(Transport,
                        #stream_connection{stream_subscriptions =
                                               StreamSubscriptions} =
@@ -2821,7 +2864,7 @@ init_reader(ConnectionTransport,
     {ok, Segment} = osiris:init_reader(LocalMemberPid, OffsetSpec,
                                        CounterSpec, Options1),
     ?LOG_DEBUG("Next offset for subscription ~tp is ~tp",
-                     [SubscriptionId, osiris_log:next_offset(Segment)]),
+               [SubscriptionId, osiris_log:next_offset(Segment)]),
     Segment.
 
 single_active_consumer(#consumer{configuration =
@@ -3995,6 +4038,19 @@ i(connected_at, #stream_connection{connected_at = T}, _) ->
 i(_Unknown, _, _) ->
     ?UNKNOWN_FIELD.
 
+resolve_offset_spec(MemberPid, OffsetSpec, Options) ->
+    case node(MemberPid) of
+        Node when Node =:= node() ->
+            osiris:resolve_offset_spec(MemberPid, OffsetSpec, Options);
+        Node ->
+            try erpc:call(Node, osiris, resolve_offset_spec,
+                          [MemberPid, OffsetSpec, Options])
+            catch
+                error:{erpc, _} ->
+                    {error, erpc_error}
+            end
+    end.
+
 -spec send(module(), rabbit_net:socket(), iodata()) -> ok.
 send(Transport, Socket, Data) when is_atom(Transport) ->
     Transport:send(Socket, Data).
diff --git a/deps/rabbitmq_stream/src/rabbit_stream_utils.erl b/deps/rabbitmq_stream/src/rabbit_stream_utils.erl
@@ -316,7 +316,8 @@ command_versions() ->
      {partitions, ?VERSION_1, ?VERSION_1},
      {stream_stats, ?VERSION_1, ?VERSION_1},
      {create_super_stream, ?VERSION_1, ?VERSION_1},
-     {delete_super_stream, ?VERSION_1, ?VERSION_1}].
+     {delete_super_stream, ?VERSION_1, ?VERSION_1},
+     {resolve_offset_spec, ?VERSION_1, ?VERSION_1}].
 
 q(VirtualHost, Name) ->
     rabbit_misc:r(VirtualHost, queue, Name).
diff --git a/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl b/deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl
@@ -76,12 +76,14 @@ groups() ->
        test_stream_test_utils,
        sac_subscription_with_partition_index_conflict_should_return_error,
        test_metadata_with_advertised_hints,
-       test_connection_properties_with_advertised_hints
+       test_connection_properties_with_advertised_hints,
+       test_resolve_offset_spec
       ]},
      %% Run `test_global_counters` on its own so the global metrics are
      %% initialised to 0 for each testcase
      {single_node_1, [], [test_global_counters]},
-     {cluster, [], [test_stream, test_stream_tls, test_metadata, java]}].
+     {cluster, [], [test_stream, test_stream_tls, test_metadata, java,
+                    test_resolve_offset_spec]}].
 
 init_per_suite(Config) ->
     case rabbit_ct_helpers:is_mixed_versions() of
@@ -1236,6 +1238,56 @@ test_connection_properties_with_advertised_hints(Config) ->
     
     ok.
 
+test_resolve_offset_spec(Config) ->
+    Stream = atom_to_binary(?FUNCTION_NAME, utf8),
+    Transport = gen_tcp,
+    Port = get_stream_port(Config),
+    Opts = [{active, false}, {mode, binary}],
+    {ok, S} = Transport:connect("localhost", Port, Opts),
+    C0 = rabbit_stream_core:init(0),
+    C1 = test_peer_properties(Transport, S, C0),
+    C2 = test_authenticate(Transport, S, C1),
+    C3 = test_create_stream(Transport, S, Stream, C2),
+
+    %% Test resolve_offset_spec on empty stream
+    C4 = test_resolve_offset_spec(Transport, S, Stream, first, #{},
+                                  ?RESPONSE_CODE_OK, 0, C3),
+    C5 = test_resolve_offset_spec(Transport, S, Stream, last, #{},
+                                  ?RESPONSE_CODE_OK, 0, C4),
+    C6 = test_resolve_offset_spec(Transport, S, Stream, next, #{},
+                                  ?RESPONSE_CODE_OK, 0, C5),
+
+    %% Publish some messages
+    PublisherId = 1,
+    C7 = test_declare_publisher(Transport, S, PublisherId, Stream, C6),
+    Body = <<"hello">>,
+    C8 = test_publish_confirm(Transport, S, PublisherId, 1, Body, C7),
+    C9 = test_publish_confirm(Transport, S, PublisherId, 2, Body, C8),
+
+    %% Test resolve_offset_spec after publishing
+    C10 = test_resolve_offset_spec(Transport, S, Stream, first, #{},
+                                   ?RESPONSE_CODE_OK, 0, C9),
+    C11 = test_resolve_offset_spec(Transport, S, Stream, last, #{},
+                                   ?RESPONSE_CODE_OK, C10),
+    C12 = test_resolve_offset_spec(Transport, S, Stream, next, #{},
+                                   ?RESPONSE_CODE_OK, 2, C11),
+    C13 = test_resolve_offset_spec(Transport, S, Stream, 0, #{},
+                                   ?RESPONSE_CODE_OK, 0, C12),
+
+    %% Test with timestamp (far future should return next offset)
+    FutureTimestamp = os:system_time(millisecond) + 3600000,
+    C14 = test_resolve_offset_spec(Transport, S, Stream, {timestamp, FutureTimestamp}, #{},
+                                   ?RESPONSE_CODE_OK, 2, C13),
+
+    %% Test on non-existent stream
+    C15 = test_resolve_offset_spec(Transport, S, <<"non_existent_stream">>, first, #{},
+                                   ?RESPONSE_CODE_STREAM_DOES_NOT_EXIST, 0, C14),
+
+    C16 = test_delete_stream(Transport, S, Stream, C15),
+    _C17 = test_close(Transport, S, C16),
+    closed = wait_for_socket_close(Transport, S, 10),
+    ok.
+
 filtered_events(Config, EventType) ->
     Events = rabbit_ct_broker_helpers:rpc(Config, 0,
                                           gen_event,
@@ -1729,6 +1781,24 @@ test_stream_stats(Transport, S, Stream, C0) ->
                  Cmd),
     C.
 
+test_resolve_offset_spec(Transport, S, Stream, OffsetSpec, Properties,
+                         ExpectedResponseCode, C0) ->
+    Frame = request({resolve_offset_spec, Stream, OffsetSpec, Properties}),
+    ok = Transport:send(S, Frame),
+    {Cmd, C} = receive_commands(Transport, S, C0),
+    ?assertMatch({response, 1, {resolve_offset_spec, ExpectedResponseCode, _}},
+                 Cmd),
+    C.
+
+test_resolve_offset_spec(Transport, S, Stream, OffsetSpec, Properties,
+                         ExpectedResponseCode, ExpectedOffset, C0) ->
+    Frame = request({resolve_offset_spec, Stream, OffsetSpec, Properties}),
+    ok = Transport:send(S, Frame),
+    {Cmd, C} = receive_commands(Transport, S, C0),
+    ?assertMatch({response, 1, {resolve_offset_spec, ExpectedResponseCode, ExpectedOffset}},
+                 Cmd),
+    C.
+
 test_close(Transport, S, C0) ->
     CloseReason = <<"OK">>,
     CloseFrame = request({close, ?RESPONSE_CODE_OK, CloseReason}),
diff --git a/deps/rabbitmq_stream_common/include/rabbit_stream.hrl b/deps/rabbitmq_stream_common/include/rabbit_stream.hrl
@@ -44,6 +44,7 @@
 -define(COMMAND_STREAM_STATS, 28).
 -define(COMMAND_CREATE_SUPER_STREAM, 29).
 -define(COMMAND_DELETE_SUPER_STREAM, 30).
+-define(COMMAND_RESOLVE_OFFSET_SPEC, 31).
 
 -define(REQUEST, 0).
 -define(RESPONSE, 1).
diff --git a/deps/rabbitmq_stream_common/src/rabbit_stream_core.erl b/deps/rabbitmq_stream_common/src/rabbit_stream_core.erl
@@ -134,7 +134,8 @@
         MaxVersion :: command_version()}]} |
      {stream_stats, Stream :: binary()} |
      {create_super_stream, stream_name(), Partitions :: [binary()], BindingKeys :: [binary()], Args :: #{binary() => binary()}} |
-     {delete_super_stream, stream_name()}} |
+     {delete_super_stream, stream_name()} |
+     {resolve_offset_spec, stream_name(), offset_spec(), Properties :: #{binary() => binary()}}} |
     {response, correlation_id(),
      {declare_publisher |
       delete_publisher |
@@ -167,7 +168,8 @@
      {exchange_command_versions, response_code(),
       [{Command :: atom(), MinVersion :: command_version(),
         MaxVersion :: command_version()}]} |
-     {stream_stats, response_code(), Stats :: #{binary() => integer()}}} |
+     {stream_stats, response_code(), Stats :: #{binary() => integer()}} |
+     {resolve_offset_spec, response_code(), osiris:offset()}} |
     {unknown, binary()}.
 
 -spec init(term()) -> state().
@@ -423,6 +425,8 @@ response_body({query_publisher_sequence = Tag, Code, Sequence}) ->
     {command_id(Tag), <<Code:16, Sequence:64>>};
 response_body({query_offset = Tag, Code, Offset}) ->
     {command_id(Tag), <<Code:16, Offset:64>>};
+response_body({resolve_offset_spec = Tag, Code, Offset}) ->
+    {command_id(Tag), <<Code:16, Offset:64>>};
 response_body({metadata = Tag, Endpoints, Metadata}) ->
     NumEps = length(Endpoints),
     {_, EndpointsBin} =
@@ -633,7 +637,30 @@ request_body({create_super_stream = Tag, SuperStream, Partitions, BindingKeys, A
            <<(length(BindingKeys)):32>>, BindingKeysBin,
            <<(map_size(Args)):32>>, ArgsBin]};
 request_body({delete_super_stream = Tag, SuperStream}) ->
-    {Tag, <<?STRING(SuperStream)>>}.
+    {Tag, <<?STRING(SuperStream)>>};
+request_body({resolve_offset_spec = Tag, Stream, OffsetSpec, Properties}) ->
+    OffsetSpecBin =
+        case OffsetSpec of
+            first ->
+                <<?OFFSET_TYPE_FIRST:16>>;
+            last ->
+                <<?OFFSET_TYPE_LAST:16>>;
+            next ->
+                <<?OFFSET_TYPE_NEXT:16>>;
+            Offset when is_integer(Offset) ->
+                <<?OFFSET_TYPE_OFFSET:16, Offset:64/unsigned>>;
+            {timestamp, Timestamp} ->
+                <<?OFFSET_TYPE_TIMESTAMP:16, Timestamp:64/signed>>
+        end,
+    PropertiesBin =
+        case map_size(Properties) of
+            0 ->
+                <<>>;
+            _ ->
+                PropsBin = generate_map(Properties),
+                [<<(map_size(Properties)):32>>, PropsBin]
+        end,
+    {Tag, [<<?STRING(Stream), OffsetSpecBin/binary>> | PropertiesBin]}.
 
 append_data(Prev, Data) when is_binary(Prev) ->
     [Prev, Data];
@@ -772,24 +799,13 @@ parse_request(<<?REQUEST:1,
                     OffsetCreditProperties,
                 {{timestamp, Timestamp}, Crdt, PropertiesBin}
         end,
-    Properties =
-        case PropsBin of
-            <<>> ->
-                #{};
-            <<_Count:32, Bin/binary>> ->
-                parse_map(Bin, #{});
-            _ ->
-                ?LOG_WARNING("Incorrect binary for subscription properties: ~w",
-                               [PropsBin]),
-                #{}
-        end,
     request(CorrelationId,
             {subscribe,
              SubscriptionId,
              Stream,
              OffsetSpec,
              Credit,
-             Properties});
+             parse_properties(PropsBin)});
 parse_request(<<?REQUEST:1,
                 ?COMMAND_QUERY_OFFSET:15,
                 ?VERSION_1:16,
@@ -930,6 +946,29 @@ parse_request(<<?REQUEST:1,
                 CorrelationId:32,
                 ?STRING(SuperStreamSize, SuperStream)>>) ->
     request(CorrelationId, {delete_super_stream, SuperStream});
+parse_request(<<?REQUEST:1,
+                ?COMMAND_RESOLVE_OFFSET_SPEC:15,
+                ?VERSION_1:16,
+                CorrelationId:32,
+                ?STRING(StreamSize, Stream),
+                OffsetType:16/signed,
+                OffsetValuePropsBin/binary>>) ->
+    {OffsetSpec, PropsBin} =
+        case OffsetType of
+            ?OFFSET_TYPE_FIRST ->
+                {first, OffsetValuePropsBin};
+            ?OFFSET_TYPE_LAST ->
+                {last, OffsetValuePropsBin};
+            ?OFFSET_TYPE_NEXT ->
+                {next, OffsetValuePropsBin};
+            ?OFFSET_TYPE_OFFSET ->
+                <<Offset:64/unsigned, Rest/binary>> = OffsetValuePropsBin,
+                {Offset, Rest};
+            ?OFFSET_TYPE_TIMESTAMP ->
+                <<Timestamp:64/signed, Rest/binary>> = OffsetValuePropsBin,
+                {{timestamp, Timestamp}, Rest}
+        end,
+    request(CorrelationId, {resolve_offset_spec, Stream, OffsetSpec, parse_properties(PropsBin)});
 parse_request(Bin) ->
     {unknown, Bin}.
 
@@ -1020,7 +1059,10 @@ parse_response_body(?COMMAND_EXCHANGE_COMMAND_VERSIONS,
 parse_response_body(?COMMAND_STREAM_STATS,
                     <<ResponseCode:16, _Count:32, StatsBin/binary>>) ->
     Info = parse_int_map(StatsBin, #{}),
-    {stream_stats, ResponseCode, Info}.
+    {stream_stats, ResponseCode, Info};
+parse_response_body(?COMMAND_RESOLVE_OFFSET_SPEC,
+                    <<ResponseCode:16, Offset:64>>) ->
+    {resolve_offset_spec, ResponseCode, Offset}.
 
 offset_spec(OffsetType, OffsetValueBin) ->
     case OffsetType of
@@ -1135,6 +1177,13 @@ list_of_longcodes(<<>>) ->
 list_of_longcodes(<<I:64, C:16, Rem/binary>>) ->
     [{I, C} | list_of_longcodes(Rem)].
 
+parse_properties(<<>>) ->
+    #{};
+parse_properties(<<_Count:32, Bin/binary>>) ->
+    parse_map(Bin, #{});
+parse_properties(_) ->
+    #{}.
+
 command_id(declare_publisher) ->
     ?COMMAND_DECLARE_PUBLISHER;
 command_id(publish) ->
@@ -1198,7 +1247,9 @@ command_id(stream_stats) ->
 command_id(create_super_stream) ->
     ?COMMAND_CREATE_SUPER_STREAM;
 command_id(delete_super_stream) ->
-    ?COMMAND_DELETE_SUPER_STREAM.
+    ?COMMAND_DELETE_SUPER_STREAM;
+command_id(resolve_offset_spec) ->
+    ?COMMAND_RESOLVE_OFFSET_SPEC.
 
 parse_command_id(?COMMAND_DECLARE_PUBLISHER) ->
     declare_publisher;
@@ -1259,7 +1310,9 @@ parse_command_id(?COMMAND_STREAM_STATS) ->
 parse_command_id(?COMMAND_CREATE_SUPER_STREAM) ->
     create_super_stream;
 parse_command_id(?COMMAND_DELETE_SUPER_STREAM) ->
-    delete_super_stream.
+    delete_super_stream;
+parse_command_id(?COMMAND_RESOLVE_OFFSET_SPEC) ->
+    resolve_offset_spec.
 
 element_index(Element, List) ->
     element_index(Element, List, 0).
diff --git a/deps/rabbitmq_stream_common/test/rabbit_stream_core_SUITE.erl b/deps/rabbitmq_stream_common/test/rabbit_stream_core_SUITE.erl
@@ -119,6 +119,17 @@ roundtrip(_Config) ->
                      [<<"stream1">>, <<"stream2">>, <<"stream3">>], [<<"bk1">>, <<"bk2">>, <<"bk3">>],
                      #{}}}),
     test_roundtrip({request, 99, {delete_super_stream, <<"super_stream_name">>}}),
+    [test_roundtrip({request, 99,
+                     {resolve_offset_spec, <<"stream_name">>, Spec, #{}}})
+     || Spec
+            <- [last,
+                next,
+                first,
+                65432,
+                {timestamp, erlang:system_time(millisecond)}]],
+    test_roundtrip({request, 99,
+                    {resolve_offset_spec, <<"stream_name">>, first,
+                     #{<<"filter.0">> => <<"value0">>}}}),
 
     %% RESPONSES
     [test_roundtrip({response, 99, {Tag, 53}})
@@ -136,6 +147,7 @@ roundtrip(_Config) ->
 
     test_roundtrip({response, 99, {query_publisher_sequence, 98, 1234}}),
     test_roundtrip({response, 99, {query_offset, 1, 12}}),
+    test_roundtrip({response, 99, {resolve_offset_spec, 1, 12345}}),
 
     test_roundtrip({response, 99,
                     {peer_properties, 1, #{<<"k1">> => <<"v1">>}}}),
diff --git a/rabbitmq-components.mk b/rabbitmq-components.mk
@@ -52,7 +52,7 @@ dep_jose = hex 1.11.10
 dep_khepri = hex 0.17.5
 dep_khepri_mnesia_migration = hex 0.8.1
 dep_meck = hex 1.0.0
-dep_osiris = git https://github.com/rabbitmq/osiris v1.11.0
+dep_osiris = git https://github.com/rabbitmq/osiris resolve-offset-spec
 dep_prometheus = hex 6.1.1
 dep_ra = hex 2.17.2
 dep_ranch = hex 2.2.0
