Streams: Soft remove policy configuration of max_segment_size_bytes

This configuration is not guaranteed to be safe to change after a stream has bee n
declared and thus we'll remove the ability to change it after the initial
declaration. Users should favour the x- queue arg for this config but it will still
be possible to configure it as a policy but it will only be evaluated at
declara tion time.

This means that if a policy is set for a stream that re-configures the
`stream-m ax-segment-size-bytes` key it will show in the UI as updated but
the pre-existing stream will not use the updated configuration.

The key has been removed from the UI but for backwards compatibility it is still
 settable.

NB: this PR adds a new command `update_config` to the stream coordinator state
machine. Strictly speaking this should require a new machine version but we're by
passing that by relying on the feature flag instead which avoids this command
being committed before all nodes have the new code version. A new machine version
can lower the availability properties during a rolling cluster upgrade so in
this case it is preferable to avoid that given the simplicity of the change.

Author: kjnilsson

deps/rabbit/src/rabbit_core_ff.erl

@@ -148,3 +148,11 @@
                          post_enable =>
                          {rabbit_khepri, khepri_db_migration_post_enable}}
      }}).
+
+-rabbit_feature_flag(
+   {stream_update_config_command,
+    #{desc          => "A new internal command that is used to update streams as "
+                        "part of a policy.",
+      stability     => stable,
+      depends_on    => [stream_queue]
+     }}).

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -236,7 +236,17 @@ delete_replica(StreamId, Node) ->
 
 policy_changed(Q) when ?is_amqqueue(Q) ->
     StreamId = maps:get(name, amqqueue:get_type_state(Q)),
-    process_command({policy_changed, StreamId, #{queue => Q}}).
+    case rabbit_feature_flags:is_enabled(stream_update_config_command) of
+        true ->
+            %% there are the only two configuration keys that are safe to
+            %% update
+            Conf = maps:with([filter_size,
+                              retention],
+                             rabbit_stream_queue:update_stream_conf(Q, #{})),
+            process_command({update_config, StreamId, Conf});
+        false ->
+            process_command({policy_changed, StreamId, #{queue => Q}})
+    end.
 
 sac_state(#?MODULE{single_active_consumer = SacState}) ->
     SacState.
@@ -1601,12 +1611,13 @@ update_stream0(#{system_time := _Ts} = _Meta,
                         M
                 end, Members0),
     Stream0#stream{members = Members};
-update_stream0(#{system_time := _Ts},
-               {policy_changed, _StreamId, #{queue := Q}},
-               #stream{conf = Conf0,
-                       members = _Members0} = Stream0) ->
+update_stream0(_Meta, {policy_changed, _StreamId, #{queue := Q}},
+               #stream{conf = Conf0} = Stream0) ->
     Conf = rabbit_stream_queue:update_stream_conf(Q, Conf0),
     Stream0#stream{conf = Conf};
+update_stream0(_Meta, {update_config, _StreamId, Conf},
+               #stream{conf = Conf0} = Stream0) ->
+    Stream0#stream{conf = maps:merge(Conf0, Conf)};
 update_stream0(_Meta, _Cmd, undefined) ->
     undefined.
 

deps/rabbit/test/rabbit_stream_queue_SUITE.erl

@@ -249,6 +249,12 @@ init_per_group1(Group, Config) ->
                          Config2, stream_queue),
             case EnableFF of
                 ok ->
+                    if Clustered ->
+                           rabbit_ct_broker_helpers:enable_feature_flag(
+                             Config2, stream_update_config_command);
+                       true ->
+                           ok
+                    end,
                     ok = rabbit_ct_broker_helpers:rpc(
                            Config2, 0, application, set_env,
                            [rabbit, channel_tick_interval, 100]),

deps/rabbitmq_management/priv/www/js/global.js

@@ -207,9 +207,6 @@ var HELP = {
     'queue-max-age':
       'How long a message published to a stream queue can live before it is discarded.',
 
-    'queue-stream-max-segment-size-bytes':
-      'Total segment size for stream segments on disk.<br/>(Sets the x-stream-max-segment-size-bytes argument.)',
-
     'queue-stream-filter-size-bytes':
       'Size of the filter data attached to each stream chunk.<br/>(Sets the x-stream-filter-size-bytes argument.)',
 

deps/rabbitmq_management/priv/www/js/tmpl/policies.ejs

@@ -139,8 +139,6 @@
                 <td>
                   <span class="argument-link" field="definition" key="max-age" type="string">Max age</span>
                   <span class="help" id="queue-max-age"></span> |
-                  <span class="argument-link" field="definition" key="stream-max-segment-size-bytes" type="number">Max segment size in bytes</span>
-                  <span class="help" id="queue-stream-max-segment-size-bytes"></span> |
                   <span class="argument-link" field="definition" key="stream-filter-size-bytes" type="number">Filter size in bytes. Valid range: 16-255</span>
                   <span class="help" id="queue-stream-filter-size-bytes"></span> |
                   <span class="argument-link" field="definition" key="queue-leader-locator" type="string">Leader locator</span>