Merge stream.read_ahead and stream.read_ahead_limit options

Read-ahead can be controlled by a single option, with zero acting the
same as `false`.

(cherry picked from commit 76ce4ee)

# Conflicts:
#	deps/rabbit/src/rabbit_stream_queue.erl

Author: the-mikedavis

deps/rabbit/priv/schema/rabbit.schema

@@ -2777,12 +2777,7 @@ fun(Conf) ->
 end}.
 
 {mapping, "stream.read_ahead", "rabbit.stream_read_ahead",
-    [{datatype, {enum, [true, false]}}]}.
-
-{mapping, "stream.read_ahead_limit", "rabbit.stream_read_ahead_limit", [
-    {datatype, [integer, string]},
-    {validators, ["is_supported_information_unit"]}
-]}.
+    [{datatype, [{enum, [true, false]}, integer, string]}]}.
 
 {mapping, "cluster_tags.$tag", "rabbit.cluster_tags", [
     {datatype, [binary]}

deps/rabbit/src/rabbit_stream_queue.erl

@@ -50,7 +50,7 @@
 -export([format_osiris_event/2]).
 -export([update_stream_conf/2]).
 -export([readers/1]).
--export([read_ahead_on/0, read_ahead_limit/0]).
+-export([read_ahead/0]).
 
 -export([parse_offset_arg/1,
          filter_spec/1]).
@@ -468,8 +468,7 @@ begin_stream(#stream_client{name = QName,
              Tag, Offset, Mode, AckRequired, Filter, Options0)
   when is_pid(LocalPid) ->
     CounterSpec = {{?MODULE, QName, Tag, self()}, []},
-    Options1 = Options0#{read_ahead => read_ahead_on(),
-                         read_ahead_limit => read_ahead_limit()},
+    Options1 = Options0#{read_ahead => read_ahead()},
     {ok, Seg0} = osiris:init_reader(LocalPid, Offset, CounterSpec, Options1),
     NextOffset = osiris_log:next_offset(Seg0) - 1,
     osiris:register_offset_listener(LocalPid, NextOffset),
@@ -1528,6 +1527,7 @@ queue_vm_ets() ->
     {[],
      []}.
 
+<<<<<<< HEAD
 shrink_all(_Node) ->
     {error, not_quorum_queue}.
 
@@ -1541,13 +1541,22 @@ read_ahead_limit() ->
             undefined;
         Bytes when is_integer(Bytes) ->
             Bytes;
+=======
+-spec read_ahead() -> boolean() | non_neg_integer().
+read_ahead() ->
+    case application:get_env(rabbit, stream_read_ahead, true) of
+        Toggle when is_boolean(Toggle) ->
+            Toggle;
+        LimitBytes when is_integer(LimitBytes) ->
+            LimitBytes;
+>>>>>>> 76ce4ee42 (Merge `stream.read_ahead` and `stream.read_ahead_limit` options)
         Limit when is_list(Limit) ->
             case rabbit_resource_monitor_misc:parse_information_unit(Limit) of
                 {ok, ParsedLimit} ->
                     ParsedLimit;
                 {error, parse_error} ->
                     ?LOG_ERROR("Unable to parse stream read ahead limit value "
                                "~tp", [Limit]),
-                    undefined
+                    true
             end
     end.

deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets

@@ -1247,7 +1247,7 @@ credential_validator.regexp = ^abc\\d+",
    []},
 
   %%
-  %% Stream read ahead on/off
+  %% Stream read ahead on/off/limit
   %%
 
   {stream_read_ahead,
@@ -1267,24 +1267,20 @@ credential_validator.regexp = ^abc\\d+",
       {stream_read_ahead, false}
      ]}],
    []},
-
-  %%
-  %% Stream read limit
-  %%
   {stream_read_ahead_limit_bytes,
    "
-    stream.read_ahead_limit = 8192
+    stream.read_ahead = 8192
    ",
    [{rabbit, [
-      {stream_read_ahead_limit, 8192}
+      {stream_read_ahead, 8192}
      ]}],
    []},
   {stream_read_ahead_limit_information_unit,
    "
-    stream.read_ahead_limit = 8KiB
+    stream.read_ahead = 8KiB
    ",
    [{rabbit, [
-      {stream_read_ahead_limit, "8KiB"}
+      {stream_read_ahead, "8KiB"}
      ]}],
    []}
 

deps/rabbitmq_stream/src/rabbit_stream_reader.erl

@@ -2814,8 +2814,7 @@ init_reader(ConnectionTransport,
     CounterSpec = {{?MODULE, QueueResource, SubscriptionId, self()}, []},
     Options0 = #{transport => ConnectionTransport,
                  chunk_selector => get_chunk_selector(Properties),
-                 read_ahead => rabbit_stream_queue:read_ahead_on(),
-                 read_ahead_limit => rabbit_stream_queue:read_ahead_limit()},
+                 read_ahead => rabbit_stream_queue:read_ahead()},
 
     Options1 = maps:merge(Options0,
                           rabbit_stream_utils:filter_spec(Properties)),

rabbitmq-components.mk

@@ -49,7 +49,7 @@ dep_jose = hex 1.11.10
 dep_khepri = hex 0.17.2
 dep_khepri_mnesia_migration = hex 0.8.0
 dep_meck = hex 1.0.0
-dep_osiris = git https://github.com/rabbitmq/osiris v1.10.1
+dep_osiris = git https://github.com/rabbitmq/osiris v1.10.2
 dep_prometheus = hex 5.1.1
 dep_ra = hex 2.17.1
 dep_ranch = hex 2.2.0