Return a protocol exception if the safe queue name is duplicated

2 files changed, 48 insertions, 22 deletions

diff --git a/src/rabbit_stream2_queue.erl b/src/rabbit_stream2_queue.erl
index 1d4530226b..124427566b 100644
--- a/src/rabbit_stream2_queue.erl
+++ b/src/rabbit_stream2_queue.erl
@@ -241,27 +241,33 @@ declare(Q0) ->
     N = ra_lib:derive_safe_string(LName, length(LName)),
     Conf = #{reference => QName,
              name => list_to_atom(N)},
-    {ok, LeaderPid, ReplicaPids} = osiris:start_cluster(N, Replicas, Conf),
-    Q1 = amqqueue:set_pid(Q0, LeaderPid),
-    NewQ1 = amqqueue:set_type_state(Q1, maps:put(replicas, ReplicaPids, Conf)),
-    case rabbit_amqqueue:internal_declare(NewQ1, false) of
-        {created, Q} ->
-            rabbit_event:notify(queue_created,
-                                [{name, QName},
-                                 {durable, true},
-                                 {auto_delete, false},
-                                 {arguments, Arguments},
-                                 {user_who_performed_action,
-                                  ActingUser}]),
-            {new, Q};
-        {error, Error} ->
-            _ = rabbit_amqqueue:internal_delete(QName, ActingUser),
-            rabbit_misc:protocol_error(
-              internal_error,
-              "Cannot declare a queue '~s' on node '~s': ~255p",
-              [rabbit_misc:rs(QName), node(), Error]);
-        {existing, _} = Ex ->
-            Ex
+    case osiris:start_cluster(N, Replicas, Conf) of
+        {ok, LeaderPid, ReplicaPids} ->
+            Q1 = amqqueue:set_pid(Q0, LeaderPid),
+            NewQ1 = amqqueue:set_type_state(Q1, maps:put(replicas, ReplicaPids, Conf)),
+            case rabbit_amqqueue:internal_declare(NewQ1, false) of
+                {created, Q} ->
+                    rabbit_event:notify(queue_created,
+                                        [{name, QName},
+                                         {durable, true},
+                                         {auto_delete, false},
+                                         {arguments, Arguments},
+                                         {user_who_performed_action,
+                                          ActingUser}]),
+                    {new, Q};
+                {error, Error} ->
+                    _ = rabbit_amqqueue:internal_delete(QName, ActingUser),
+                    rabbit_misc:protocol_error(
+                      internal_error,
+                      "Cannot declare a queue '~s' on node '~s': ~255p",
+                      [rabbit_misc:rs(QName), node(), Error]);
+                {existing, _} = Ex ->
+                    Ex
+            end;
+        {error, {already_started, _}} ->
+            rabbit_misc:protocol_error(precondition_failed,
+                                       "safe queue name already in use '~s'",
+                                       [N])
     end.
 
 recover(Q0) ->
diff --git a/test/rabbit_stream2_queue_SUITE.erl b/test/rabbit_stream2_queue_SUITE.erl
index 1985d4b4dc..eed313aa98 100644
--- a/test/rabbit_stream2_queue_SUITE.erl
+++ b/test/rabbit_stream2_queue_SUITE.erl
@@ -51,7 +51,8 @@ all_tests() ->
      time_travel,
      idempotent_declare_queue,
      delete_queue,
-     zenflix
+     zenflix,
+     declare_queue
     ].
 
 %% -------------------------------------------------------------------
@@ -399,6 +400,25 @@ zenflix(Config) ->
     flush(100),
     ok.
 
+declare_queue(Config) ->
+    [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
+
+    QName = <<"declare_queue">>,
+    SimilarQName = <<"'declare_queue'">>,
+
+    ?assertEqual({'queue.declare_ok', QName, 0, 0},
+                 declare(Ch, QName, [{<<"x-queue-type">>, longstr,
+                                      ?config(queue_type, Config)}])),
+
+    ?assertExit({{shutdown, {connection_closing, {server_initiated_close, 541, _}}}, _},
+                declare(Ch, SimilarQName, [{<<"x-queue-type">>, longstr,
+                                            ?config(queue_type, Config)}])),
+
+    flush(100),
+    ok.
+
 %% HELPERS
 
 assert_declare(Ch, QName, Type) ->