Hunt BOSH interleave requests issue

apps/ejabberd/src/mod_bosh_socket.erl

@@ -417,7 +417,7 @@ handle_stream_event({EventTag, Body, Rid} = Event, Handler,
                 is_expected_rid(Rid, ExpectedRid),
                 is_acceptable_rid(Rid, ExpectedRid)}
     of
-        {_, {true, CachedResponse}, _, _} ->
+        {_, {true, CachedResponse}, _, _} when Handler /= none ->
             case CachedResponse of
                 none ->
                     NS;
@@ -894,7 +894,8 @@ bosh_wrap(Elements, Rid, #state{} = S) ->
     end,
     MaybeAck = maybe_ack(Rid, NS),
     {MaybeReport, NNS} = maybe_report(NS),
-    MaybeStreamPrefix = maybe_stream_prefix(Children),
+    HasStreamPrefix = (exml_query:attr(Body, <<"xmlns:stream">>) /= undefined),
+    MaybeStreamPrefix = maybe_stream_prefix(HasStreamPrefix, Children),
     ExtraAttrs = MaybeAck ++ MaybeReport ++ MaybeStreamPrefix,
     {Body#xmlel{attrs = Body#xmlel.attrs ++ ExtraAttrs,
                 children = maybe_add_default_ns_to_children(Children)}, NNS}.
@@ -965,16 +966,19 @@ bosh_stream_end_body() ->
                     {<<"xmlns">>, ?NS_HTTPBIND}],
            children = []}.
 
-maybe_stream_prefix(Stanzas) ->
-    case lists:any(fun is_stream_error/1, Stanzas) of
+maybe_stream_prefix(true, _) ->
+    [];
+maybe_stream_prefix(_, Stanzas) ->
+    case lists:any(fun is_stream_prefix/1, Stanzas) of
         false ->
             [];
         true ->
             [{<<"xmlns:stream">>, ?NS_STREAM}]
     end.
 
-is_stream_error(#xmlel{name = Name}) ->
-    Name =:= <<"stream:error">>.
+is_stream_prefix(#xmlel{name = <<"stream:error">>}) -> true;
+is_stream_prefix(#xmlel{name = <<"stream:features">>}) -> true;
+is_stream_prefix(_) -> false.
 
 %%--------------------------------------------------------------------
 %% ejabberd_socket compatibility

test/ejabberd_tests/rebar.config

@@ -11,7 +11,7 @@
         {katt, ".*", {git, "git://github.com/for-GET/katt.git", {tag, "1.5.2"}}},
         {erlsh, ".*", {git, "git://github.com/proger/erlsh.git", "2fce513"}},
         {jiffy, ".*", {git, "git://github.com/davisp/jiffy.git", "0.14.8"}},
-        {proper, ".*", {git, "https://github.com/manopapad/proper.git", "20e62bc32f9bd43fe2ff52944a4ef99eb71d1399"}},
+        {proper, ".*", {git, "https://github.com/manopapad/proper.git", "v1.2"}},
         {cowboy, ".*", {git, "git://github.com/ninenines/cowboy.git", {tag, "1.0.4"}}}
 ]}.
 

test/ejabberd_tests/src/ct_tty_hook.erl

@@ -93,7 +93,8 @@ post_end_per_testcase(TC, _Config, Return, State) ->
 
 %% @doc Called after post_init_per_suite, post_end_per_suite, post_init_per_group,
 %% post_end_per_group and post_end_per_testcase if the suite, group or test case failed.
-on_tc_fail(_TC, _Reason, State) ->
+on_tc_fail(TC, Reason, State) ->
+    ct:print("~p~n~p", [TC, Reason]),
     State.
 
 %% @doc Called when a test case is skipped by either user action
@@ -148,4 +149,4 @@ print_case_enter(Group, #state{print_case = true}, Msg) ->
     escalus_ejabberd:rpc(error_logger, warning_msg, ["====== ~s CASE ~p",
         [Msg, Group]]);
 print_case_enter(_Group, _State, _Msg) ->
-    ok.
+    ok.

test/ejabberd_tests/tests/bosh_SUITE.erl

@@ -38,7 +38,8 @@ all() ->
      {group, acks},
 
      {group, essential_https},
-     {group, chat_https}
+     {group, chat_https},
+     {group, interleave_requests_statem}
      ].
 
 groups() ->
@@ -47,7 +48,9 @@ groups() ->
      {chat, [shuffle], chat_test_cases()},
      {chat_https, [shuffle], chat_test_cases()},
      {time, [parallel], time_test_cases()},
-     {acks, [shuffle], acks_test_cases()}].
+     {acks, [shuffle], acks_test_cases()},
+     {interleave_requests_statem, [parallel], [interleave_requests_statem]}
+     ].
 
 suite() ->
     escalus:suite().
@@ -69,7 +72,8 @@ chat_test_cases() ->
      cant_send_invalid_rid,
      multiple_stanzas,
      namespace,
