1: %% Stream Management tests
2: -module(sm_SUITE).
3:
4: -compile([export_all, nowarn_export_all]).
5: -include_lib("exml/include/exml.hrl").
6: -include_lib("eunit/include/eunit.hrl").
7: -include_lib("escalus/include/escalus.hrl").
8: -include_lib("common_test/include/ct.hrl").
9:
10: %% Injected code callbacks
11: -export([rpc_start_hook_handler/3,
12: rpc_stop_hook_handler/2,
13: hook_handler_fn/3,
14: regression_handler/5]).
15:
16: -export([rpc_start_filter_hook_handler/3,
17: rpc_stop_filter_hook_handler/2,
18: filter_hook_handler_fn/3]).
19:
20: -import(distributed_helper, [mim/0,
21: require_rpc_nodes/1,
22: rpc/4]).
23:
24: -import(domain_helper, [host_type/0]).
25:
26: -import(sm_helper, [connect_fresh/3,
27: connect_fresh/4,
28: connect_spec/2,
29: connect_spec/3,
30: connect_same/2,
31: connect_same/3,
32: connect_resume/2,
33: process_initial_stanza/1,
34: send_initial_presence/1,
35: get_ack/1,
36: ack_initial_presence/1]).
37:
38: -define(MOD_SM, mod_stream_management).
39: -define(CONSTRAINT_CHECK_TIMEOUT, 5000).
40: -define(LONG_TIMEOUT, 3600).
41: -define(SHORT_TIMEOUT, 1).
42: -define(SMALL_SM_BUFFER, 3).
43: -define(PING_REQUEST_TIMEOUT, 1).
44: -define(PING_INTERVAL, 3).
45:
46: %%--------------------------------------------------------------------
47: %% Suite configuration
48: %%--------------------------------------------------------------------
49:
50: suite() ->
51: require_rpc_nodes([mim]) ++ escalus:suite().
52:
53: all() ->
54: ct_helper:groups_to_all(groups()).
55:
56: groups() ->
57: G = [
58: % {parallel, [parallel], parallel_cases()},
59: % {parallel_manual_ack_freq_1, [parallel], parallel_manual_ack_freq_1_cases()},
60: % {manual_ack_freq_2, [], manual_ack_freq_2_cases()},
61: % {stale_h, [], stale_h_cases()},
62: % {parallel_unacknowledged_message_hook, [parallel], parallel_unacknowledged_message_hook_cases()}
63: {ping, [], [ping_timeout]}
64: ],
65: ct_helper:repeat_all_until_any_fail(G, 1000).
66:
67: parallel_cases() ->
68: [server_announces_sm,
69: server_enables_sm_before_session,
70: server_enables_sm_after_session,
71: server_returns_failed_after_start,
72: server_returns_failed_after_auth,
73: server_enables_resumption,
74: client_enables_sm_twice_fails_with_correct_error_stanza,
75: session_resumed_then_old_session_is_closed_gracefully_with_correct_error_stanza,
76: session_resumed_and_old_session_dead_doesnt_route_error_to_new_session,
77: basic_ack,
78: h_ok_before_session,
79: h_ok_after_session_enabled_before_session,
80: h_ok_after_session_enabled_after_session,
81: h_ok_after_a_chat,
82: h_non_given_closes_stream_gracefully,
83: resend_unacked_on_reconnection,
84: session_established,
85: wait_for_resumption,
86: resume_session,
87: resume_session_with_wrong_h_does_not_leak_sessions,
88: resume_session_with_wrong_sid_returns_item_not_found,
89: resume_session_with_wrong_namespace_is_a_noop,
90: resume_dead_session_results_in_item_not_found,
91: resume_session_kills_old_C2S_gracefully,
92: carboncopy_works,
93: carboncopy_works_after_resume,
94: aggressively_pipelined_resume,
95: replies_are_processed_by_resumed_session,
96: subscription_requests_are_buffered_properly,
97: messages_are_properly_flushed_during_resumption].
98:
99: parallel_manual_ack_freq_1_cases() ->
100: [client_acks_more_than_sent,
101: too_many_unacked_stanzas,
102: resend_unacked_after_resume_timeout,
103: resume_session_state_send_message_with_ack,
104: resume_session_state_send_message_without_ack,
105: resume_session_state_stop_c2s,
106: server_requests_ack_after_session,
107: resend_more_offline_messages_than_buffer_size,
108: server_requests_ack].
109:
110: manual_ack_freq_2_cases() ->
111: [server_requests_ack_freq_2].
112:
113: stale_h_cases() ->
114: [resume_expired_session_returns_correct_h,
115: gc_repeat_after_never_means_no_cleaning,
116: gc_repeat_after_timeout_does_clean].
117:
118: stream_mgmt_disabled_cases() ->
119: [no_crash_if_stream_mgmt_disabled_but_client_requests_stream_mgmt,
120: no_crash_if_stream_mgmt_disabled_but_client_requests_stream_mgmt_with_resumption].
121:
122: manual_ack_freq_long_session_timeout_cases() ->
123: [preserve_order].
124:
125: parallel_unacknowledged_message_hook_cases() ->
126: [unacknowledged_message_hook_bounce,
127: unacknowledged_message_hook_offline,
128: unacknowledged_message_hook_resume,
129: unacknowledged_message_hook_filter].
130:
131: %%--------------------------------------------------------------------
132: %% Init & teardown
133: %%--------------------------------------------------------------------
134:
135: init_per_suite(Config) ->
136: NewConfig = dynamic_modules:save_modules(host_type(), Config),
137: NewConfigWithSM = escalus_users:update_userspec(NewConfig, alice, stream_management, true),
138: mongoose_helper:inject_module(?MODULE),
139: escalus:init_per_suite(NewConfigWithSM).
140:
141: end_per_suite(Config) ->
142: escalus_fresh:clean(),
143: dynamic_modules:restore_modules(Config),
144: escalus:end_per_suite(Config).
145:
146: init_per_group(Group, Config) when Group =:= parallel_unacknowledged_message_hook;
147: Group =:= manual_ack_freq_long_session_timeout;
148: Group =:= parallel_manual_ack_freq_1;
149: Group =:= manual_ack_freq_2 ->
150: dynamic_modules:ensure_modules(host_type(), required_modules(group, Group)),
151: Config;
152: init_per_group(stale_h, Config) ->
153: Config;
154: init_per_group(stream_mgmt_disabled, Config) ->
155: dynamic_modules:stop(host_type(), ?MOD_SM),
156: rpc(mim(), mnesia, delete_table, [sm_session]),
157: Config;
158: init_per_group(ping, Config) ->
159: Config;
160: init_per_group(Group, Config) ->
161: dynamic_modules:ensure_modules(host_type(), required_modules(group, Group)),
162: Config.
163:
164: end_per_group(_Group, _Config) ->
165: ok.
166:
167: init_per_testcase(resume_expired_session_returns_correct_h = CN, Config) ->
168: dynamic_modules:ensure_modules(host_type(), required_modules(testcase, CN)),
169: escalus:init_per_testcase(CN, Config);
170: init_per_testcase(CN, Config) when CN =:= gc_repeat_after_never_means_no_cleaning;
171: CN =:= gc_repeat_after_timeout_does_clean ->
172: dynamic_modules:ensure_modules(host_type(), required_modules(testcase, CN)),
173: Config2 = register_some_smid_h(Config),
174: escalus:init_per_testcase(CN, Config2);
175: init_per_testcase(ping_timeout = CN, Config) ->
176: ok = rpc(mim(), meck, new, [mod_ping, [passthrough, no_link]]),
177: dynamic_modules:ensure_modules(host_type(), required_modules(testcase, CN)),
178: escalus:init_per_testcase(CN, Config);
179: init_per_testcase(server_requests_ack_freq_2 = CN, Config) ->
180: escalus:init_per_testcase(CN, Config);
181: init_per_testcase(replies_are_processed_by_resumed_session = CN, Config) ->
182: register_handler(),
183: escalus:init_per_testcase(CN, Config);
184: init_per_testcase(CaseName, Config) ->
185: escalus:init_per_testcase(CaseName, Config).
186:
187: end_per_testcase(CN, Config) when CN =:= resume_expired_session_returns_correct_h;
188: CN =:= gc_repeat_after_never_means_no_cleaning;
189: CN =:= gc_repeat_after_timeout_does_clean ->
190: Name = rpc(mim(), gen_mod, get_module_proc, [host_type(), stream_management_stale_h]),
191: rpc(mim(), ejabberd_sup, stop_child, [Name]),
192: escalus:end_per_testcase(CN, Config);
193: end_per_testcase(replies_are_processed_by_resumed_session = CN, Config) ->
194: unregister_handler(),
195: escalus:end_per_testcase(CN, Config);
196: end_per_testcase(ping_timeout = CN, Config) ->
197: rpc(mim(), meck, unload, [mod_ping]),
198: escalus:end_per_testcase(CN, Config);
199: end_per_testcase(CaseName, Config) ->
200: escalus:end_per_testcase(CaseName, Config).
201:
202: %% Module configuration per group (in case of stale_h group it is per testcase)
203:
204: required_modules(Scope, Name) ->
205: SMConfig = case required_sm_opts(Scope, Name) of
206: stopped -> stopped;
207: ExtraOpts -> maps:merge(common_sm_opts(), ExtraOpts)
208: end,
209: Backend = mongoose_helper:mnesia_or_rdbms_backend(),
210: BaseModules = [
211: {mod_stream_management, config_parser_helper:mod_config(mod_stream_management, SMConfig)},
212: {mod_offline, config_parser_helper:mod_config(mod_offline, #{backend => Backend})}
213: ],
214: case Name of
215: ping_timeout ->
216: BaseModules ++ [{mod_ping, config_parser_helper:mod_config(mod_ping, mod_ping_opts())}];
217: _ ->
218: BaseModules
219: end.
220:
221: required_sm_opts(group, parallel) ->
222: #{ack_freq => never};
223: required_sm_opts(group, parallel_manual_ack_freq_1) ->
224: #{ack_freq => 1,
225: resume_timeout => ?LONG_TIMEOUT};
226: required_sm_opts(group, manual_ack_freq_2) ->
227: #{ack_freq => 2};
228: required_sm_opts(group, stream_mgmt_disabled) ->
229: stopped;
230: required_sm_opts(group, parallel_unacknowledged_message_hook) ->
231: #{ack_freq => 1};
232: required_sm_opts(group, manual_ack_freq_long_session_timeout) ->
233: #{ack_freq => 1, buffer_max => 1000};
234: required_sm_opts(testcase, resume_expired_session_returns_correct_h) ->
235: #{ack_freq => 1,
236: resume_timeout => ?SHORT_TIMEOUT,
237: stale_h => stale_h(?LONG_TIMEOUT, ?LONG_TIMEOUT)};
238: required_sm_opts(testcase, gc_repeat_after_never_means_no_cleaning) ->
239: #{stale_h => stale_h(?LONG_TIMEOUT, ?SHORT_TIMEOUT)};
240: required_sm_opts(testcase, gc_repeat_after_timeout_does_clean) ->
241: #{stale_h => stale_h(?SHORT_TIMEOUT, ?SHORT_TIMEOUT)};
242: required_sm_opts(testcase, ping_timeout) ->
243: #{ack_freq => 1,
244: resume_timeout => ?SHORT_TIMEOUT}.
245:
246: common_sm_opts() ->
247: Backend = ct_helper:get_internal_database(),
248: #{buffer_max => ?SMALL_SM_BUFFER, backend => Backend}.
249:
250: stale_h(RepeatAfter, Geriatric) ->
251: #{enabled => true,
252: repeat_after => RepeatAfter,
253: geriatric => Geriatric}.
254:
255: make_smid() ->
256: base64:encode(crypto:strong_rand_bytes(21)).
257:
258: register_smid(IntSmidId) ->
259: S = {SMID = make_smid(), IntSmidId},
260: ok = rpc(mim(), ?MOD_SM, register_stale_smid_h, [host_type(), SMID, IntSmidId]),
261: S.
262:
263: register_some_smid_h(Config) ->
264: TestSmids = lists:map(fun register_smid/1, lists:seq(1, 3)),
265: [{smid_test, TestSmids} | Config].
266:
267: mod_ping_opts() ->
268: #{send_pings => true,
269: ping_interval => ?PING_INTERVAL,
270: ping_req_timeout => ?PING_REQUEST_TIMEOUT,
271: timeout_action => kill}.
272:
273: %%--------------------------------------------------------------------
274: %% Tests
275: %%--------------------------------------------------------------------
276:
277: server_announces_sm(Config) ->
278: AliceSpec = escalus_fresh:freshen_spec(Config, alice),
279: {ok, #client{props = Props}, Features} = escalus_connection:start(AliceSpec,
280: [start_stream]),
281: true = escalus_session:can_use_stream_management(Props, Features).
282:
283:
284: server_enables_sm_before_session(Config) ->
285: connect_fresh(Config, alice, sm_after_bind).
286:
287: server_enables_sm_after_session(Config) ->
288: connect_fresh(Config, alice, sm_after_session).
289:
290: server_returns_failed_after_start(Config) ->
291: Alice = connect_fresh(Config, alice, before_auth),
292: server_returns_failed(Alice).
293:
294: server_returns_failed_after_auth(Config) ->
295: Alice = connect_fresh(Config, alice, auth),
296: server_returns_failed(Alice).
297:
298: server_enables_resumption(Config) ->
299: Alice = connect_fresh(Config, alice, sr_presence),
300: escalus_connection:stop(Alice).
301:
302: server_returns_failed(Alice) ->
303: escalus_connection:send(Alice, escalus_stanza:enable_sm()),
304: escalus:assert(is_sm_failed, [<<"unexpected-request">>],
305: escalus_connection:get_stanza(Alice, enable_sm_failed)).
306:
307: client_enables_sm_twice_fails_with_correct_error_stanza(Config) ->
308: Alice = connect_fresh(Config, alice, sm_before_session),
309: escalus_connection:send(Alice, escalus_stanza:enable_sm()),
310: escalus:assert(is_sm_failed, [<<"unexpected-request">>],
311: escalus_connection:get_stanza(Alice, enable_sm_failed)),
312: escalus:assert(is_stream_end,
313: escalus_connection:get_stanza(Alice, enable_sm_failed)),
314: true = escalus_connection:wait_for_close(Alice, timer:seconds(5)).
315:
316: session_resumed_then_old_session_is_closed_gracefully_with_correct_error_stanza(Config) ->
317: %% GIVEN USER WITH STREAM RESUMPTION ENABLED
318: Alice = connect_fresh(Config, alice, sr_presence),
319: SMH = escalus_connection:get_sm_h(Alice),
320: %% WHEN USER RESUMES SESSION FROM NEW CLIENT
321: Alice2 = connect_resume(Alice, SMH),
322: process_initial_stanza(Alice2),
323: %% THEN: Old session is gracefully closed with the correct error stanza
324: escalus:assert(is_stream_error, [<<"conflict">>, <<>>],
325: escalus_connection:get_stanza(Alice, close_old_stream)),
326: escalus:assert(is_stream_end,
327: escalus_connection:get_stanza(Alice, close_old_stream)),
328: true = escalus_connection:wait_for_close(Alice, timer:seconds(5)),
329: true = escalus_connection:is_connected(Alice2),
330: escalus_connection:stop(Alice2).
331:
332: session_resumed_and_old_session_dead_doesnt_route_error_to_new_session(Config) ->
333: %% GIVEN USER WITH STREAM RESUMPTION ENABLED
334: Alice = connect_fresh(Config, alice, sr_presence),
335: %% WHEN FIRST SESSION DIES AND USER RESUMES FROM NEW CLIENT
336: Alice2 = sm_helper:kill_and_connect_resume(Alice),
337: process_initial_stanza(Alice2),
338: %% THEN new session does not have any message rerouted
339: false = escalus_client:has_stanzas(Alice2),
340: true = escalus_connection:is_connected(Alice2),
341: escalus_connection:stop(Alice2).
342:
343: basic_ack(Config) ->
344: Alice = connect_fresh(Config, alice, sm_after_session),
345: escalus_connection:send(Alice, escalus_stanza:roster_get()),
346: escalus:assert(is_roster_result,
347: escalus_connection:get_stanza(Alice, roster_result)),
348: escalus_connection:send(Alice, escalus_stanza:sm_request()),
349: escalus:assert(is_sm_ack,
350: escalus_connection:get_stanza(Alice, stream_mgmt_ack)).
351:
352: %% Test that "h" value is valid when:
353: %% - SM is enabled *before* the session is established
354: %% - <r/> is sent *before* the session is established
355: h_ok_before_session(Config) ->
356: Alice = connect_fresh(Config, alice, sm_after_bind),
357: escalus_connection:send(Alice, escalus_stanza:sm_request()),
358: escalus:assert(is_sm_ack, [0],
359: escalus_connection:get_stanza(Alice, stream_mgmt_ack)).
360:
361: %% Test that "h" value is valid when:
362: %% - SM is enabled *before* the session is established
363: %% - <r/> is sent *after* the session is established
364: h_ok_after_session_enabled_before_session(Config) ->
365: Alice = connect_fresh(Config, alice, sm_before_session),
366: escalus_connection:send(Alice, escalus_stanza:sm_request()),
367: escalus:assert(is_sm_ack, [1],
368: escalus_connection:get_stanza(Alice, stream_mgmt_ack)).
369:
370: %% Test that "h" value is valid when:
371: %% - SM is enabled *after* the session is established
372: %% - <r/> is sent *after* the session is established
373: h_ok_after_session_enabled_after_session(Config) ->
374: Alice = connect_fresh(Config, alice, sm_after_session),
375: escalus_connection:send(Alice, escalus_stanza:roster_get()),
376: escalus:assert(is_roster_result,
377: escalus_connection:get_stanza(Alice, roster_result)),
378: escalus_connection:send(Alice, escalus_stanza:sm_request()),
379: escalus:assert(is_sm_ack, [1],
380: escalus_connection:get_stanza(Alice, stream_mgmt_ack)).
381:
382: %% Test that "h" value is valid after exchanging a few messages.
383: h_ok_after_a_chat(ConfigIn) ->
384: Config = escalus_users:update_userspec(ConfigIn, alice,
385: stream_management, true),
386: escalus:fresh_story(Config, [{alice,1}, {bob,1}], fun(Alice, Bob) ->
387: escalus:send(Alice, escalus_stanza:chat_to(Bob, <<"Hi, Bob!">>)),
388: escalus:assert(is_chat_message, [<<"Hi, Bob!">>],
389: escalus:wait_for_stanza(Bob)),
390: escalus:send(Bob, escalus_stanza:chat_to(Alice, <<"Hi, Alice!">>)),
391: escalus:assert(is_chat_message, [<<"Hi, Alice!">>],
392: escalus:wait_for_stanza(Alice)),
393: escalus:send(Bob, escalus_stanza:chat_to(Alice, <<"How's life?">>)),
394: escalus:assert(is_chat_message, [<<"How's life?">>],
395: escalus:wait_for_stanza(Alice)),
396: escalus:send(Alice, escalus_stanza:chat_to(Bob, <<"Pretty !@#$%^$">>)),
397: escalus:assert(is_chat_message, [<<"Pretty !@#$%^$">>],
398: escalus:wait_for_stanza(Bob)),
399: escalus:send(Alice, escalus_stanza:sm_request()),
400: escalus:assert(is_sm_ack, [3], escalus:wait_for_stanza(Alice)),
401: %% Ack, so that unacked messages don't go into offline store.
402: escalus:send(Alice, escalus_stanza:sm_ack(3))
403: end).
404:
405: h_non_given_closes_stream_gracefully(ConfigIn) ->
406: AStanza = #xmlel{name = <<"a">>,
407: attrs = [{<<"xmlns">>, <<"urn:xmpp:sm:3">>}]},
408: Config = escalus_users:update_userspec(ConfigIn, alice,
409: stream_management, true),
410: escalus:fresh_story(Config, [{alice,1}], fun(Alice) ->
411: C2SPid = mongoose_helper:get_session_pid(Alice),
412: escalus:send(Alice, AStanza),
413: escalus:assert(is_stream_error,
414: [<<"policy-violation">>, <<>>],
415: escalus:wait_for_stanza(Alice)),
416: mongoose_helper:wait_for_pid_to_die(C2SPid),
417: escalus:assert(is_stream_end, escalus_connection:get_stanza(Alice, stream_end)),
418: true = escalus_connection:wait_for_close(Alice, timer:seconds(5))
419: end).
420:
421: client_acks_more_than_sent(Config) ->
422: Alice = connect_fresh(Config, alice, sm_after_session),
423: escalus:send(Alice, escalus_stanza:sm_ack(5)),
424: StreamErrorStanza = escalus:wait_for_stanza(Alice),
425: %% Assert "undefined-condition" children
426: escalus:assert(is_stream_error, [<<"undefined-condition">>, <<>>], StreamErrorStanza),
427: %% Assert "handled-count-too-high" children with correct attributes
428: HandledCountSubElement = exml_query:path(StreamErrorStanza,
429: [{element_with_ns,
430: <<"handled-count-too-high">>,
431: <<"urn:xmpp:sm:3">>}]),
432: <<"5">> = exml_query:attr(HandledCountSubElement, <<"h">>),
433: <<"0">> = exml_query:attr(HandledCountSubElement, <<"send-count">>),
434: %% Assert graceful stream end
435: escalus:assert(is_stream_end, escalus_connection:get_stanza(Alice, stream_end)),
436: true = escalus_connection:wait_for_close(Alice, timer:seconds(5)).
437:
438: too_many_unacked_stanzas(Config) ->
439: Bob = connect_fresh(Config, bob, presence),
440: Alice = connect_fresh(Config, alice, sm_presence, manual),
441: get_ack(Alice),
442: [escalus:send(Bob, escalus_stanza:chat_to(Alice,
443: <<(integer_to_binary(N))/binary, ": Hi, Alice!">>))
444: || N <- lists:seq(1,?SMALL_SM_BUFFER)],
445: escalus:wait_for_stanzas(Alice, ?SMALL_SM_BUFFER * 2), % messages and ack requests
446: escalus:assert(is_stream_error, [<<"resource-constraint">>,
447: <<"too many unacked stanzas">>],
448: %% wait for deferred buffer check
449: escalus:wait_for_stanza(Alice, ?CONSTRAINT_CHECK_TIMEOUT + 1000)).
450:
451: server_requests_ack(Config) ->
452: server_requests_ack(Config, 1).
453:
454: server_requests_ack_freq_2(Config) ->
455: server_requests_ack(Config, 2).
456:
457: server_requests_ack(Config, N) ->
458: Bob = connect_fresh(Config, bob, presence),
459: Alice = connect_fresh(Config, alice, sm_presence, manual),
460: %% ack request after initial presence
461: maybe_assert_ack_request(1, N, Alice),
462: escalus:send(Bob, escalus_stanza:chat_to(Alice, <<"Hi, Alice!">>)),
463: escalus:assert(is_chat_message, [<<"Hi, Alice!">>],
464: escalus:wait_for_stanza(Alice)),
465: maybe_assert_ack_request(2, N, Alice).
466:
467: maybe_assert_ack_request(StanzasRec, AckRequests, Alice) ->
468: ct:log("StanzasRec: ~p, AckRequests: ~p", [StanzasRec, AckRequests]),
469: case StanzasRec rem AckRequests of
470: 0 ->
471: escalus:assert(is_sm_ack_request, escalus:wait_for_stanza(Alice));
472: _ ->
473: ok
474: end,
475: StanzasRec.
476:
477: server_requests_ack_after_session(Config) ->
478: Alice = connect_fresh(Config, alice, sm_before_session, manual),
479: escalus:assert(is_sm_ack_request, escalus_connection:get_stanza(Alice, stream_mgmt_req)).
480:
481: resend_more_offline_messages_than_buffer_size(Config) ->
482: %% connect bob and alice
483: Bob = connect_fresh(Config, bob, presence),
484: AliceSpec = escalus_fresh:create_fresh_user(Config, alice),
485:
486: % sent some messages - more than unacked buffer size
487: MessagesToSend = ?SMALL_SM_BUFFER + 1,
488: AliceJid = common_helper:get_bjid(AliceSpec),
489: [escalus_connection:send(Bob, escalus_stanza:chat_to(AliceJid, integer_to_binary(I)))
490: || I <- lists:seq(1, MessagesToSend)],
491: mongoose_helper:wait_for_n_offline_messages(AliceJid, MessagesToSend),
492:
493: % connect alice who wants to receive all messages from offline storage
494: Alice = connect_spec(AliceSpec, sm_after_session, manual),
495: send_initial_presence(Alice),
496: escalus:wait_for_stanzas(Alice, MessagesToSend * 2), %messages and ack requests
497:
498: escalus_connection:get_stanza(Alice, presence),
499: get_ack(Alice), % ack request
500:
501: % confirm messages + presence
502: escalus_connection:send(Alice, escalus_stanza:sm_ack(4)),
503: % wait for check constraint message on server side
504:
505: ct:sleep(?CONSTRAINT_CHECK_TIMEOUT + 1000),
506: false = escalus_client:has_stanzas(Alice),
507: % should not receive anything especially any stream errors
508:
509: escalus_connection:stop(Alice),
510: escalus_connection:stop(Bob).
511:
512: resend_unacked_on_reconnection(Config) ->
513: Texts = three_texts(),
514: Bob = connect_fresh(Config, bob, presence),
515: Alice = connect_fresh(Config, alice, sm_presence),
516: AliceSpec = sm_helper:client_to_spec0(Alice),
517: %% Bob sends some messages to Alice.
518: sm_helper:send_messages(Bob, Alice, Texts),
519: %% Alice receives the messages.
520: sm_helper:wait_for_messages(Alice, Texts),
521: %% Alice disconnects without acking the messages.
522: sm_helper:stop_client_and_wait_for_termination(Alice),
523: %% Messages go to the offline store.
524: %% Alice receives the messages from the offline store.
525: NewAlice = connect_spec(AliceSpec, session, manual),
526: send_initial_presence(NewAlice),
527: sm_helper:wait_for_messages(NewAlice, Texts),
528: %% Alice acks the delayed messages so they won't go again
529: %% to the offline store.
530: escalus_connection:send(NewAlice, escalus_stanza:sm_ack(3)).
531:
532: %% Remove wait_for_n_offline_messages and you will get anything, but preserve_order
533: %% TODO Test without wait_for_n_offline_messages. It would require changes in SM
534: %% and more strict tests, reproducing delays in SM and in mod_offline.
535: preserve_order(Config) ->
536: %% connect bob and alice
537: Bob = connect_fresh(Config, bob, presence),
538: Alice = connect_fresh(Config, alice, sr_presence, manual),
539: AliceSpec = sm_helper:client_to_spec(Alice),
540: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"1">>)),
541:
542: %% kill alice connection
543: escalus_connection:kill(Alice),
544: C2SPid = mongoose_helper:get_session_pid(Alice),
545: sm_helper:wait_until_resume_session(C2SPid),
546:
547: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"2">>)),
548: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"3">>)),
549:
550: NewAlice = connect_spec(AliceSpec, session, manual),
551: escalus_connection:send(NewAlice, escalus_stanza:enable_sm([resume])),
552:
553: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"4">>)),
554: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"5">>)),
555:
556: %% Without this check we will get stuff out of order
557: mongoose_helper:wait_for_n_offline_messages(NewAlice, 5),
558:
559: send_initial_presence(NewAlice),
560: %% Without this check we can get "6, 1, 2, 3, 4, 5" messages in the next receive_all_ordered
561: mongoose_helper:wait_for_n_offline_messages(NewAlice, 0),
562: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"6">>)),
563:
564: %% "2, 3, 4, 5, 6, 1" is possible (where only 1 is from offline storage, the rest is from sm)
565: receive_all_ordered(NewAlice, 6),
566:
567: % replace connection
568: NewAlice2 = connect_spec(AliceSpec, session, manual),
569: % allow messages to go to the offline storage
570: mongoose_helper:wait_for_n_offline_messages(NewAlice, 6),
571:
572: send_initial_presence(NewAlice2),
573:
574: % receves messages in correct order
575: receive_all_ordered(NewAlice2, 6),
576:
577: escalus_connection:stop(Bob),
578: escalus_connection:stop(NewAlice2).
579:
580: receive_all_ordered(Conn, Last) ->
581: receive_all_ordered(Conn, 1, Last, []).
582:
583: %% Receive messages from N to Last.
584: %% Ignores acks and presences.
585: %% Handles case when out of order and when not enough stanzas.
586: receive_all_ordered(_Conn, N, Last, Acc) when N > Last ->
587: Texts = lists:map(fun integer_to_binary/1, lists:seq(1, Last)),
588: sm_helper:assert_messages(Acc, Texts);
589: receive_all_ordered(Conn, N, Last, Acc) ->
590: Stanzas = escalus:wait_for_stanzas(Conn, 1),
591: case Stanzas of
592: [#xmlel{name = <<"message">>}] ->
593: receive_all_ordered(Conn, N + 1, Last, Acc ++ Stanzas);
594: [_] -> %% Ack or presence
595: receive_all_ordered(Conn, N, Last, Acc);
596: [] ->
597: ct:fail({timeout_waiting, N, Acc})
598: end.
599:
600: resend_unacked_after_resume_timeout(Config) ->
601: %% connect bob and alice
602: Bob = connect_fresh(Config, bob, presence),
603: Alice = connect_fresh(Config, alice, sr_presence),
604: AliceSpec = sm_helper:client_to_spec(Alice),
605:
606: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"msg-1">>)),
607: %% kill alice connection
608: escalus_connection:kill(Alice),
609:
610: %% ensure there is no session
611: C2SPid = mongoose_helper:get_session_pid(Alice),
612: sm_helper:wait_until_resume_session(C2SPid),
613:
614: %% alice come back and receives unacked message
615: NewAlice = connect_spec(AliceSpec, session),
616: send_initial_presence(NewAlice),
617:
618: escalus_new_assert:mix_match([is_presence, is_chat(<<"msg-1">>)],
619: escalus:wait_for_stanzas(NewAlice, 2)),
620:
621: escalus_connection:stop(Bob),
622: escalus_connection:stop(NewAlice).
623:
624: ping_timeout(Config) ->
625: %% make sure there are no leftover stanzas in the history
626: ?assertEqual([], get_stanzas_filtered_by_mod_ping()),
627:
628: %% connect Alice and wait for the session to close
629: Alice = connect_fresh(Config, alice, sr_presence),
630: SMH = escalus_connection:get_sm_h(Alice),
631: SMID = sm_helper:client_to_smid(Alice),
632:
633: escalus_client:wait_for_stanza(Alice),
634: ct:sleep(?PING_REQUEST_TIMEOUT + ?PING_INTERVAL + timer:seconds(1)),
635:
636: %% attempt to resume the session after the connection drop
637: sm_helper:kill_client_and_wait_for_termination(Alice),
638: NewAlice = connect_same(Alice, auth),
639: escalus_connection:send(NewAlice, escalus_stanza:resume(SMID, SMH)),
640:
641: %% after resume_timeout, we expect the session to be closed
642: escalus_connection:get_stanza(NewAlice, failed_resumption),
643:
644: %% bind a new session and expect unacknowledged messages to be resent
645: escalus_session:session(escalus_session:bind(NewAlice)),
646: send_initial_presence(NewAlice),
647:
648: %% check if the error stanza was handled by mod_ping
649: [Stanza] = get_stanzas_filtered_by_mod_ping(),
650: escalus:assert(is_iq_error, Stanza),
651: ?assertNotEqual(undefined,
652: exml_query:subelement_with_name_and_ns(Stanza, <<"ping">>, <<"urn:xmpp:ping">>)),
653:
654: %% stop the connection
655: escalus_connection:stop(NewAlice).
656:
657: resume_expired_session_returns_correct_h(Config) ->
658: %% connect bob and alice
659: Bob = connect_fresh(Config, bob, sr_presence),
660: Alice = connect_fresh(Config, alice, sr_presence, manual),
661: get_ack(Alice),
662:
663: %% Bob sends a message to Alice, and Alice receives it but doesn't acknowledge
664: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"msg-1">>)),
665: escalus:assert(is_chat_message, [<<"msg-1">>], escalus:wait_for_stanza(Alice)),
666: %% alice comes back, but too late, so resumption doesn't work,
667: %% but she receives the previous h = 1 anyway
668: %% NewAlice is also manual ack
669: NewAlice = sm_helper:kill_and_connect_with_resume_session_without_waiting_for_result(Alice),
670: FailedResumption = escalus_connection:get_stanza(NewAlice, failed_resumption),
671: <<"1">> = exml_query:attr(FailedResumption, <<"h">>),
672: %% And we can continue with bind and session
673: escalus_session:session(escalus_session:bind(NewAlice)),
674: send_initial_presence(NewAlice),
675: Stanzas = [escalus_connection:get_stanza(NewAlice, {msg, 1}),
676: escalus_connection:get_stanza(NewAlice, {msg, 2})],
677: escalus_new_assert:mix_match([is_presence, is_chat(<<"msg-1">>)], Stanzas),
678: escalus_connection:stop(Bob),
679: escalus_connection:stop(NewAlice).
680:
681: gc_repeat_after_never_means_no_cleaning(Config) ->
682: [{SMID1, _}, {SMID2, _}, {SMID3, _}] = ?config(smid_test, Config),
683: {stale_h, 1} = rpc(mim(), ?MOD_SM, get_session_from_smid, [host_type(), SMID1]),
684: {stale_h, 2} = rpc(mim(), ?MOD_SM, get_session_from_smid, [host_type(), SMID2]),
685: {stale_h, 3} = rpc(mim(), ?MOD_SM, get_session_from_smid, [host_type(), SMID3]).
686:
687: gc_repeat_after_timeout_does_clean(Config) ->
688: [{SMID1, _} | _ ] = ?config(smid_test, Config),
689: mongoose_helper:wait_until(fun() ->
690: rpc(mim(), ?MOD_SM, get_stale_h, [host_type(), SMID1])
691: end,
692: {error, smid_not_found},
693: #{name => smid_garbage_collected}).
694:
695: resume_session_state_send_message_without_ack(Config) ->
696: resume_session_state_send_message_generic(Config, no_ack).
697:
698: resume_session_state_send_message_with_ack(Config) ->
699: resume_session_state_send_message_generic(Config, ack).
700:
701: resume_session_state_send_message_generic(Config, AckInitialPresence) ->
702: %% connect bob and alice
703: Bob = connect_fresh(Config, bob, presence),
704: Alice = connect_fresh(Config, alice, sr_presence, manual),
705: maybe_ack_initial_presence(Alice, AckInitialPresence),
706: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"msg-1">>)),
707: %% kill alice connection
708: C2SPid = mongoose_helper:get_session_pid(Alice),
709: escalus_connection:kill(Alice),
710: sm_helper:wait_until_resume_session(C2SPid),
711: sm_helper:assert_alive_resources(Alice, 1),
712:
713: %% send some messages and check if c2s can handle it
714: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"msg-2">>)),
715: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"msg-3">>)),
716: %% suspend the process to ensure that Alice has enough time to reconnect,
717: %% before resumption timeout occurs.
718: ok = rpc(mim(), sys, suspend, [C2SPid]),
719:
720: %% alice comes back and receives unacked message
721: NewAlice = connect_same(Alice, presence),
722: %% now we can resume c2s process of the old connection
723: %% and let it process session resumption timeout
724: ok = rpc(mim(), sys, resume, [C2SPid]),
725: Stanzas = escalus:wait_for_stanzas(NewAlice, 3),
726:
727: % what about order ?
728: % alice receive presence from herself and 3 unacked messages from bob
729: escalus_new_assert:mix_match([is_chat(<<"msg-1">>),
730: is_chat(<<"msg-2">>),
731: is_chat(<<"msg-3">>)],
732: Stanzas),
733: escalus_connection:stop(Bob),
734: escalus_connection:stop(NewAlice).
735:
736: %%for instance it can be done by mod ping
737: resume_session_state_stop_c2s(Config) ->
738: Bob = connect_fresh(Config, bob, presence),
739: Alice = connect_fresh(Config, alice, sr_presence, manual),
740:
741: get_ack(Alice),
742: ack_initial_presence(Alice),
743:
744: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"msg-1">>)),
745: escalus:assert(is_chat_message, [<<"msg-1">>], escalus_connection:get_stanza(Alice, msg)),
746:
747: %% get pid of c2s
748: C2SPid = mongoose_helper:get_session_pid(Alice),
749: %% Wait c2s process to process our presence ack.
750: %% Otherwise, we can receive two initial presences sometimes.
751: sm_helper:wait_for_c2s_unacked_count(C2SPid, 1),
752:
753: % kill alice connection
754: escalus_connection:kill(Alice),
755: % session should be alive
756: sm_helper:assert_alive_resources(Alice, 1),
757: rpc(mim(), mongoose_c2s, stop, [C2SPid, normal]),
758: sm_helper:wait_until_resume_session(C2SPid),
759: %% suspend the process to ensure that Alice has enough time to reconnect,
760: %% before resumption timeout occurs.
761: ok = rpc(mim(), sys, suspend, [C2SPid]),
762:
763: %% alice comes back and receives unacked message
764: NewAlice = connect_same(Alice, presence, manual),
765: %% now we can resume c2s process of the old connection
766: %% and let it process session resumption timeout
767: ok = rpc(mim(), sys, resume, [C2SPid]),
768:
769: escalus:assert(is_chat_message, [<<"msg-1">>], escalus_connection:get_stanza(NewAlice, msg)),
770: escalus_connection:stop(Bob),
771: escalus_connection:stop(NewAlice).
772:
773: %% This test only verifies the validity of helpers (get_session_pid,
774: %% get_c2s_state_name) written for wait_for_resumption
775: %% testcase.
776: session_established(Config) ->
777: Alice = connect_fresh(Config, alice, presence),
778: C2SPid = mongoose_helper:get_session_pid(Alice),
779: session_established = mongoose_helper:get_c2s_state_name(C2SPid),
780: escalus_connection:stop(Alice).
781:
782: %% Ensure that after a violent disconnection,
783: %% the c2s waits for resumption (but don't resume yet).
784: wait_for_resumption(Config) ->
785: AliceSpec = escalus_fresh:create_fresh_user(Config, alice),
786: Bob = connect_fresh(Config, bob, session),
787: Texts = three_texts(),
788: {C2SPid, _} = buffer_unacked_messages_and_die(Config, AliceSpec, Bob, Texts),
789: sm_helper:wait_until_resume_session(C2SPid).
790:
791: unacknowledged_message_hook_filter(Config) ->
792: FilterText = <<"filter">>,
793: Bob = connect_fresh(Config, bob, presence),
794: AliceSpec = escalus_fresh:create_fresh_user(Config, alice),
795: Resource = proplists:get_value(username, AliceSpec),
796: HookHandlerExtra = start_filter_hook_listener(FilterText, Resource),
797: Alice = connect_spec([{resource, Resource} | AliceSpec], sr_presence, manual),
798: %% Ack the presence stanza
799: get_ack(Alice),
800: ack_initial_presence(Alice),
801: Messages = [<<"msg-1">>, <<"msg-2">>, <<"msg-3">>, <<"msg-4">>],
802: All = [<<"msg-1">>, FilterText, <<"msg-2">>, FilterText, <<"msg-3">>, <<"msg-4">>],
803: [ escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, Body)) || Body <- All ],
804: %% kill alice connection
805: C2SPid = mongoose_helper:get_session_pid(Alice),
806: escalus_connection:kill(Alice),
807: sm_helper:wait_until_resume_session(C2SPid),
808: sm_helper:assert_alive_resources(Alice, 1),
809: %% ensure second C2S is registered so all the messages are bounced properly
810: NewAlice = connect_spec([{resource, <<"2">>}| AliceSpec], sr_presence, manual),
811: send_initial_presence(NewAlice),
812: sm_helper:wait_for_resource_count(NewAlice, 2),
813: ok = rpc(mim(), sys, terminate, [C2SPid, normal]),
814: %% verify that the filtered message is never received
815: verify_no_receive_filtertext(NewAlice, FilterText, Messages),
816: stop_hook_listener(HookHandlerExtra),
817: escalus_connection:stop(Bob).
818:
819: verify_no_receive_filtertext(_, _, []) ->
820: ok;
821: verify_no_receive_filtertext(Client, FilterText, Messages) ->
822: case escalus:wait_for_stanzas(Client, 1, 500) of
823: [] -> ct:fail("Messages pending: ~p~n", [Messages]);
824: [St] ->
825: case exml_query:path(St, [{element, <<"body">>}, cdata]) of
826: undefined ->
827: verify_no_receive_filtertext(Client, FilterText, Messages);
828: FilterText ->
829: ct:fail("Composing forwarded from a different c2s process");
830: Message ->
831: Pred = fun(Msg) -> Msg =/= Message end,
832: {Pending, _} = lists:partition(Pred, Messages),
833: verify_no_receive_filtertext(Client, FilterText, Pending)
834: end
835: end.
836:
837: unacknowledged_message_hook_resume(Config) ->
838: unacknowledged_message_hook_common(fun unacknowledged_message_hook_resume/4, Config).
839:
840: unacknowledged_message_hook_resume(AliceSpec, Resource, SMID, _C2SPid) ->
841: NewAlice = connect_spec(AliceSpec, {resume, SMID, 1}, manual),
842: send_initial_presence(NewAlice),
843: {Resource, NewAlice}.
844:
845: unacknowledged_message_hook_bounce(Config) ->
846: unacknowledged_message_hook_common(fun unacknowledged_message_hook_bounce/4, Config).
847:
848: unacknowledged_message_hook_bounce(AliceSpec, Resource, _SMID, C2SPid) ->
849: NewResource = <<"new_", Resource/binary>>,
850: NewSpec = lists:keystore(resource, 1, AliceSpec, {resource, NewResource}),
851: NewAlice = connect_spec(NewSpec, sr_session, manual),
852: send_initial_presence(NewAlice),
853: %% ensure second C2S is registered so all the messages are bounced properly
854: sm_helper:wait_for_resource_count(NewAlice, 2),
855: ok = rpc(mim(), sys, terminate, [C2SPid, normal]),
856: {NewResource, NewAlice}.
857:
858: unacknowledged_message_hook_offline(Config) ->
859: unacknowledged_message_hook_common(fun unacknowledged_message_hook_offline/4, Config).
860:
861: unacknowledged_message_hook_offline(AliceSpec, Resource, _SMID, C2SPid) ->
862: C2SRef = erlang:monitor(process, C2SPid),
863: sm_helper:wait_for_process_termination(C2SRef),
864: %% reset the session, so old C2S process is stopped
865: NewAlice = connect_spec(AliceSpec, sr_session, manual),
866: %% wait for old C2S termination before send presence. other way
867: %% some of the latest unacknowledged messages can be bounced to
868: %% the new C2S process instead of going to the mod_offline storage.
869: %% looks like all the unacknowledged messages arrive to the new
870: %% C2S, but the message sequence is broken (the bounced messages
871: %% delivered before the messages from the mod_offline storage)
872: send_initial_presence(NewAlice),
873: {Resource, NewAlice}.
874:
875: unacknowledged_message_hook_common(RestartConnectionFN, Config) ->
876: %% connect bob and alice
877: Bob = connect_fresh(Config, bob, presence),
878:
879: AliceSpec0 = escalus_fresh:create_fresh_user(Config, alice),
880: Resource = proplists:get_value(username, AliceSpec0),
881: AliceSpec = [{resource, Resource} | AliceSpec0],
882: HookHandlerExtra = start_hook_listener(Resource),
883: Alice = connect_spec(AliceSpec, sr_presence, manual),
884: %% Ack the presence stanza
885: get_ack(Alice),
886: ack_initial_presence(Alice),
887:
888: SMID = sm_helper:client_to_smid(Alice),
889:
890: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"msg-1">>)),
891: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"msg-2">>)),
892: %% kill alice connection
893: C2SPid = mongoose_helper:get_session_pid(Alice),
894: escalus_connection:kill(Alice),
895: sm_helper:wait_until_resume_session(C2SPid),
896: sm_helper:assert_alive_resources(Alice, 1),
897:
898: escalus:assert(is_chat_message, [<<"msg-1">>], wait_for_unacked_msg_hook(0, Resource, 100)),
899: escalus:assert(is_chat_message, [<<"msg-2">>], wait_for_unacked_msg_hook(0, Resource, 100)),
900: ?assertEqual(timeout, wait_for_unacked_msg_hook(0, Resource, 100)),
901:
902: %% send some messages and check if c2s can handle it
903: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"msg-3">>)),
904: escalus_connection:send(Bob, escalus_stanza:chat_to_short_jid(Alice, <<"msg-4">>)),
905: escalus:assert(is_chat_message, [<<"msg-3">>], wait_for_unacked_msg_hook(0, Resource, 100)),
906: escalus:assert(is_chat_message, [<<"msg-4">>], wait_for_unacked_msg_hook(0, Resource, 100)),
907: ?assertEqual(timeout, wait_for_unacked_msg_hook(0, Resource, 100)),
908:
909: %% alice comes back and receives unacked message
910: {NewResource, NewAlice} = RestartConnectionFN(AliceSpec, Resource, SMID, C2SPid),
911:
912: mongoose_helper:wait_until(
913: fun() ->
914: Stanza = escalus_connection:get_stanza(NewAlice, msg),
915: escalus:assert(is_chat_message, [<<"msg-4">>], Stanza),
916: ok
917: end, ok),
918:
919: NewC2SPid = mongoose_helper:get_session_pid(NewAlice),
920: escalus_connection:kill(NewAlice),
921: sm_helper:wait_until_resume_session(NewC2SPid),
922:
923: escalus:assert(is_chat_message, [<<"msg-1">>], wait_for_unacked_msg_hook(1, NewResource, 100)),
924: escalus:assert(is_chat_message, [<<"msg-2">>], wait_for_unacked_msg_hook(1, NewResource, 100)),
925: escalus:assert(is_chat_message, [<<"msg-3">>], wait_for_unacked_msg_hook(1, NewResource, 100)),
926: escalus:assert(is_chat_message, [<<"msg-4">>], wait_for_unacked_msg_hook(1, NewResource, 100)),
927: ?assertEqual(timeout, wait_for_unacked_msg_hook(0, Resource, 100)),
928: stop_hook_listener(HookHandlerExtra),
929: escalus_connection:stop(Bob).
930:
931: resume_session(Config) ->
932: AliceSpec = escalus_fresh:create_fresh_user(Config, alice),
933: Texts = three_texts(),
934: escalus:fresh_story(Config, [{bob, 1}], fun(Bob) ->
935: {_, SMID} = buffer_unacked_messages_and_die(Config, AliceSpec, Bob, Texts),
936: %% Resume the session.
937: Alice = connect_spec(AliceSpec, {resume, SMID, 1}, manual),
938: %% Alice receives the unacked messages from the previous
939: %% interrupted session.
940: sm_helper:wait_for_messages(Alice, Texts),
941: %% Alice acks the received messages.
942: escalus_connection:send(Alice, escalus_stanza:sm_ack(5)),
943: escalus_connection:stop(Alice)
944: end).
945:
946: resume_session_with_wrong_h_does_not_leak_sessions(Config) ->
947: AliceSpec = escalus_fresh:create_fresh_user(Config, alice),
948: Messages = three_texts(),
949: HostType = host_type(),
950: escalus:fresh_story(Config, [{bob, 1}], fun(Bob) ->
951: {_, SMID} = buffer_unacked_messages_and_die(Config, AliceSpec, Bob, Messages),
952: %% Resume the session.
953: Alice = connect_spec(AliceSpec, auth, manual),
954: Resumed = sm_helper:try_to_resume_stream(Alice, SMID, 30),
955: escalus:assert(is_stream_error, [<<"undefined-condition">>, <<>>], Resumed),
956: escalus_connection:wait_for_close(Alice, timer:seconds(5)),
957: Fun = fun() ->
958: [] = sm_helper:get_user_present_resources(Alice),
959: sm_helper:get_sid_by_stream_id(HostType, SMID)
960: end,
961: mongoose_helper:wait_until(Fun, {error, smid_not_found}, #{name => smid_is_cleaned})
962: end).
963:
964: resume_session_with_wrong_sid_returns_item_not_found(Config) ->
965: session_resumption_expects_item_not_found(Config, <<"wrong-sid">>).
966:
967: resume_session_with_wrong_namespace_is_a_noop(Config) ->
968: Alice = connect_fresh(Config, alice, auth),
969: #xmlel{attrs = Attrs} = Resume = escalus_stanza:resume(<<"doesnt_matter">>, 4),
970: Attrs2 = lists:keyreplace(<<"xmlns">>, 1, Attrs, {<<"xmlns">>, <<"not-stream-mgnt">>}),
971: escalus_connection:send(Alice, Resume#xmlel{attrs = Attrs2}),
972: escalus_assert:has_no_stanzas(Alice),
973: [] = sm_helper:get_user_present_resources(Alice),
974: true = escalus_connection:is_connected(Alice),
975: escalus_connection:stop(Alice).
976:
977: resume_dead_session_results_in_item_not_found(Config) ->
978: SMID = base64:encode(crypto:strong_rand_bytes(21)),
979: SID = {os:timestamp(), undefined},
980: HostType = host_type(),
981: rpc(mim(), ?MOD_SM, register_smid, [HostType, SMID, SID]),
982: session_resumption_expects_item_not_found(Config, SMID).
983:
984: session_resumption_expects_item_not_found(Config, SMID) ->
985: Alice = connect_fresh(Config, alice, auth),
986: Resumed = sm_helper:try_to_resume_stream(Alice, SMID, 2),
987: escalus:assert(is_sm_failed, [<<"item-not-found">>], Resumed),
988: [] = sm_helper:get_user_present_resources(Alice),
989: true = escalus_connection:is_connected(Alice),
990: escalus_connection:stop(Alice).
991:
992: resume_session_kills_old_C2S_gracefully(Config) ->
993: Alice = connect_fresh(Config, alice, sr_presence, manual),
994: C2SPid = mongoose_helper:get_session_pid(Alice),
995:
996: %% Monitor the C2S process and disconnect Alice.
997: MonitorRef = sm_helper:monitor_session(Alice),
998: escalus_client:kill_connection(Config, Alice),
999:
1000: %% Ensure the c2s process is waiting for resumption.
1001: sm_helper:wait_until_resume_session(C2SPid),
1002:
1003: %% Resume the session.
1004: NewAlice = connect_resume(Alice, 1),
1005:
1006: %% C2S process should die gracefully with Reason=normal.
1007: sm_helper:wait_for_process_termination(MonitorRef),
1008: escalus_connection:stop(NewAlice).
1009:
1010: carboncopy_works(Config) ->
1011: escalus:fresh_story(Config, [{alice, 2}, {bob, 1}], fun(Alice1, Alice, Bob) ->
1012: mongoose_helper:enable_carbons([Alice1, Alice]),
1013: escalus_connection:send(Bob, escalus_stanza:chat_to(Alice1, <<"msg-4">>)),
1014: sm_helper:wait_for_messages(Alice1, [<<"msg-4">>]),
1015: carboncopy_helper:wait_for_carbon_chat_with_body(Alice, <<"msg-4">>, #{from => Bob, to => Alice1})
1016: end).
1017:
1018: carboncopy_works_after_resume(Config) ->
1019: Texts = three_texts(),
1020: escalus:fresh_story(Config, [{alice, 1}, {bob, 1}], fun(Alice1, Bob) ->
1021: AliceSpec = [{resource, <<"res2">>} | sm_helper:client_to_spec(Alice1)],
1022: F = fun(Alice2) ->
1023: [escalus:assert(is_presence_with_type, [<<"available">>], escalus:wait_for_stanza(A)) || A <- [Alice1, Alice2]],
1024: mongoose_helper:enable_carbons([Alice1, Alice2])
1025: end,
1026: {_, SMID} = buffer_unacked_messages_and_die(Config, AliceSpec, Bob, Texts, F),
1027: %% Resume the session.
1028: Alice = connect_spec(AliceSpec, {resume, SMID, 1}, manual),
1029: wait_for_carbon_with_bodies(Alice1, Texts, #{from => Bob, to => Alice}),
1030: %% Get a presence from Alice1 again
1031: escalus:assert(is_presence_with_type, [<<"available">>], escalus:wait_for_stanza(Alice)),
1032: %% Alice receives an IQ result from the carbon copy enable request
1033: escalus:assert(is_iq_result, [], escalus:wait_for_stanza(Alice)),
1034: %% Alice receives the unacked messages from the previous
1035: %% interrupted session.
1036: sm_helper:wait_for_messages(Alice, Texts),
1037: %% Alice acks the received messages.
1038: escalus_connection:send(Alice, escalus_stanza:sm_ack(5)),
1039: %% Direct send
1040: escalus_connection:send(Bob, escalus_stanza:chat_to(Alice1, <<"msg-4">>)),
1041: sm_helper:wait_for_messages(Alice1, [<<"msg-4">>]),
1042: carboncopy_helper:wait_for_carbon_chat_with_body(Alice, <<"msg-4">>, #{from => Bob, to => Alice1}),
1043: escalus_connection:stop(Alice)
1044: end).
1045:
1046: wait_for_carbon_with_bodies(Client, Texts, Params) ->
1047: [carboncopy_helper:wait_for_carbon_chat_with_body(Client, Text, Params) || Text <- Texts].
1048:
1049: buffer_unacked_messages_and_die(Config, AliceSpec, Bob, Texts) ->
1050: F = fun(_Client) -> ok end,
1051: buffer_unacked_messages_and_die(Config, AliceSpec, Bob, Texts, F).
1052:
1053: buffer_unacked_messages_and_die(Config, AliceSpec, Bob, Texts, F) ->
1054: Alice = connect_spec(AliceSpec, sr_presence, manual),
1055: F(Alice),
1056: C2SPid = mongoose_helper:get_session_pid(Alice),
1057: %% Bobs sends some messages to Alice.
1058: sm_helper:send_messages(Bob, Alice, Texts),
1059: %% Alice receives them, but doesn't ack.
1060: sm_helper:wait_for_messages(Alice, Texts),
1061: %% Alice's connection is violently terminated.
1062: escalus_client:kill_connection(Config, Alice),
1063: sm_helper:wait_until_resume_session(C2SPid),
1064: SMID = sm_helper:client_to_smid(Alice),
1065: {C2SPid, SMID}.
1066:
1067: aggressively_pipelined_resume(Config) ->
1068: AliceSpec = [{manual_ack, true}, {parser_opts, [{start_tag, <<"stream:stream">>}]}
1069: | escalus_fresh:create_fresh_user(Config, alice)],
1070: UnackedMessages = three_texts(),
1071: escalus:fresh_story(Config, [{bob, 1}], fun(Bob) ->
1072: {_, SMID} = buffer_unacked_messages_and_die(Config, AliceSpec, Bob, UnackedMessages),
1073: %% Resume the session.
1074: Alice = escalus_connection:connect(AliceSpec),
1075:
1076: Username = proplists:get_value(username, AliceSpec),
1077: Password = proplists:get_value(password, AliceSpec),
1078: Payload = <<0:8,Username/binary,0:8,Password/binary>>,
1079: Server = proplists:get_value(server, AliceSpec),
1080:
1081: Stream = escalus_stanza:stream_start(Server, <<"jabber:client">>),
1082: Auth = escalus_stanza:auth(<<"PLAIN">>, [#xmlcdata{content = base64:encode(Payload)}]),
1083: AuthStream = escalus_stanza:stream_start(Server, <<"jabber:client">>),
1084: Resume = escalus_stanza:resume(SMID, 2),
1085:
1086: escalus_client:send(Alice, [Stream, Auth, AuthStream, Resume]),
1087: Messages = [escalus_connection:get_stanza(Alice, {get_resumed, I}) || I <- lists:seq(1, 6)],
1088: escalus:assert(is_sm_resumed, [SMID], lists:last(Messages)),
1089:
1090: escalus_connection:stop(Alice)
1091: end).
1092:
1093: %% This is a regression test for a case when a session processes a request, which will
1094: %% receive a response from the server, i.e. will have the same origin SID in mongoose_acc.
1095: %% Without proper handling, the reply would be rejected because the resumed session
1096: %% has new SID.
1097: replies_are_processed_by_resumed_session(Config) ->
1098: %% GIVEN a session and registered special IQ handler (added in init_per_testcase),
1099: %% that waits for old session process to terminate (at this point new process
1100: %% has fully taken over) and then actually sends the reply.
1101: Alice = connect_fresh(Config, alice, sr_presence),
1102:
1103: %% WHEN a client sends IQ request to the special handler...
1104: IQReq = escalus_stanza:iq_get(regression_ns(), []),
1105: escalus:send(Alice, IQReq),
1106:
1107: %% ... goes down and session is resumed.
1108: Alice2 = sm_helper:kill_and_connect_resume(Alice),
1109:
1110: %% THEN the client receives the reply properly.
1111: IQReply = escalus:wait_for_stanza(Alice2),
1112: escalus:assert(is_iq_result, [IQReq], IQReply),
1113: escalus_connection:stop(Alice2).
1114:
1115: %% This is a regression test for a bug, which manifested in following scenario
1116: %% (due to improper presence sub requests buffering):
1117: %% 1. Bob is online, Alice is offline
1118: %% 2. Bob subscribes to Alice's presence;
1119: %% 3. Alice becomes online
1120: %% 4. Bob sends a message to Alice
1121: %% 5. Alice doesn't SM-ack the request or message, terminates the connection
1122: %% 6. Alice reconnects but with session *replace*, not resume
1123: %% 7. Packet rerouting crashes on the buffered sub request, preventing resending whole buffer
1124: %% 8. Alice doesn't receive the buffered message
1125: subscription_requests_are_buffered_properly(Config) ->
1126: AliceSpec = escalus_fresh:create_fresh_user(Config, alice),
1127: MsgBody = <<"buffered">>,
1128: escalus:fresh_story(Config, [{bob, 1}], fun(Bob) ->
1129: % GIVEN Bob's pending subscription to Alice's presence
1130: AliceJid = common_helper:get_bjid(AliceSpec),
1131: escalus:send(Bob, escalus_stanza:presence_direct(AliceJid, <<"subscribe">>)),
1132: _RosterPushReq = escalus:wait_for_stanza(Bob),
1133:
1134: % WHEN Alice becomes online...
1135: Alice = connect_spec(AliceSpec, sr_session, manual),
1136: send_initial_presence(Alice),
1137: %% subscribe could come before the initial presence
1138: escalus:assert_many([is_presence(<<"available">>), is_presence(<<"subscribe">>)],
1139: escalus:wait_for_stanzas(Alice, 2)),
1140:
1141: % ...and Bob sends a message to Alice...
1142: escalus:send(Bob, escalus_stanza:chat_to(Alice, MsgBody)),
1143: MsgStanza = escalus:wait_for_stanza(Alice),
1144: escalus:assert(is_chat_message, [MsgBody], MsgStanza),
1145:
1146: % ...and Alice terminates connection without acking anything...
1147: escalus_client:kill_connection(Config, Alice),
1148:
1149: % ...and reconnects with session replacement.
1150: Alice2 = connect_spec(AliceSpec, session, manual),
1151:
1152: % THEN Alice receives (without sending initial presence):
1153: % * buffered Bob's message (like above)
1154: % Alice DOESN'T receive:
1155: % * buffered subscription request because it is dropped by ejabberd_sm
1156: % because it's treated like repeated sub request to bare JID, so it's not
1157: % processed by any sub req handler (like mod_roster)
1158: % * buffered available presence from Alice - because it is addressed to another SID
1159: % and Alice2 is a brand new session
1160: escalus:assert(is_chat_message, [MsgBody], escalus:wait_for_stanza(Alice2)),
1161: sm_helper:send_and_receive(Bob, Alice2, <<"flush1">>),
1162: escalus_assert:has_no_stanzas(Alice2),
1163:
1164: %% Only once an initial presence is sent, a subscription request is sent
1165: send_initial_presence(Alice2),
1166: escalus:assert_many([is_presence(<<"available">>), is_presence(<<"subscribe">>)],
1167: escalus:wait_for_stanzas(Alice2, 2)),
1168:
1169: sm_helper:send_and_receive(Bob, Alice2, <<"flush2">>),
1170: escalus_assert:has_no_stanzas(Alice2),
1171:
1172: escalus_connection:stop(Alice2)
1173: end).
1174:
1175: %% This is a regression test for a bug, due to which messages sent to old session
1176: %% in a middle of state handover were not appended properly to SM buffer.
1177: %% Scenario to reproduce:
1178: %% 1. Online Bob and Alice
1179: %% 2. Alice kills the connection
1180: %% 3. Alice's session is suspended
1181: %% 4. Alice resumes session with new connection. At this moment new session is still not
1182: %% present in session table. `resume` request is stuck in old proc mailbox.
1183: %% 5. Bob sends a message to Alice. Only old proc is present in session table so now
1184: %% old session has two messages in mailbox: `resume` and XML from Bob
1185: %% 6. We resume old process and it begins session handover
1186: %% 7. Bob's message is appended to SM buffer in "flush" step
1187: %% 8. With bug fixed, the message is retransmitted properly
1188: messages_are_properly_flushed_during_resumption(Config) ->
1189: escalus:fresh_story(Config, [{bob, 1}], fun(Bob) ->
1190: % GIVEN (online Bob) and (Alice in resume state); Alice's session is suspended
1191: Alice = connect_fresh(Config, alice, sr_presence),
1192: SMH = escalus_connection:get_sm_h(Alice),
1193: escalus_client:kill_connection(Config, Alice),
1194: %% The receiver process would stop now
1195: C2SPid = mongoose_helper:get_session_pid(Alice),
1196: sm_helper:wait_until_resume_session(C2SPid),
1197:
1198: sm_helper:wait_for_queue_length(C2SPid, 0),
1199: ok = rpc(mim(), sys, suspend, [C2SPid]),
1200:
1201: % WHEN new session requests resumption
1202: % we wait until that old session has resumption request enqueued;
1203: % we need it to ensure the order of messages: resume first, Bob's chat second.
1204: % Actual wait and message sent by Bob is done in separate process
1205: % because new client start will block until old process is resumed
1206:
1207: MsgBody = <<"flush-regression">>,
1208: spawn_link(fun() ->
1209: sm_helper:wait_for_queue_length(C2SPid, 1),
1210:
1211: % Bob sends a message...
1212: escalus:send(Bob, escalus_stanza:chat_to(Alice, MsgBody)),
1213:
1214: % ...we ensure that a message is enqueued in Alice's session...
1215: % (2 messages = resume request + Bob's message)
1216: sm_helper:wait_for_queue_length(C2SPid, 2),
1217:
1218: % ...and old process is resumed.
1219: ok = rpc(mim(), sys, resume, [C2SPid])
1220: end),
1221: Alice2 = connect_resume(Alice, SMH),
1222: % THEN Alice's new session receives Bob's message
1223: RecvMsg = escalus:wait_for_stanza(Alice2),
1224: escalus:assert(is_chat_message, [MsgBody], RecvMsg)
1225: end).
1226:
1227: no_crash_if_stream_mgmt_disabled_but_client_requests_stream_mgmt(Config) ->
1228: Alice = connect_fresh(Config, alice, session, manual),
1229: %% Should not crash anything!
1230: escalus_connection:send(Alice, escalus_stanza:enable_sm()),
1231: Response = escalus_connection:get_stanza(Alice, service_unavailable),
1232: escalus:assert(is_sm_failed, [<<"feature-not-implemented">>], Response),
1233: escalus_connection:stop(Alice).
1234:
1235: no_crash_if_stream_mgmt_disabled_but_client_requests_stream_mgmt_with_resumption(Config) ->
1236: Alice = connect_fresh(Config, alice, session, manual),
1237: %% Should not crash anything!
1238: escalus_connection:send(Alice, escalus_stanza:enable_sm([resume])),
1239: Response = escalus_connection:get_stanza(Alice, service_unavailable),
1240: escalus:assert(is_sm_failed, [<<"feature-not-implemented">>], Response),
1241: escalus_connection:stop(Alice).
1242:
1243: %%--------------------------------------------------------------------
1244: %% Helpers
1245: %%--------------------------------------------------------------------
1246: start_hook_listener(Resource) ->
1247: TestCasePid = self(),
1248: rpc(mim(), ?MODULE, rpc_start_hook_handler, [TestCasePid, Resource, host_type()]).
1249:
1250: stop_hook_listener(HookExtra) ->
1251: rpc(mim(), ?MODULE, rpc_stop_hook_handler, [HookExtra, host_type()]).
1252:
1253: rpc_start_hook_handler(TestCasePid, User, HostType) ->
1254: LUser = jid:nodeprep(User),
1255: Extra = #{luser => LUser, pid => TestCasePid},
1256: gen_hook:add_handler(unacknowledged_message, HostType,
1257: fun ?MODULE:hook_handler_fn/3,
1258: Extra, 50),
1259: Extra.
1260:
1261: rpc_stop_hook_handler(HookExtra, HostType) ->
1262: gen_hook:delete_handler(unacknowledged_message, HostType,
1263: fun ?MODULE:hook_handler_fn/3,
1264: HookExtra, 50).
1265:
1266: hook_handler_fn(Acc,
1267: #{jid := Jid} = _Params,
1268: #{luser := LUser, pid := TestCasePid} = _Extra) ->
1269: {U, _S, R} = jid:to_lower(Jid),
1270: case U of
1271: LUser ->
1272: Counter = mongoose_acc:get(sm_test, counter, 0, Acc),
1273: El = mongoose_acc:element(Acc),
1274: TestCasePid ! {sm_test, Counter, R, El},
1275: {ok, mongoose_acc:set_permanent(sm_test, counter, Counter + 1, Acc)};
1276: _ -> {ok, Acc}
1277: end.
1278:
1279: wait_for_unacked_msg_hook(Counter, Res, Timeout) ->
1280: receive
1281: {sm_test, AccCounter, Resource, Stanza} = Msg ->
1282: ?assertEqual(Counter, AccCounter, Msg),
1283: ?assertEqual(Res, Resource, Msg),
1284: Stanza
1285: after Timeout ->
1286: timeout
1287: end.
1288:
1289: start_filter_hook_listener(FilterText, Resource) ->
1290: rpc(mim(), ?MODULE, rpc_start_filter_hook_handler, [FilterText, Resource, host_type()]).
1291:
1292: stop_filter_hook_listener(HookExtra) ->
1293: rpc(mim(), ?MODULE, rpc_stop_filter_hook_handler, [HookExtra, host_type()]).
1294:
1295: rpc_start_filter_hook_handler(FilterText, User, HostType) ->
1296: LUser = jid:nodeprep(User),
1297: Extra = #{luser => LUser, filter_text => FilterText},
1298: gen_hook:add_handler(filter_unacknowledged_messages, HostType,
1299: fun ?MODULE:filter_hook_handler_fn/3,
1300: Extra, 50),
1301: Extra.
1302:
1303: rpc_stop_filter_hook_handler(HookExtra, HostType) ->
1304: gen_hook:delete_handler(filter_unacknowledged_messages, HostType,
1305: fun ?MODULE:filter_hook_handler_fn/3,
1306: HookExtra, 50).
1307:
1308: filter_hook_handler_fn(Buffer,
1309: #{jid := Jid} = _Params,
1310: #{luser := LUser, filter_text := FilterText} = _Extra) ->
1311: {U, _} = jid:to_lus(Jid),
1312: case U of
1313: LUser ->
1314: F = fun(Acc) -> filter_text(Acc, FilterText) end,
1315: NewBuffer = lists:filter(F, Buffer),
1316: {ok, NewBuffer};
1317: _ -> {ok, Buffer}
1318: end.
1319:
1320: filter_text(Acc, FilterText) ->
1321: case mongoose_acc:stanza_name(Acc) of
1322: <<"message">> ->
1323: El = mongoose_acc:element(Acc),
1324: FilterText =/= exml_query:path(El, [{element, <<"body">>}, cdata]);
1325: _ ->
1326: true
1327: end.
1328:
1329: is_chat(Content) ->
1330: fun(Stanza) -> escalus_pred:is_chat_message(Content, Stanza) end.
1331:
1332: is_presence(Type) ->
1333: fun(Stanza) -> escalus_pred:is_presence_with_type(Type, Stanza) end.
1334:
1335: three_texts() ->
1336: [<<"msg-1">>, <<"msg-2">>, <<"msg-3">>].
1337:
1338: get_stanzas_filtered_by_mod_ping() ->
1339: History = rpc(mim(), meck, history, [mod_ping]),
1340: [Stanza ||
1341: {_Pid,
1342: {_Mod,
1343: filter_local_packet = _Func,
1344: [{_, _, _, Stanza} = _Acc, _Params, _Extra] = _Args
1345: },
1346: {stop, drop} = _Result
1347: } <- History
1348: ].
1349: %%--------------------------------------------------------------------
1350: %% IQ handler necessary for reproducing "replies_are_processed_by_resumed_session"
1351: %%--------------------------------------------------------------------
1352:
1353: regression_ns() ->
1354: <<"regression">>.
1355:
1356: register_handler() ->
1357: HostType = host_type(),
1358: rpc(mim(), gen_iq_handler, add_iq_handler_for_domain,
1359: [HostType, regression_ns(), ejabberd_sm,
1360: fun ?MODULE:regression_handler/5, #{}, one_queue]).
1361:
1362: unregister_handler() ->
1363: HostType = host_type(),
1364: rpc(mim(), gen_iq_handler, remove_iq_handler_for_domain,
1365: [HostType, regression_ns(), ejabberd_sm]).
1366:
1367: regression_handler(Acc, _From, _To, IQ, _Extra) ->
1368: %% A bit of a hack - will no longer work when the SID format changes
1369: {_, Pid} = mongoose_acc:get(c2s, origin_sid, undefined, Acc),
1370: erlang:monitor(process, Pid),
1371: receive
1372: {'DOWN', _, _, _, _} ->
1373: ok
1374: after
1375: 10000 ->
1376: error({c2s_not_stopped_after_timeout, Pid})
1377: end,
1378: %% We avoid another race condition - there is a short window where user session
1379: %% is not registered in ejabberd_sm: between old process termination and the moment
1380: %% when the new process stores new session in memory. It should be fixed separately.
1381: wait_for_session(mongoose_acc:get(c2s, origin_jid, undefined, Acc), 50, 100),
1382: {Acc, jlib:make_result_iq_reply(IQ)}.
1383:
1384: wait_for_session(JID, Retries, SleepTime) ->
1385: case ejabberd_sm:get_session(JID) of
1386: offline ->
1387: timer:sleep(SleepTime),
1388: wait_for_session(JID, Retries - 1, SleepTime);
1389: _ ->
1390: ok
1391: end.
1392:
1393: maybe_ack_initial_presence(Alice, ack) ->
1394: ack_initial_presence(Alice);
1395: maybe_ack_initial_presence(_Alice, no_ack) ->
1396: ok.