Merge pull request #15381 from rabbitmq/stream-resolve-offset-spec

Add resolve_stream_offset stream protocol command

Author: kjnilsson

deps/rabbitmq_stream/docs/PROTOCOL.adoc

@@ -240,6 +240,11 @@ used to make the difference between a request (0) and a response (1). Example fo
 |0x001e
 |Yes
 
+|<<resolveoffsetspec>>
+|Client
+|0x001f
+|Yes
+
 |===
 
 === DeclarePublisher
@@ -795,6 +800,43 @@ Delete => Key Version CorrelationId Name
   Name => string
 ```
 
+=== ResolveOffsetSpec
+
+```
+ResolveOffsetSpecRequest => Key Version CorrelationId Stream OffsetSpecification Properties
+  Key => uint16 // 0x001f
+  Version => uint16
+  CorrelationId => uint32
+  Stream => string
+  OffsetSpecification => OffsetType Offset
+  OffsetType => uint16 // 1 (first), 2 (last), 3 (next), 4 (offset), 5 (timestamp)
+  Offset => uint64 (for offset) | int64 (for timestamp)
+  Properties => [Property]
+  Property => Key Value
+  Key => string
+  Value => string
+
+ResolveOffsetSpecResponse => Key Version CorrelationId ResponseCode OffsetType Offset
+  Key => uint16 // 0x801f
+  Version => uint16
+  CorrelationId => uint32
+  ResponseCode => uint16
+  OffsetType => uint16 // 4 (offset)
+  Offset => uint64
+```
+
+Resolves an offset specification to a concrete offset in a stream.
+The client sends an offset specification (first, last, next, a specific offset, or a timestamp) and the server resolves it to the actual offset in the stream.
+
+The response includes the type of the resolved value and the value itself.
+Currently, the resolved value type is always `offset` (4).
+
+Supported request properties:
+
+* `filter.` (e.g. `filter.0`, `filter.1`, etc): prefix to use to define filter values
+* `match-unfiltered`: whether to consider messages without any filter value or not
+* `chunk_selector`: chunk selector to use for offset resolution
+
 == Authentication
 
 Once a client is connected to the server, it initiates an authentication

deps/rabbitmq_stream/src/rabbit_stream_reader.erl

@@ -2123,6 +2123,50 @@ 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_TYPE_OFFSET, Offset}}),
+    send(Transport, S, Frame),
+    {Connection1, State};
 handle_frame_post_auth(Transport,
                        #stream_connection{stream_subscriptions =
                                               StreamSubscriptions} =
@@ -2821,7 +2865,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 +4039,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).

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).

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
@@ -1237,6 +1239,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,
@@ -1730,6 +1782,26 @@ 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,
+                                ?OFFSET_TYPE_OFFSET, _}},
+                 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,
+                                ?OFFSET_TYPE_OFFSET, ExpectedOffset}},
+                 Cmd),
+    C.
+
 test_close(Transport, S, C0) ->
     CloseReason = <<"OK">>,
     CloseFrame = request({close, ?RESPONSE_CODE_OK, CloseReason}),

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).