-     stream_error].
+     stream_error
+    ].
 
 time_test_cases() ->
     [disconnect_inactive,
@@ -268,23 +272,41 @@ interleave_requests(Config) ->
         Rid = get_bosh_rid(Carol),
         Sid = get_bosh_sid(Carol),
 
-        Empty2 = escalus_bosh:empty_body(Rid + 1, Sid),
-        Chat2 = Empty2#xmlel{
-                children = [escalus_stanza:chat_to(Geralt, <<"2nd!">>)]},
-        escalus_bosh:send_raw(Carol, Chat2),
+        Msg1 = <<"1st!">>,
+        Msg2 = <<"2nd!">>,
+        Msg3 = <<"3rd!">>,
+        Msg4 = <<"4th!">>,
+
+        send_message_with_rid(Carol, Geralt, Rid + 1, Sid, Msg2),
+        send_message_with_rid(Carol, Geralt, Rid, Sid, Msg1),
 
-        Empty1 = escalus_bosh:empty_body(Rid, Sid),
-        Chat1 = Empty1#xmlel{
-                children = [escalus_stanza:chat_to(Geralt, <<"1st!">>)]},
-        escalus_bosh:send_raw(Carol, Chat1),
+        send_message_with_rid(Carol, Geralt, Rid + 2,   Sid, Msg3),
+        send_message_with_rid(Carol, Geralt, Rid + 3,   Sid, Msg4),
 
-        escalus:assert(is_chat_message, [<<"1st!">>],
+        escalus:assert(is_chat_message, [Msg1],
+                       escalus_client:wait_for_stanza(Geralt)),
+        escalus:assert(is_chat_message, [Msg2],
+                       escalus_client:wait_for_stanza(Geralt)),
+        escalus:assert(is_chat_message, [Msg3],
+                       escalus_client:wait_for_stanza(Geralt)),
+        escalus:assert(is_chat_message, [Msg4],
                        escalus_client:wait_for_stanza(Geralt)),
-        escalus:assert(is_chat_message, [<<"2nd!">>],
-                       escalus_client:wait_for_stanza(Geralt))
 
+        true = escalus_bosh:is_connected(Carol)
     end).
 
