Merge pull request #16031 from rabbitmq/mergify/bp/v4.2.x/pr-16030

Follow-up to #16020, #16023: integrate the special value of `0` (backport #16029) (backport #16030)

Author: michaelklishin

deps/rabbitmq_stream/src/rabbit_stream_reader.erl

@@ -110,7 +110,8 @@
 -ifdef(TEST).
 -export([ensure_token_expiry_timer/2,
          evaluate_state_after_secret_update/4,
-         clean_subscriptions/4]).
+         clean_subscriptions/4,
+         negotiate_frame_max/2]).
 -endif.
 
 callback_mode() ->
@@ -1470,10 +1471,8 @@ handle_frame_pre_auth(_Transport,
                       {tune, FrameMax, Heartbeat}) ->
     ?LOG_DEBUG("Tuning response ~tp ~tp ",
                                 [FrameMax, Heartbeat]),
-    %% The client must not negotiate a frame_max larger than the
-    %% value advertised by the server, so clamp it to the configured
-    %% ceiling before pushing the negotiated value into the parser.
-    NegotiatedFrameMax = min(FrameMax, ConfiguredFrameMax),
+    %% 0 on either side means "no limit" and must not be clamped to 0.
+    NegotiatedFrameMax = negotiate_frame_max(FrameMax, ConfiguredFrameMax),
     CoreState = rabbit_stream_core:set_frame_max(NegotiatedFrameMax,
                                                  CoreState0),
     Parent = self(),
@@ -1567,6 +1566,16 @@ handle_frame_pre_auth(_Transport, Connection, State, Command) ->
                                   [Command]),
     {Connection#stream_connection{connection_step = failure}, State}.
 
+%% 0 on either side means "no limit"; fall back to the other value.
+-spec negotiate_frame_max(non_neg_integer(), non_neg_integer()) ->
+    non_neg_integer().
+negotiate_frame_max(0, Configured) ->
+    Configured;
+negotiate_frame_max(Client, 0) ->
+    Client;
+negotiate_frame_max(Client, Configured) ->
+    min(Client, Configured).
+
 auth_fail(Username, Msg, Args, Connection, ConnectionState) ->
     notify_auth_result(Username,
                        user_authentication_failure,

deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl

@@ -77,6 +77,7 @@ groups() ->
        oversized_frame_rejected_post_auth,
        oversized_frame_rejected_after_tune_negotiation,
        frame_max_clamped_when_client_negotiates_higher,
+       client_tune_response_with_zero_frame_max_is_unlimited,
        test_stream_test_utils,
        sac_subscription_with_partition_index_conflict_should_return_error,
        test_metadata_with_advertised_hints,
@@ -671,6 +672,32 @@ frame_max_clamped_when_client_negotiates_higher(Config) ->
     ?assertMatch({request, _, {close, ?RESPONSE_CODE_FRAME_TOO_LARGE, _}}, Cmd),
     ?assertEqual(closed, wait_for_socket_close(gen_tcp, S, 1)).
 
+%% The stream protocol TUNE exchange follows the AMQP 0-9-1 convention
+%% where 0 means "no limit". A client that echoes FrameMax = 0 back to
+%% the server must not cause the server to clamp its parser to a
+%% 0-byte ceiling; the configured value must apply instead.
+client_tune_response_with_zero_frame_max_is_unlimited(Config) ->
+    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 = sasl_handshake(Transport, S, C1),
+    C3 = test_plain_sasl_authenticate(Transport, S, C2, <<"guest">>),
+    {{tune, _ServerFrameMax, _}, C4} = receive_commands(Transport, S, C3),
+    TuneResponse = frame({response, 0, {tune, 0, 0}}),
+    ok = Transport:send(S, TuneResponse),
+    OpenFrame = request(3, {open, <<"/">>}),
+    ok = Transport:send(S, OpenFrame),
+    %% If the server naively used min(0, Configured) = 0 for its
+    %% parser ceiling, the open frame itself would be rejected and
+    %% the server would send a close request instead of the expected
+    %% open response.
+    {Cmd, _C5} = receive_commands(Transport, S, C4),
+    ?assertMatch({response, 3, {open, ?RESPONSE_CODE_OK, _}}, Cmd),
+    gen_tcp:close(S).
+
 timeout_tcp_connected(Config) ->
     Port = get_stream_port(Config),
     {ok, S} = gen_tcp:connect("localhost", Port, [{active, false}, {mode, binary}]),

deps/rabbitmq_stream/test/rabbit_stream_reader_SUITE.erl

@@ -21,7 +21,8 @@
 -include_lib("rabbitmq_stream/src/rabbit_stream_reader.hrl").
 -include_lib("rabbitmq_stream_common/include/rabbit_stream.hrl").
 
--import(rabbit_stream_reader, [ensure_token_expiry_timer/2]).
+-import(rabbit_stream_reader, [ensure_token_expiry_timer/2,
+                               negotiate_frame_max/2]).
 
 %%%===================================================================
 %%% Common Test callbacks
@@ -228,6 +229,21 @@ partial_frame_buffering_test(_) ->
     ?assertEqual([], Commands2),
     ok.
 
+%% Covers every branch of negotiate_frame_max/2, including the cases
+%% where either side proposes 0 (protocol convention: "no limit").
+negotiate_frame_max_test(_) ->
+    %% Both sides propose explicit, positive limits: pick the lower.
+    ?assertEqual(1024, negotiate_frame_max(1024, 2048)),
+    ?assertEqual(1024, negotiate_frame_max(2048, 1024)),
+    ?assertEqual(1024, negotiate_frame_max(1024, 1024)),
+    %% Client proposes 0 (unlimited): the configured ceiling wins.
+    ?assertEqual(2048, negotiate_frame_max(0, 2048)),
+    %% Server is configured as unlimited: the client proposal wins.
+    ?assertEqual(1024, negotiate_frame_max(1024, 0)),
+    %% Both unlimited: stays unlimited (0 on the wire).
+    ?assertEqual(0, negotiate_frame_max(0, 0)),
+    ok.
+
 consumer(S, Pid) ->
     #consumer{configuration = #consumer_configuration{stream = S,
                                                       member_pid = Pid}}.

