1 /*
2 * Copyright (c) 2021 The WebRTC project authors. All Rights Reserved.
3 *
4 * Use of this source code is governed by a BSD-style license
5 * that can be found in the LICENSE file in the root of the source
6 * tree. An additional intellectual property rights grant can be found
7 * in the file PATENTS. All contributing project authors may
8 * be found in the AUTHORS file in the root of the source tree.
9 */
10 #include "net/dcsctp/socket/stream_reset_handler.h"
11
12 #include <array>
13 #include <cstdint>
14 #include <memory>
15 #include <type_traits>
16 #include <vector>
17
18 #include "absl/types/optional.h"
19 #include "api/array_view.h"
20 #include "api/task_queue/task_queue_base.h"
21 #include "net/dcsctp/common/handover_testing.h"
22 #include "net/dcsctp/common/internal_types.h"
23 #include "net/dcsctp/packet/chunk/reconfig_chunk.h"
24 #include "net/dcsctp/packet/parameter/incoming_ssn_reset_request_parameter.h"
25 #include "net/dcsctp/packet/parameter/outgoing_ssn_reset_request_parameter.h"
26 #include "net/dcsctp/packet/parameter/parameter.h"
27 #include "net/dcsctp/packet/parameter/reconfiguration_response_parameter.h"
28 #include "net/dcsctp/public/dcsctp_message.h"
29 #include "net/dcsctp/rx/data_tracker.h"
30 #include "net/dcsctp/rx/reassembly_queue.h"
31 #include "net/dcsctp/socket/mock_context.h"
32 #include "net/dcsctp/socket/mock_dcsctp_socket_callbacks.h"
33 #include "net/dcsctp/testing/data_generator.h"
34 #include "net/dcsctp/testing/testing_macros.h"
35 #include "net/dcsctp/timer/timer.h"
36 #include "net/dcsctp/tx/mock_send_queue.h"
37 #include "net/dcsctp/tx/retransmission_queue.h"
38 #include "rtc_base/gunit.h"
39 #include "test/gmock.h"
40
41 namespace dcsctp {
42 namespace {
43 using ::testing::IsEmpty;
44 using ::testing::NiceMock;
45 using ::testing::Return;
46 using ::testing::SizeIs;
47 using ::testing::UnorderedElementsAre;
48 using ResponseResult = ReconfigurationResponseParameter::Result;
49
50 constexpr TSN kMyInitialTsn = MockContext::MyInitialTsn();
51 constexpr ReconfigRequestSN kMyInitialReqSn = ReconfigRequestSN(*kMyInitialTsn);
52 constexpr TSN kPeerInitialTsn = MockContext::PeerInitialTsn();
53 constexpr ReconfigRequestSN kPeerInitialReqSn =
54 ReconfigRequestSN(*kPeerInitialTsn);
55 constexpr uint32_t kArwnd = 131072;
56 constexpr DurationMs kRto = DurationMs(250);
57
58 constexpr std::array<uint8_t, 4> kShortPayload = {1, 2, 3, 4};
59
60 MATCHER_P3(SctpMessageIs, stream_id, ppid, expected_payload, "") {
61 if (arg.stream_id() != stream_id) {
62 *result_listener << "the stream_id is " << *arg.stream_id();
63 return false;
64 }
65
66 if (arg.ppid() != ppid) {
67 *result_listener << "the ppid is " << *arg.ppid();
68 return false;
69 }
70
71 if (std::vector<uint8_t>(arg.payload().begin(), arg.payload().end()) !=
72 std::vector<uint8_t>(expected_payload.begin(), expected_payload.end())) {
73 *result_listener << "the payload is wrong";
74 return false;
75 }
76 return true;
77 }
78
AddTo(TSN tsn,int delta)79 TSN AddTo(TSN tsn, int delta) {
80 return TSN(*tsn + delta);
81 }
82
AddTo(ReconfigRequestSN req_sn,int delta)83 ReconfigRequestSN AddTo(ReconfigRequestSN req_sn, int delta) {
84 return ReconfigRequestSN(*req_sn + delta);
85 }
86
87 class StreamResetHandlerTest : public testing::Test {
88 protected:
StreamResetHandlerTest()89 StreamResetHandlerTest()
90 : ctx_(&callbacks_),
91 timer_manager_([this](webrtc::TaskQueueBase::DelayPrecision precision) {
92 return callbacks_.CreateTimeout(precision);
93 }),
94 delayed_ack_timer_(timer_manager_.CreateTimer(
95 "test/delayed_ack",
__anond23c25b90302() 96 []() { return absl::nullopt; },
97 TimerOptions(DurationMs(0)))),
98 t3_rtx_timer_(timer_manager_.CreateTimer(
99 "test/t3_rtx",
__anond23c25b90402() 100 []() { return absl::nullopt; },
101 TimerOptions(DurationMs(0)))),
102 data_tracker_(std::make_unique<DataTracker>("log: ",
103 delayed_ack_timer_.get(),
104 kPeerInitialTsn)),
105 reasm_(std::make_unique<ReassemblyQueue>("log: ",
106 kPeerInitialTsn,
107 kArwnd)),
108 retransmission_queue_(std::make_unique<RetransmissionQueue>(
109 "",
110 &callbacks_,
111 kMyInitialTsn,
112 kArwnd,
113 producer_,
__anond23c25b90502(DurationMs rtt_ms) 114 [](DurationMs rtt_ms) {},
__anond23c25b90602() 115 []() {},
116 *t3_rtx_timer_,
117 DcSctpOptions())),
118 handler_(
119 std::make_unique<StreamResetHandler>("log: ",
120 &ctx_,
121 &timer_manager_,
122 data_tracker_.get(),
123 reasm_.get(),
124 retransmission_queue_.get())) {
125 EXPECT_CALL(ctx_, current_rto).WillRepeatedly(Return(kRto));
126 }
127
AdvanceTime(DurationMs duration)128 void AdvanceTime(DurationMs duration) {
129 callbacks_.AdvanceTime(kRto);
130 for (;;) {
131 absl::optional<TimeoutID> timeout_id = callbacks_.GetNextExpiredTimeout();
132 if (!timeout_id.has_value()) {
133 break;
134 }
135 timer_manager_.HandleTimeout(*timeout_id);
136 }
137 }
138
139 // Handles the passed in RE-CONFIG `chunk` and returns the responses
140 // that are sent in the response RE-CONFIG.
HandleAndCatchResponse(ReConfigChunk chunk)141 std::vector<ReconfigurationResponseParameter> HandleAndCatchResponse(
142 ReConfigChunk chunk) {
143 handler_->HandleReConfig(std::move(chunk));
144
145 std::vector<uint8_t> payload = callbacks_.ConsumeSentPacket();
146 if (payload.empty()) {
147 EXPECT_TRUE(false);
148 return {};
149 }
150
151 std::vector<ReconfigurationResponseParameter> responses;
152 absl::optional<SctpPacket> p = SctpPacket::Parse(payload);
153 if (!p.has_value()) {
154 EXPECT_TRUE(false);
155 return {};
156 }
157 if (p->descriptors().size() != 1) {
158 EXPECT_TRUE(false);
159 return {};
160 }
161 absl::optional<ReConfigChunk> response_chunk =
162 ReConfigChunk::Parse(p->descriptors()[0].data);
163 if (!response_chunk.has_value()) {
164 EXPECT_TRUE(false);
165 return {};
166 }
167 for (const auto& desc : response_chunk->parameters().descriptors()) {
168 if (desc.type == ReconfigurationResponseParameter::kType) {
169 absl::optional<ReconfigurationResponseParameter> response =
170 ReconfigurationResponseParameter::Parse(desc.data);
171 if (!response.has_value()) {
172 EXPECT_TRUE(false);
173 return {};
174 }
175 responses.emplace_back(*std::move(response));
176 }
177 }
178 return responses;
179 }
180
PerformHandover()181 void PerformHandover() {
182 EXPECT_TRUE(handler_->GetHandoverReadiness().IsReady());
183 EXPECT_TRUE(data_tracker_->GetHandoverReadiness().IsReady());
184 EXPECT_TRUE(reasm_->GetHandoverReadiness().IsReady());
185 EXPECT_TRUE(retransmission_queue_->GetHandoverReadiness().IsReady());
186
187 DcSctpSocketHandoverState state;
188 handler_->AddHandoverState(state);
189 data_tracker_->AddHandoverState(state);
190 reasm_->AddHandoverState(state);
191
192 retransmission_queue_->AddHandoverState(state);
193
194 g_handover_state_transformer_for_test(&state);
195
196 data_tracker_ = std::make_unique<DataTracker>(
197 "log: ", delayed_ack_timer_.get(), kPeerInitialTsn);
198 data_tracker_->RestoreFromState(state);
199 reasm_ =
200 std::make_unique<ReassemblyQueue>("log: ", kPeerInitialTsn, kArwnd);
201 reasm_->RestoreFromState(state);
202 retransmission_queue_ = std::make_unique<RetransmissionQueue>(
203 "", &callbacks_, kMyInitialTsn, kArwnd, producer_,
204 [](DurationMs rtt_ms) {}, []() {}, *t3_rtx_timer_, DcSctpOptions(),
205 /*supports_partial_reliability=*/true,
206 /*use_message_interleaving=*/false);
207 retransmission_queue_->RestoreFromState(state);
208 handler_ = std::make_unique<StreamResetHandler>(
209 "log: ", &ctx_, &timer_manager_, data_tracker_.get(), reasm_.get(),
210 retransmission_queue_.get(), &state);
211 }
212
213 DataGenerator gen_;
214 NiceMock<MockDcSctpSocketCallbacks> callbacks_;
215 NiceMock<MockContext> ctx_;
216 NiceMock<MockSendQueue> producer_;
217 TimerManager timer_manager_;
218 std::unique_ptr<Timer> delayed_ack_timer_;
219 std::unique_ptr<Timer> t3_rtx_timer_;
220 std::unique_ptr<DataTracker> data_tracker_;
221 std::unique_ptr<ReassemblyQueue> reasm_;
222 std::unique_ptr<RetransmissionQueue> retransmission_queue_;
223 std::unique_ptr<StreamResetHandler> handler_;
224 };
225
TEST_F(StreamResetHandlerTest,ChunkWithNoParametersReturnsError)226 TEST_F(StreamResetHandlerTest, ChunkWithNoParametersReturnsError) {
227 EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0);
228 EXPECT_CALL(callbacks_, OnError).Times(1);
229 handler_->HandleReConfig(ReConfigChunk(Parameters()));
230 }
231
TEST_F(StreamResetHandlerTest,ChunkWithInvalidParametersReturnsError)232 TEST_F(StreamResetHandlerTest, ChunkWithInvalidParametersReturnsError) {
233 Parameters::Builder builder;
234 // Two OutgoingSSNResetRequestParameter in a RE-CONFIG is not valid.
235 builder.Add(OutgoingSSNResetRequestParameter(ReconfigRequestSN(1),
236 ReconfigRequestSN(10),
237 kPeerInitialTsn, {StreamID(1)}));
238 builder.Add(OutgoingSSNResetRequestParameter(ReconfigRequestSN(2),
239 ReconfigRequestSN(10),
240 kPeerInitialTsn, {StreamID(2)}));
241
242 EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0);
243 EXPECT_CALL(callbacks_, OnError).Times(1);
244 handler_->HandleReConfig(ReConfigChunk(builder.Build()));
245 }
246
TEST_F(StreamResetHandlerTest,FailToDeliverWithoutResettingStream)247 TEST_F(StreamResetHandlerTest, FailToDeliverWithoutResettingStream) {
248 reasm_->Add(kPeerInitialTsn, gen_.Ordered({1, 2, 3, 4}, "BE"));
249 reasm_->Add(AddTo(kPeerInitialTsn, 1), gen_.Ordered({1, 2, 3, 4}, "BE"));
250
251 data_tracker_->Observe(kPeerInitialTsn);
252 data_tracker_->Observe(AddTo(kPeerInitialTsn, 1));
253 EXPECT_THAT(reasm_->FlushMessages(),
254 UnorderedElementsAre(
255 SctpMessageIs(StreamID(1), PPID(53), kShortPayload),
256 SctpMessageIs(StreamID(1), PPID(53), kShortPayload)));
257
258 gen_.ResetStream();
259 reasm_->Add(AddTo(kPeerInitialTsn, 2), gen_.Ordered({1, 2, 3, 4}, "BE"));
260 EXPECT_THAT(reasm_->FlushMessages(), IsEmpty());
261 }
262
TEST_F(StreamResetHandlerTest,ResetStreamsNotDeferred)263 TEST_F(StreamResetHandlerTest, ResetStreamsNotDeferred) {
264 reasm_->Add(kPeerInitialTsn, gen_.Ordered({1, 2, 3, 4}, "BE"));
265 reasm_->Add(AddTo(kPeerInitialTsn, 1), gen_.Ordered({1, 2, 3, 4}, "BE"));
266
267 data_tracker_->Observe(kPeerInitialTsn);
268 data_tracker_->Observe(AddTo(kPeerInitialTsn, 1));
269 EXPECT_THAT(reasm_->FlushMessages(),
270 UnorderedElementsAre(
271 SctpMessageIs(StreamID(1), PPID(53), kShortPayload),
272 SctpMessageIs(StreamID(1), PPID(53), kShortPayload)));
273
274 Parameters::Builder builder;
275 builder.Add(OutgoingSSNResetRequestParameter(
276 kPeerInitialReqSn, ReconfigRequestSN(3), AddTo(kPeerInitialTsn, 1),
277 {StreamID(1)}));
278
279 std::vector<ReconfigurationResponseParameter> responses =
280 HandleAndCatchResponse(ReConfigChunk(builder.Build()));
281 EXPECT_THAT(responses, SizeIs(1));
282 EXPECT_EQ(responses[0].result(), ResponseResult::kSuccessPerformed);
283
284 gen_.ResetStream();
285 reasm_->Add(AddTo(kPeerInitialTsn, 2), gen_.Ordered({1, 2, 3, 4}, "BE"));
286 EXPECT_THAT(reasm_->FlushMessages(),
287 UnorderedElementsAre(
288 SctpMessageIs(StreamID(1), PPID(53), kShortPayload)));
289 }
290
TEST_F(StreamResetHandlerTest,ResetStreamsDeferred)291 TEST_F(StreamResetHandlerTest, ResetStreamsDeferred) {
292 DataGeneratorOptions opts;
293 opts.message_id = MID(0);
294 reasm_->Add(kPeerInitialTsn, gen_.Ordered({1, 2, 3, 4}, "BE", opts));
295
296 opts.message_id = MID(1);
297 reasm_->Add(AddTo(kPeerInitialTsn, 1),
298 gen_.Ordered({1, 2, 3, 4}, "BE", opts));
299
300 data_tracker_->Observe(kPeerInitialTsn);
301 data_tracker_->Observe(AddTo(kPeerInitialTsn, 1));
302 EXPECT_THAT(reasm_->FlushMessages(),
303 UnorderedElementsAre(
304 SctpMessageIs(StreamID(1), PPID(53), kShortPayload),
305 SctpMessageIs(StreamID(1), PPID(53), kShortPayload)));
306
307 Parameters::Builder builder;
308 builder.Add(OutgoingSSNResetRequestParameter(
309 kPeerInitialReqSn, ReconfigRequestSN(3), AddTo(kPeerInitialTsn, 3),
310 {StreamID(1)}));
311
312 std::vector<ReconfigurationResponseParameter> responses =
313 HandleAndCatchResponse(ReConfigChunk(builder.Build()));
314 EXPECT_THAT(responses, SizeIs(1));
315 EXPECT_EQ(responses[0].result(), ResponseResult::kInProgress);
316
317 opts.message_id = MID(1);
318 opts.ppid = PPID(5);
319 reasm_->Add(AddTo(kPeerInitialTsn, 5),
320 gen_.Ordered({1, 2, 3, 4}, "BE", opts));
321 reasm_->MaybeResetStreamsDeferred(AddTo(kPeerInitialTsn, 1));
322
323 opts.message_id = MID(0);
324 opts.ppid = PPID(4);
325 reasm_->Add(AddTo(kPeerInitialTsn, 4),
326 gen_.Ordered({1, 2, 3, 4}, "BE", opts));
327 reasm_->MaybeResetStreamsDeferred(AddTo(kPeerInitialTsn, 1));
328
329 opts.message_id = MID(3);
330 opts.ppid = PPID(3);
331 reasm_->Add(AddTo(kPeerInitialTsn, 3),
332 gen_.Ordered({1, 2, 3, 4}, "BE", opts));
333 reasm_->MaybeResetStreamsDeferred(AddTo(kPeerInitialTsn, 1));
334
335 opts.message_id = MID(2);
336 opts.ppid = PPID(2);
337 reasm_->Add(AddTo(kPeerInitialTsn, 2),
338 gen_.Ordered({1, 2, 3, 4}, "BE", opts));
339 reasm_->MaybeResetStreamsDeferred(AddTo(kPeerInitialTsn, 5));
340
341 EXPECT_THAT(
342 reasm_->FlushMessages(),
343 UnorderedElementsAre(SctpMessageIs(StreamID(1), PPID(2), kShortPayload),
344 SctpMessageIs(StreamID(1), PPID(3), kShortPayload),
345 SctpMessageIs(StreamID(1), PPID(4), kShortPayload),
346 SctpMessageIs(StreamID(1), PPID(5), kShortPayload)));
347 }
348
TEST_F(StreamResetHandlerTest,SendOutgoingRequestDirectly)349 TEST_F(StreamResetHandlerTest, SendOutgoingRequestDirectly) {
350 EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
351 handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
352
353 EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
354 EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
355 .WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
356
357 absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest();
358 ASSERT_TRUE(reconfig.has_value());
359 ASSERT_HAS_VALUE_AND_ASSIGN(
360 OutgoingSSNResetRequestParameter req,
361 reconfig->parameters().get<OutgoingSSNResetRequestParameter>());
362
363 EXPECT_EQ(req.request_sequence_number(), kMyInitialReqSn);
364 EXPECT_EQ(req.sender_last_assigned_tsn(),
365 TSN(*retransmission_queue_->next_tsn() - 1));
366 EXPECT_THAT(req.stream_ids(), UnorderedElementsAre(StreamID(42)));
367 }
368
TEST_F(StreamResetHandlerTest,ResetMultipleStreamsInOneRequest)369 TEST_F(StreamResetHandlerTest, ResetMultipleStreamsInOneRequest) {
370 EXPECT_CALL(producer_, PrepareResetStream(StreamID(40)));
371 EXPECT_CALL(producer_, PrepareResetStream(StreamID(41)));
372 EXPECT_CALL(producer_, PrepareResetStream(StreamID(42))).Times(2);
373 EXPECT_CALL(producer_, PrepareResetStream(StreamID(43)));
374 EXPECT_CALL(producer_, PrepareResetStream(StreamID(44)));
375 handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
376 handler_->ResetStreams(
377 std::vector<StreamID>({StreamID(43), StreamID(44), StreamID(41)}));
378 handler_->ResetStreams(std::vector<StreamID>({StreamID(42), StreamID(40)}));
379
380 EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
381 EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
382 .WillOnce(Return(
383 std::vector<StreamID>({StreamID(40), StreamID(41), StreamID(42),
384 StreamID(43), StreamID(44)})));
385 absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest();
386 ASSERT_TRUE(reconfig.has_value());
387 ASSERT_HAS_VALUE_AND_ASSIGN(
388 OutgoingSSNResetRequestParameter req,
389 reconfig->parameters().get<OutgoingSSNResetRequestParameter>());
390
391 EXPECT_EQ(req.request_sequence_number(), kMyInitialReqSn);
392 EXPECT_EQ(req.sender_last_assigned_tsn(),
393 TSN(*retransmission_queue_->next_tsn() - 1));
394 EXPECT_THAT(req.stream_ids(),
395 UnorderedElementsAre(StreamID(40), StreamID(41), StreamID(42),
396 StreamID(43), StreamID(44)));
397 }
398
TEST_F(StreamResetHandlerTest,SendOutgoingRequestDeferred)399 TEST_F(StreamResetHandlerTest, SendOutgoingRequestDeferred) {
400 EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
401 handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
402
403 EXPECT_CALL(producer_, HasStreamsReadyToBeReset())
404 .WillOnce(Return(false))
405 .WillOnce(Return(false))
406 .WillOnce(Return(true));
407
408 EXPECT_FALSE(handler_->MakeStreamResetRequest().has_value());
409 EXPECT_FALSE(handler_->MakeStreamResetRequest().has_value());
410 EXPECT_TRUE(handler_->MakeStreamResetRequest().has_value());
411 }
412
TEST_F(StreamResetHandlerTest,SendOutgoingResettingOnPositiveResponse)413 TEST_F(StreamResetHandlerTest, SendOutgoingResettingOnPositiveResponse) {
414 EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
415 handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
416
417 EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
418 EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
419 .WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
420
421 absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest();
422 ASSERT_TRUE(reconfig.has_value());
423 ASSERT_HAS_VALUE_AND_ASSIGN(
424 OutgoingSSNResetRequestParameter req,
425 reconfig->parameters().get<OutgoingSSNResetRequestParameter>());
426
427 Parameters::Builder builder;
428 builder.Add(ReconfigurationResponseParameter(
429 req.request_sequence_number(), ResponseResult::kSuccessPerformed));
430 ReConfigChunk response_reconfig(builder.Build());
431
432 EXPECT_CALL(producer_, CommitResetStreams);
433 EXPECT_CALL(producer_, RollbackResetStreams).Times(0);
434
435 // Processing a response shouldn't result in sending anything.
436 EXPECT_CALL(callbacks_, OnError).Times(0);
437 EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0);
438 handler_->HandleReConfig(std::move(response_reconfig));
439 }
440
TEST_F(StreamResetHandlerTest,SendOutgoingResetRollbackOnError)441 TEST_F(StreamResetHandlerTest, SendOutgoingResetRollbackOnError) {
442 EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
443 handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
444
445 EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
446 EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
447 .WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
448
449 absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest();
450 ASSERT_TRUE(reconfig.has_value());
451 ASSERT_HAS_VALUE_AND_ASSIGN(
452 OutgoingSSNResetRequestParameter req,
453 reconfig->parameters().get<OutgoingSSNResetRequestParameter>());
454
455 Parameters::Builder builder;
456 builder.Add(ReconfigurationResponseParameter(
457 req.request_sequence_number(), ResponseResult::kErrorBadSequenceNumber));
458 ReConfigChunk response_reconfig(builder.Build());
459
460 EXPECT_CALL(producer_, CommitResetStreams).Times(0);
461 EXPECT_CALL(producer_, RollbackResetStreams);
462
463 // Only requests should result in sending responses.
464 EXPECT_CALL(callbacks_, OnError).Times(0);
465 EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0);
466 handler_->HandleReConfig(std::move(response_reconfig));
467 }
468
TEST_F(StreamResetHandlerTest,SendOutgoingResetRetransmitOnInProgress)469 TEST_F(StreamResetHandlerTest, SendOutgoingResetRetransmitOnInProgress) {
470 static constexpr StreamID kStreamToReset = StreamID(42);
471
472 EXPECT_CALL(producer_, PrepareResetStream(kStreamToReset));
473 handler_->ResetStreams(std::vector<StreamID>({kStreamToReset}));
474
475 EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
476 EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
477 .WillOnce(Return(std::vector<StreamID>({kStreamToReset})));
478
479 absl::optional<ReConfigChunk> reconfig1 = handler_->MakeStreamResetRequest();
480 ASSERT_TRUE(reconfig1.has_value());
481 ASSERT_HAS_VALUE_AND_ASSIGN(
482 OutgoingSSNResetRequestParameter req1,
483 reconfig1->parameters().get<OutgoingSSNResetRequestParameter>());
484
485 // Simulate that the peer responded "In Progress".
486 Parameters::Builder builder;
487 builder.Add(ReconfigurationResponseParameter(req1.request_sequence_number(),
488 ResponseResult::kInProgress));
489 ReConfigChunk response_reconfig(builder.Build());
490
491 EXPECT_CALL(producer_, CommitResetStreams()).Times(0);
492 EXPECT_CALL(producer_, RollbackResetStreams()).Times(0);
493
494 // Processing a response shouldn't result in sending anything.
495 EXPECT_CALL(callbacks_, OnError).Times(0);
496 EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0);
497 handler_->HandleReConfig(std::move(response_reconfig));
498
499 // Let some time pass, so that the reconfig timer expires, and retries the
500 // same request.
501 EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(1);
502 AdvanceTime(kRto);
503
504 std::vector<uint8_t> payload = callbacks_.ConsumeSentPacket();
505 ASSERT_FALSE(payload.empty());
506
507 ASSERT_HAS_VALUE_AND_ASSIGN(SctpPacket packet, SctpPacket::Parse(payload));
508 ASSERT_THAT(packet.descriptors(), SizeIs(1));
509 ASSERT_HAS_VALUE_AND_ASSIGN(
510 ReConfigChunk reconfig2,
511 ReConfigChunk::Parse(packet.descriptors()[0].data));
512
513 ASSERT_HAS_VALUE_AND_ASSIGN(
514 OutgoingSSNResetRequestParameter req2,
515 reconfig2.parameters().get<OutgoingSSNResetRequestParameter>());
516
517 EXPECT_EQ(req2.request_sequence_number(),
518 AddTo(req1.request_sequence_number(), 1));
519 EXPECT_THAT(req2.stream_ids(), UnorderedElementsAre(kStreamToReset));
520 }
521
TEST_F(StreamResetHandlerTest,ResetWhileRequestIsSentWillQueue)522 TEST_F(StreamResetHandlerTest, ResetWhileRequestIsSentWillQueue) {
523 EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
524 handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
525
526 EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
527 EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
528 .WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
529
530 absl::optional<ReConfigChunk> reconfig1 = handler_->MakeStreamResetRequest();
531 ASSERT_TRUE(reconfig1.has_value());
532 ASSERT_HAS_VALUE_AND_ASSIGN(
533 OutgoingSSNResetRequestParameter req1,
534 reconfig1->parameters().get<OutgoingSSNResetRequestParameter>());
535 EXPECT_EQ(req1.request_sequence_number(), kMyInitialReqSn);
536 EXPECT_EQ(req1.sender_last_assigned_tsn(),
537 AddTo(retransmission_queue_->next_tsn(), -1));
538 EXPECT_THAT(req1.stream_ids(), UnorderedElementsAre(StreamID(42)));
539
540 // Streams reset while the request is in-flight will be queued.
541 EXPECT_CALL(producer_, PrepareResetStream(StreamID(41)));
542 EXPECT_CALL(producer_, PrepareResetStream(StreamID(43)));
543 StreamID stream_ids[] = {StreamID(41), StreamID(43)};
544 handler_->ResetStreams(stream_ids);
545 EXPECT_EQ(handler_->MakeStreamResetRequest(), absl::nullopt);
546
547 Parameters::Builder builder;
548 builder.Add(ReconfigurationResponseParameter(
549 req1.request_sequence_number(), ResponseResult::kSuccessPerformed));
550 ReConfigChunk response_reconfig(builder.Build());
551
552 EXPECT_CALL(producer_, CommitResetStreams()).Times(1);
553 EXPECT_CALL(producer_, RollbackResetStreams()).Times(0);
554
555 // Processing a response shouldn't result in sending anything.
556 EXPECT_CALL(callbacks_, OnError).Times(0);
557 EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0);
558 handler_->HandleReConfig(std::move(response_reconfig));
559
560 // Response has been processed. A new request can be sent.
561 EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
562 EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
563 .WillOnce(Return(std::vector<StreamID>({StreamID(41), StreamID(43)})));
564
565 absl::optional<ReConfigChunk> reconfig2 = handler_->MakeStreamResetRequest();
566 ASSERT_TRUE(reconfig2.has_value());
567 ASSERT_HAS_VALUE_AND_ASSIGN(
568 OutgoingSSNResetRequestParameter req2,
569 reconfig2->parameters().get<OutgoingSSNResetRequestParameter>());
570 EXPECT_EQ(req2.request_sequence_number(), AddTo(kMyInitialReqSn, 1));
571 EXPECT_EQ(req2.sender_last_assigned_tsn(),
572 TSN(*retransmission_queue_->next_tsn() - 1));
573 EXPECT_THAT(req2.stream_ids(),
574 UnorderedElementsAre(StreamID(41), StreamID(43)));
575 }
576
TEST_F(StreamResetHandlerTest,SendIncomingResetJustReturnsNothingPerformed)577 TEST_F(StreamResetHandlerTest, SendIncomingResetJustReturnsNothingPerformed) {
578 Parameters::Builder builder;
579 builder.Add(
580 IncomingSSNResetRequestParameter(kPeerInitialReqSn, {StreamID(1)}));
581
582 std::vector<ReconfigurationResponseParameter> responses =
583 HandleAndCatchResponse(ReConfigChunk(builder.Build()));
584 ASSERT_THAT(responses, SizeIs(1));
585 EXPECT_THAT(responses[0].response_sequence_number(), kPeerInitialReqSn);
586 EXPECT_THAT(responses[0].result(), ResponseResult::kSuccessNothingToDo);
587 }
588
TEST_F(StreamResetHandlerTest,SendSameRequestTwiceIsIdempotent)589 TEST_F(StreamResetHandlerTest, SendSameRequestTwiceIsIdempotent) {
590 // Simulate that receiving the same chunk twice (due to network issues,
591 // or retransmissions, causing a RECONFIG to be re-received) is idempotent.
592 for (int i = 0; i < 2; ++i) {
593 Parameters::Builder builder;
594 builder.Add(OutgoingSSNResetRequestParameter(
595 kPeerInitialReqSn, ReconfigRequestSN(3), AddTo(kPeerInitialTsn, 1),
596 {StreamID(1)}));
597
598 std::vector<ReconfigurationResponseParameter> responses1 =
599 HandleAndCatchResponse(ReConfigChunk(builder.Build()));
600 EXPECT_THAT(responses1, SizeIs(1));
601 EXPECT_EQ(responses1[0].result(), ResponseResult::kInProgress);
602 }
603 }
604
TEST_F(StreamResetHandlerTest,HandoverIsAllowedOnlyWhenNoStreamIsBeingOrWillBeReset)605 TEST_F(StreamResetHandlerTest,
606 HandoverIsAllowedOnlyWhenNoStreamIsBeingOrWillBeReset) {
607 EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
608 handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
609 EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
610 EXPECT_EQ(
611 handler_->GetHandoverReadiness(),
612 HandoverReadinessStatus(HandoverUnreadinessReason::kPendingStreamReset));
613
614 EXPECT_CALL(producer_, HasStreamsReadyToBeReset())
615 .WillOnce(Return(true))
616 .WillOnce(Return(false));
617 EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
618 .WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
619
620 ASSERT_TRUE(handler_->MakeStreamResetRequest().has_value());
621 EXPECT_EQ(handler_->GetHandoverReadiness(),
622 HandoverReadinessStatus(
623 HandoverUnreadinessReason::kPendingStreamResetRequest));
624
625 // Reset more streams while the request is in-flight.
626 EXPECT_CALL(producer_, PrepareResetStream(StreamID(41)));
627 EXPECT_CALL(producer_, PrepareResetStream(StreamID(43)));
628 StreamID stream_ids[] = {StreamID(41), StreamID(43)};
629 handler_->ResetStreams(stream_ids);
630
631 EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
632 EXPECT_EQ(handler_->GetHandoverReadiness(),
633 HandoverReadinessStatus()
634 .Add(HandoverUnreadinessReason::kPendingStreamResetRequest)
635 .Add(HandoverUnreadinessReason::kPendingStreamReset));
636
637 // Processing a response to first request.
638 EXPECT_CALL(producer_, CommitResetStreams()).Times(1);
639 handler_->HandleReConfig(
640 ReConfigChunk(Parameters::Builder()
641 .Add(ReconfigurationResponseParameter(
642 kMyInitialReqSn, ResponseResult::kSuccessPerformed))
643 .Build()));
644 EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
645 EXPECT_EQ(
646 handler_->GetHandoverReadiness(),
647 HandoverReadinessStatus(HandoverUnreadinessReason::kPendingStreamReset));
648
649 // Second request can be sent.
650 EXPECT_CALL(producer_, HasStreamsReadyToBeReset())
651 .WillOnce(Return(true))
652 .WillOnce(Return(false));
653 EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
654 .WillOnce(Return(std::vector<StreamID>({StreamID(41), StreamID(43)})));
655
656 ASSERT_TRUE(handler_->MakeStreamResetRequest().has_value());
657 EXPECT_EQ(handler_->GetHandoverReadiness(),
658 HandoverReadinessStatus(
659 HandoverUnreadinessReason::kPendingStreamResetRequest));
660
661 // Processing a response to second request.
662 EXPECT_CALL(producer_, CommitResetStreams()).Times(1);
663 handler_->HandleReConfig(ReConfigChunk(
664 Parameters::Builder()
665 .Add(ReconfigurationResponseParameter(
666 AddTo(kMyInitialReqSn, 1), ResponseResult::kSuccessPerformed))
667 .Build()));
668
669 // Seconds response has been processed. No pending resets.
670 EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(false));
671
672 EXPECT_TRUE(handler_->GetHandoverReadiness().IsReady());
673 }
674
TEST_F(StreamResetHandlerTest,HandoverInInitialState)675 TEST_F(StreamResetHandlerTest, HandoverInInitialState) {
676 PerformHandover();
677
678 EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
679 handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
680
681 EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
682 EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
683 .WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
684
685 absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest();
686 ASSERT_TRUE(reconfig.has_value());
687 ASSERT_HAS_VALUE_AND_ASSIGN(
688 OutgoingSSNResetRequestParameter req,
689 reconfig->parameters().get<OutgoingSSNResetRequestParameter>());
690
691 EXPECT_EQ(req.request_sequence_number(), kMyInitialReqSn);
692 EXPECT_EQ(req.sender_last_assigned_tsn(),
693 TSN(*retransmission_queue_->next_tsn() - 1));
694 EXPECT_THAT(req.stream_ids(), UnorderedElementsAre(StreamID(42)));
695 }
696
TEST_F(StreamResetHandlerTest,HandoverAfterHavingResetOneStream)697 TEST_F(StreamResetHandlerTest, HandoverAfterHavingResetOneStream) {
698 // Reset one stream
699 {
700 EXPECT_CALL(producer_, PrepareResetStream(StreamID(42)));
701 handler_->ResetStreams(std::vector<StreamID>({StreamID(42)}));
702
703 EXPECT_CALL(producer_, HasStreamsReadyToBeReset())
704 .WillOnce(Return(true))
705 .WillOnce(Return(false));
706 EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
707 .WillOnce(Return(std::vector<StreamID>({StreamID(42)})));
708
709 ASSERT_HAS_VALUE_AND_ASSIGN(ReConfigChunk reconfig,
710 handler_->MakeStreamResetRequest());
711 ASSERT_HAS_VALUE_AND_ASSIGN(
712 OutgoingSSNResetRequestParameter req,
713 reconfig.parameters().get<OutgoingSSNResetRequestParameter>());
714 EXPECT_EQ(req.request_sequence_number(), kMyInitialReqSn);
715 EXPECT_EQ(req.sender_last_assigned_tsn(),
716 TSN(*retransmission_queue_->next_tsn() - 1));
717 EXPECT_THAT(req.stream_ids(), UnorderedElementsAre(StreamID(42)));
718
719 EXPECT_CALL(producer_, CommitResetStreams()).Times(1);
720 handler_->HandleReConfig(
721 ReConfigChunk(Parameters::Builder()
722 .Add(ReconfigurationResponseParameter(
723 req.request_sequence_number(),
724 ResponseResult::kSuccessPerformed))
725 .Build()));
726 }
727
728 PerformHandover();
729
730 // Reset another stream after handover
731 {
732 EXPECT_CALL(producer_, PrepareResetStream(StreamID(43)));
733 handler_->ResetStreams(std::vector<StreamID>({StreamID(43)}));
734
735 EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true));
736 EXPECT_CALL(producer_, GetStreamsReadyToBeReset())
737 .WillOnce(Return(std::vector<StreamID>({StreamID(43)})));
738
739 ASSERT_HAS_VALUE_AND_ASSIGN(ReConfigChunk reconfig,
740 handler_->MakeStreamResetRequest());
741 ASSERT_HAS_VALUE_AND_ASSIGN(
742 OutgoingSSNResetRequestParameter req,
743 reconfig.parameters().get<OutgoingSSNResetRequestParameter>());
744
745 EXPECT_EQ(req.request_sequence_number(),
746 ReconfigRequestSN(kMyInitialReqSn.value() + 1));
747 EXPECT_EQ(req.sender_last_assigned_tsn(),
748 TSN(*retransmission_queue_->next_tsn() - 1));
749 EXPECT_THAT(req.stream_ids(), UnorderedElementsAre(StreamID(43)));
750 }
751 }
752
TEST_F(StreamResetHandlerTest,PerformCloseAfterOneFirstFailing)753 TEST_F(StreamResetHandlerTest, PerformCloseAfterOneFirstFailing) {
754 // Inject a stream reset on the first expected TSN (which hasn't been seen).
755 Parameters::Builder builder;
756 builder.Add(OutgoingSSNResetRequestParameter(
757 kPeerInitialReqSn, ReconfigRequestSN(3), kPeerInitialTsn, {StreamID(1)}));
758
759 // The socket is expected to say "in progress" as that TSN hasn't been seen.
760 std::vector<ReconfigurationResponseParameter> responses =
761 HandleAndCatchResponse(ReConfigChunk(builder.Build()));
762 EXPECT_THAT(responses, SizeIs(1));
763 EXPECT_EQ(responses[0].result(), ResponseResult::kInProgress);
764
765 // Let the socket receive the TSN.
766 DataGeneratorOptions opts;
767 opts.message_id = MID(0);
768 reasm_->Add(kPeerInitialTsn, gen_.Ordered({1, 2, 3, 4}, "BE", opts));
769 reasm_->MaybeResetStreamsDeferred(kPeerInitialTsn);
770 data_tracker_->Observe(kPeerInitialTsn);
771
772 // And emulate that time has passed, and the peer retries the stream reset,
773 // but now with an incremented request sequence number.
774 Parameters::Builder builder2;
775 builder2.Add(OutgoingSSNResetRequestParameter(
776 ReconfigRequestSN(*kPeerInitialReqSn + 1), ReconfigRequestSN(3),
777 kPeerInitialTsn, {StreamID(1)}));
778
779 // This is supposed to be handled well.
780 std::vector<ReconfigurationResponseParameter> responses2 =
781 HandleAndCatchResponse(ReConfigChunk(builder2.Build()));
782 EXPECT_THAT(responses2, SizeIs(1));
783 EXPECT_EQ(responses2[0].result(), ResponseResult::kSuccessPerformed);
784 }
785 } // namespace
786 } // namespace dcsctp
787