+interleave_requests_statem(Config) ->
+    true = bosh_interleave_reqs:test([{user, carol} | Config]).
+
+interleave_requests_statem_https(Config) ->
+    interleave_requests_statem([{user, carol_s} | Config]).
+
+send_message_with_rid(From, To, Rid, Sid, Msg) ->
+    Empty = escalus_bosh:empty_body(Rid, Sid),
+    Chat = Empty#xmlel{
+             children = [escalus_stanza:chat_to(To, Msg)]},
+    escalus_bosh:send_raw(From, Chat).
+
 
 simple_chat(Config) ->
     escalus:story(Config, [{?config(user, Config), 1}, {geralt, 1}], fun(Carol, Geralt) ->

test/ejabberd_tests/tests/bosh_interleave_reqs.erl

@@ -0,0 +1,179 @@
+-module(bosh_interleave_reqs).
+
+-behaviour(proper_statem).
+
+-include_lib("proper/include/proper.hrl").
+% -include_lib("eqc/include/eqc_statem.hrl").
+% -include_lib("eqc/include/eqc.hrl").
+-include_lib("escalus/include/escalus.hrl").
+-include_lib("exml/include/exml.hrl").
+
+-export([test/1, sample/0, prop/1]).
+
+-export([initial_state/1, command/1, precondition/2, postcondition/3,
+         next_state/3]).
+
+-export([read_config/1,
+         connect_carol/1,
+         connect_geralt/1,
+         send_from_carol/2,
+         send_from_geralt/2,
+         wait_for_msgs_carol/2,
+         wait_for_msgs_geralt/2]).
+
+-export([ct_config_giver/1]).
+
+-record(state, {carol,
+                geralt,
+                msgs_to_carol,
+                msgs_to_geralt,
+                config_pid}).
+
+test(Config) ->
+    proper:quickcheck(?MODULE:prop(Config)).
+
+sample() ->
+    proper_gen:pick(commands(?MODULE)).
+
+prop(Config) ->
+    Pid = spawn_link(?MODULE, ct_config_giver, [Config]),
+    ?FORALL(Cmds, commands(?MODULE, initial_state(Pid)),
+            ?TRAPEXIT(
+               begin
+                   {History, State, Result} = run_commands(?MODULE, Cmds),
+                   maybe_stop_client(State#state.carol),
+                   maybe_stop_client(State#state.geralt),
+                   ?WHENFAIL(ct:log(error, "History: ~p~nState: ~p\nResult: ~p~n",
+                                    [History, State, Result]),
+                             aggregate(command_names(Cmds), Result =:= ok))
+               end)).
+
+ct_config_giver(Config) ->
+    receive
+        {give_me_config, Pid} ->
+            Pid ! {ok, Config},
+            ct_config_giver(Config)
+    end.
+
+maybe_stop_client(undefined) -> ok;
+maybe_stop_client(Client) ->
+    escalus_client:stop(Client).
+
+initial_state(Pid) ->
+    #state{carol = undefined,
+           geralt = undefined,
+           msgs_to_carol = [],
+           msgs_to_geralt = [],
+           config_pid = Pid}.
+
+command(S) ->
+    Cmds = possible_commands(S),
+    oneof(Cmds).
+
+possible_commands(S) ->
+    Carol = (S#state.carol /= undefined),
+    Geralt = (S#state.geralt /= undefined),
+    Users = Carol andalso Geralt,
+    MsgsToCarol = (S#state.msgs_to_carol /= []),
+    MsgsToGeralt = (S#state.msgs_to_geralt /= []),
+    Pid = S#state.config_pid,
+    [{call, ?MODULE, connect_carol, [Pid]} || not Carol] ++
+    [{call, ?MODULE, connect_geralt, [Pid]} || not Geralt] ++
+    [{call, ?MODULE, send_from_carol, [S#state.carol, S#state.geralt]} || Users] ++
+    [{call, ?MODULE, send_from_geralt, [S#state.geralt, S#state.carol]} || Users] ++
+    [{call, ?MODULE, wait_for_msgs_carol, [S#state.carol, S#state.msgs_to_carol]}
+     || MsgsToCarol] ++
+    [{call, ?MODULE, wait_for_msgs_geralt, [S#state.geralt, S#state.msgs_to_geralt]}
+     || MsgsToGeralt].
+
+
+precondition(_, _) -> true.
+postcondition(_, _, _) -> true.
+
+next_state(S, V, {call, _, connect_carol, [_]}) ->
+    S#state{carol = V};
+next_state(S, V, {call, _, connect_geralt, [_]}) ->
+    S#state{geralt = V};
+next_state(#state{msgs_to_geralt = Msgs} = S, V, {call, _, send_from_carol, _}) ->
+    S#state{msgs_to_geralt = [V | Msgs]};
+next_state(#state{msgs_to_carol = Msgs} = S, V, {call, _, send_from_geralt, _}) ->
+    S#state{msgs_to_carol = [V | Msgs]};
+next_state(S, _, {call, _, wait_for_msgs_carol, _}) ->
+    S#state{msgs_to_carol = []};
+next_state(S, _, {call, _, wait_for_msgs_geralt, _}) ->
+    S#state{msgs_to_geralt = []};
+next_state(S, _, _) ->
+    S.
+
+read_config(Pid) ->
+    Pid ! {give_me_config, self()},
+    receive
+        {ok, Config} ->
+            Config
+    after 100 ->
+              error
+    end.
+
+connect_carol(Pid) ->
+    Spec = given_fresh_spec(read_config(Pid), carol),
+    connect_user([{keepalive, true} | Spec]).
+
+connect_geralt(Pid) ->
+    Spec = given_fresh_spec(read_config(Pid), alice),
+    connect_user(Spec).
+
+given_fresh_spec(Config, User) ->
+    NewConfig = escalus_fresh:create_users(Config, [{User, 1}]),
+    escalus_users:get_userspec(NewConfig, User).
+
+connect_user(Spec) ->
+    Res = base64:encode(crypto:rand_bytes(4)),
+    {ok, Conn, Props, _} = escalus_connection:start([{resource, Res} | Spec]),
+    JID = make_jid(Props),
+    escalus:send(Conn, escalus_stanza:presence(<<"available">>)),
+    escalus:wait_for_stanza(Conn, timer:seconds(5)),
+    Conn#client{jid = JID}.
+
+make_jid(Proplist) ->
+    {username, U} = lists:keyfind(username, 1, Proplist),
+    {server, S} = lists:keyfind(server, 1, Proplist),
+    {resource, R} = lists:keyfind(resource, 1, Proplist),
+    <<U/binary, "@", S/binary, "/", R/binary>>.
+
+send_from_carol(Carol, Geralt) ->
+    Msg = gen_msg(),
+    escalus:send(Carol, escalus_stanza:chat_to(Geralt, Msg)),
+    {escalus_client:short_jid(Carol),
+     Msg
+    }.
+
+send_from_geralt(Geralt, Carol) ->
+    Msg = gen_msg(),
+    escalus:send(Geralt, escalus_stanza:chat_to(Carol, Msg)),
+    {escalus_client:short_jid(Geralt),
+     Msg
+    }.
+
+gen_msg() ->
+    Msg = base64:encode(crypto:rand_bytes(15)),
+    Msg.
+
+wait_for_msgs_carol(Carol, Msgs) ->
+    L = length(Msgs),
+    Stanzas = escalus:wait_for_stanzas(Carol, L, timer:seconds(5)),
+    RMsgs = [exml_query:path(E, [{element, <<"body">>}, cdata])
+             || E <- Stanzas],
+    SortedMsgs = lists:sort([Msg || {_, Msg} <- Msgs]),
+    SortedMsgs = lists:sort(RMsgs),
+    ok.
+
+wait_for_msgs_geralt(Geralt, Msgs) ->
+    wait_for_msgs(Geralt, lists:reverse(Msgs)).
+
+wait_for_msgs(_Client, []) ->
+    ok;
+wait_for_msgs(Client, [{_, Msg} | Rest]) ->
+    escalus:assert(is_chat_message, [Msg],
+                   escalus:wait_for_stanza(Client, timer:seconds(5))),
+    wait_for_msgs(Client, Rest).
+