deps/rabbitmq_stream_common/src/rabbit_stream_core.erl

@@ -174,7 +174,7 @@
 
 -spec init(#{frame_max => non_neg_integer() | unlimited} | term()) -> state().
 init(Opts) when is_map(Opts) ->
-    FrameMax = maps:get(frame_max, Opts, unlimited),
+    FrameMax = normalise_frame_max(maps:get(frame_max, Opts, unlimited)),
     #?MODULE{cfg = #cfg{frame_max = FrameMax}};
 init(_) ->
     #?MODULE{cfg = #cfg{frame_max = unlimited}}.
@@ -184,7 +184,16 @@ init(_) ->
 %% the initial ceiling.
 -spec set_frame_max(non_neg_integer() | unlimited, state()) -> state().
 set_frame_max(FrameMax, #?MODULE{cfg = Cfg} = State) ->
-    State#?MODULE{cfg = Cfg#cfg{frame_max = FrameMax}}.
+    State#?MODULE{cfg = Cfg#cfg{frame_max = normalise_frame_max(FrameMax)}}.
+
+%% `frame_max` = 0 means "no limit" (AMQP 0-9-1 convention);
+%% the parser expresses that as the atom `unlimited`.
+-spec normalise_frame_max(non_neg_integer() | unlimited) ->
+    pos_integer() | unlimited.
+normalise_frame_max(0) ->
+    unlimited;
+normalise_frame_max(FrameMax) ->
+    FrameMax.
 
 -spec next_command(state()) -> {command(), state()} | empty.
 next_command(#?MODULE{commands = Commands0} = State) ->

deps/rabbitmq_stream_common/test/rabbit_stream_core_SUITE.erl

@@ -34,6 +34,8 @@ groups() ->
        set_frame_max_tightens_limit,
        set_frame_max_allows_in_flight_frame_to_complete,
        set_frame_max_to_unlimited,
+       init_with_zero_frame_max_means_unlimited,
+       set_frame_max_to_zero_means_unlimited,
        set_frame_max_preserves_pending_commands,
        prop_frame_within_limit_accepted,
        prop_frame_exceeding_limit_rejected,
@@ -367,6 +369,29 @@ set_frame_max_to_unlimited(_Config) ->
     {[{unknown, LargePayload}], _} = rabbit_stream_core:all_commands(State2),
     ok.
 
+%% The protocol convention is that frame_max = 0 means "no limit",
+%% as historically has been the case in RabbitMQ, AMQP 0-9-1.
+%% `init/1` must translate 0 to the parser's 'unlimited' sentinel,
+%% otherwise the parser would reject every frame as too large.
+init_with_zero_frame_max_means_unlimited(_Config) ->
+    Init = rabbit_stream_core:init(#{frame_max => 0}),
+    LargePayload = binary:copy(<<0>>, 5000),
+    LargeData = <<5000:32, LargePayload/binary>>,
+    State = rabbit_stream_core:incoming_data(LargeData, Init),
+    {[{unknown, LargePayload}], _} = rabbit_stream_core:all_commands(State),
+    ok.
+
+%% Same contract as above for `set_frame_max/2`: a post-TUNE update
+%% to 0 must resolve to "unlimited", not a 0-byte ceiling.
+set_frame_max_to_zero_means_unlimited(_Config) ->
+    Init = rabbit_stream_core:init(#{frame_max => 100}),
+    State1 = rabbit_stream_core:set_frame_max(0, Init),
+    LargePayload = binary:copy(<<0>>, 5000),
+    LargeData = <<5000:32, LargePayload/binary>>,
+    State2 = rabbit_stream_core:incoming_data(LargeData, State1),
+    {[{unknown, LargePayload}], _} = rabbit_stream_core:all_commands(State2),
+    ok.
+
 set_frame_max_preserves_pending_commands(_Config) ->
     InitialMax = 1000,
     TightenedMax = 50,