fix rabbit_direct_reply_to:compute_key_and_suffix_v1 type signature

deps/rabbit/BUILD.bazel

@@ -1046,6 +1046,14 @@ suites = [
             ":quorum_queue_utils",
         ],
     ),
+    rabbitmq_integration_suite(
+        PACKAGE,
+        name = "rabbit_direct_reply_to_prop_SUITE",
+        size = "small",
+        deps = [
+            "@proper//:erlang_app",
+        ],
+    ),
 ]
 
 assert_suites(

deps/rabbit/src/rabbit_direct_reply_to.erl

@@ -22,7 +22,7 @@
 %% API
 %%
 
--type decoded_pid_and_key() :: {ok, pid(), binary()} | {error, any()}.
+-type decoded_pid_and_key() :: {ok, pid(), binary()}.
 
 -spec compute_key_and_suffix_v1(pid()) -> {binary(), binary()}.
 %% This original pid encoding function produces values that exceed routing key length limit
@@ -63,16 +63,17 @@ compute_key_and_suffix_v2(Pid) ->
 
 -spec decode_reply_to_v2(binary(), #{non_neg_integer() => node()}) -> decoded_pid_and_key() | {error, any()}.
 decode_reply_to_v2(Bin, CandidateNodes) ->
-    case string:lexemes(Bin, ".") of
-        [PidEnc, Key] ->
-            RawPidBin = base64:decode(PidEnc),
-            PidParts0 = #{node := ShortenedNodename} = pid_recomposition:from_binary(RawPidBin),
-            {_, NodeHash} = rabbit_nodes_common:parts(ShortenedNodename),
-            case maps:get(list_to_integer(NodeHash), CandidateNodes, undefined) of
-                undefined -> error;
-                Candidate ->
-                    PidParts = maps:update(node, Candidate, PidParts0),
-                    {ok, pid_recomposition:recompose(PidParts), unicode:characters_to_binary(Key)}
-            end;
-        _             -> {error, unrecognized_format}
+    try
+        [PidEnc, Key] = binary:split(Bin, <<".">>),
+        RawPidBin = base64:decode(PidEnc),
+        PidParts0 = #{node := ShortenedNodename} = pid_recomposition:from_binary(RawPidBin),
+        {_, NodeHash} = rabbit_nodes_common:parts(ShortenedNodename),
+        case maps:get(list_to_integer(NodeHash), CandidateNodes, undefined) of
+            undefined -> {error, target_node_not_found};
+            Candidate ->
+                PidParts = maps:update(node, Candidate, PidParts0),
+                {ok, pid_recomposition:recompose(PidParts), Key}
+        end
+    catch
+        error:_ -> {error, unrecognized_format}
     end.

deps/rabbit/test/rabbit_direct_reply_to_prop_SUITE.erl

@@ -0,0 +1,70 @@
+-module(rabbit_direct_reply_to_prop_SUITE).
+
+-compile(export_all).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("proper/include/proper.hrl").
+
+-define(ITERATIONS_TO_RUN_UNTIL_CONFIDENT, 10000).
+
+all() ->
+    [
+     decode_reply_to_v2
+    ].
+
+init_per_suite(Config) ->
+    Config.
+
+end_per_suite(_Config) ->
+    ok.
+
+init_per_group(_Group, Config) ->
+    Config.
+
+end_per_group(_Group, _Config) ->
+    ok.
+
+init_per_testcase(_TestCase, Config) ->
+    Config.
+
+end_per_testcase(_TestCase, _Config) ->
+    ok.
+
+%%% Tests %%%
+
+
+decode_reply_to_v2(Config) ->
+    rabbit_ct_proper_helpers:run_proper(
+      fun() -> prop_decode_reply_to(Config) end,
+      [],
+      ?ITERATIONS_TO_RUN_UNTIL_CONFIDENT).
+
+prop_decode_reply_to(_) ->
+    ?FORALL({Len, Random}, {pos_integer(), binary()},
+        begin
+            Key      = <<"apple">>,
+            NodeList = lists:map(
+                         fun(I) -> {I, list_to_atom(integer_to_list(I))} end,
+                         lists:seq(1, Len)
+                        ),
+
+            [ {Ix, Node} | NoNodeList ] = NodeList,
+
+            PidParts   = #{node => Node, id => 0, serial => 0, creation => 0},
+            IxParts    = PidParts#{node := rabbit_nodes_common:make("banana", Ix)},
+            IxPartsEnc = base64:encode(pid_recomposition:to_binary(IxParts)),
+            IxBin      = <<IxPartsEnc/binary, ".", Key/binary>>,
+
+            NodeMap   = maps:from_list(NodeList),
+            NoNodeMap = maps:from_list(NoNodeList),
+
+            %% There is non-zero chance Random is a valid encoded Pid.
+            NonB64 = <<0, Random/binary>>, 
+
+            {ok, pid_recomposition:recompose(PidParts), Key} =:=
+                rabbit_direct_reply_to:decode_reply_to_v2(IxBin, NodeMap)
+            andalso {error, target_node_not_found} =:=
+                rabbit_direct_reply_to:decode_reply_to_v2(IxBin, NoNodeMap)
+            andalso {error, unrecognized_format} =:=
+                rabbit_direct_reply_to:decode_reply_to_v2(NonB64, NodeMap)
+        end).