diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-19 00:47:55 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-19 00:47:55 +0000 |
commit | 26a029d407be480d791972afb5975cf62c9360a6 (patch) | |
tree | f435a8308119effd964b339f76abb83a57c29483 /third_party/libwebrtc/net/dcsctp/socket/stream_reset_handler_test.cc | |
parent | Initial commit. (diff) | |
download | firefox-26a029d407be480d791972afb5975cf62c9360a6.tar.xz firefox-26a029d407be480d791972afb5975cf62c9360a6.zip |
Adding upstream version 124.0.1.upstream/124.0.1
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'third_party/libwebrtc/net/dcsctp/socket/stream_reset_handler_test.cc')
-rw-r--r-- | third_party/libwebrtc/net/dcsctp/socket/stream_reset_handler_test.cc | 916 |
1 files changed, 916 insertions, 0 deletions
diff --git a/third_party/libwebrtc/net/dcsctp/socket/stream_reset_handler_test.cc b/third_party/libwebrtc/net/dcsctp/socket/stream_reset_handler_test.cc new file mode 100644 index 0000000000..091d717f8a --- /dev/null +++ b/third_party/libwebrtc/net/dcsctp/socket/stream_reset_handler_test.cc @@ -0,0 +1,916 @@ +/* + * Copyright (c) 2021 The WebRTC project authors. All Rights Reserved. + * + * Use of this source code is governed by a BSD-style license + * that can be found in the LICENSE file in the root of the source + * tree. An additional intellectual property rights grant can be found + * in the file PATENTS. All contributing project authors may + * be found in the AUTHORS file in the root of the source tree. + */ +#include "net/dcsctp/socket/stream_reset_handler.h" + +#include <array> +#include <cstdint> +#include <memory> +#include <type_traits> +#include <vector> + +#include "absl/types/optional.h" +#include "api/array_view.h" +#include "api/task_queue/task_queue_base.h" +#include "net/dcsctp/common/handover_testing.h" +#include "net/dcsctp/common/internal_types.h" +#include "net/dcsctp/packet/chunk/forward_tsn_common.h" +#include "net/dcsctp/packet/chunk/reconfig_chunk.h" +#include "net/dcsctp/packet/parameter/incoming_ssn_reset_request_parameter.h" +#include "net/dcsctp/packet/parameter/outgoing_ssn_reset_request_parameter.h" +#include "net/dcsctp/packet/parameter/parameter.h" +#include "net/dcsctp/packet/parameter/reconfiguration_response_parameter.h" +#include "net/dcsctp/public/dcsctp_message.h" +#include "net/dcsctp/public/types.h" +#include "net/dcsctp/rx/data_tracker.h" +#include "net/dcsctp/rx/reassembly_queue.h" +#include "net/dcsctp/socket/mock_context.h" +#include "net/dcsctp/socket/mock_dcsctp_socket_callbacks.h" +#include "net/dcsctp/testing/data_generator.h" +#include "net/dcsctp/testing/testing_macros.h" +#include "net/dcsctp/timer/timer.h" +#include "net/dcsctp/tx/mock_send_queue.h" +#include "net/dcsctp/tx/retransmission_queue.h" +#include "rtc_base/gunit.h" +#include "test/gmock.h" + +namespace dcsctp { +namespace { +using ::testing::IsEmpty; +using ::testing::NiceMock; +using ::testing::Property; +using ::testing::Return; +using ::testing::SizeIs; +using ::testing::UnorderedElementsAre; +using ResponseResult = ReconfigurationResponseParameter::Result; +using SkippedStream = AnyForwardTsnChunk::SkippedStream; + +constexpr TSN kMyInitialTsn = MockContext::MyInitialTsn(); +constexpr ReconfigRequestSN kMyInitialReqSn = ReconfigRequestSN(*kMyInitialTsn); +constexpr TSN kPeerInitialTsn = MockContext::PeerInitialTsn(); +constexpr ReconfigRequestSN kPeerInitialReqSn = + ReconfigRequestSN(*kPeerInitialTsn); +constexpr uint32_t kArwnd = 131072; +constexpr DurationMs kRto = DurationMs(250); + +constexpr std::array<uint8_t, 4> kShortPayload = {1, 2, 3, 4}; + +MATCHER_P3(SctpMessageIs, stream_id, ppid, expected_payload, "") { + if (arg.stream_id() != stream_id) { + *result_listener << "the stream_id is " << *arg.stream_id(); + return false; + } + + if (arg.ppid() != ppid) { + *result_listener << "the ppid is " << *arg.ppid(); + return false; + } + + if (std::vector<uint8_t>(arg.payload().begin(), arg.payload().end()) != + std::vector<uint8_t>(expected_payload.begin(), expected_payload.end())) { + *result_listener << "the payload is wrong"; + return false; + } + return true; +} + +TSN AddTo(TSN tsn, int delta) { + return TSN(*tsn + delta); +} + +ReconfigRequestSN AddTo(ReconfigRequestSN req_sn, int delta) { + return ReconfigRequestSN(*req_sn + delta); +} + +class StreamResetHandlerTest : public testing::Test { + protected: + StreamResetHandlerTest() + : ctx_(&callbacks_), + timer_manager_([this](webrtc::TaskQueueBase::DelayPrecision precision) { + return callbacks_.CreateTimeout(precision); + }), + delayed_ack_timer_(timer_manager_.CreateTimer( + "test/delayed_ack", + []() { return absl::nullopt; }, + TimerOptions(DurationMs(0)))), + t3_rtx_timer_(timer_manager_.CreateTimer( + "test/t3_rtx", + []() { return absl::nullopt; }, + TimerOptions(DurationMs(0)))), + data_tracker_(std::make_unique<DataTracker>("log: ", + delayed_ack_timer_.get(), + kPeerInitialTsn)), + reasm_(std::make_unique<ReassemblyQueue>("log: ", + kPeerInitialTsn, + kArwnd)), + retransmission_queue_(std::make_unique<RetransmissionQueue>( + "", + &callbacks_, + kMyInitialTsn, + kArwnd, + producer_, + [](DurationMs rtt_ms) {}, + []() {}, + *t3_rtx_timer_, + DcSctpOptions())), + handler_( + std::make_unique<StreamResetHandler>("log: ", + &ctx_, + &timer_manager_, + data_tracker_.get(), + reasm_.get(), + retransmission_queue_.get())) { + EXPECT_CALL(ctx_, current_rto).WillRepeatedly(Return(kRto)); + } + + void AdvanceTime(DurationMs duration) { + callbacks_.AdvanceTime(kRto); + for (;;) { + absl::optional<TimeoutID> timeout_id = callbacks_.GetNextExpiredTimeout(); + if (!timeout_id.has_value()) { + break; + } + timer_manager_.HandleTimeout(*timeout_id); + } + } + + // Handles the passed in RE-CONFIG `chunk` and returns the responses + // that are sent in the response RE-CONFIG. + std::vector<ReconfigurationResponseParameter> HandleAndCatchResponse( + ReConfigChunk chunk) { + handler_->HandleReConfig(std::move(chunk)); + + std::vector<uint8_t> payload = callbacks_.ConsumeSentPacket(); + if (payload.empty()) { + EXPECT_TRUE(false); + return {}; + } + + std::vector<ReconfigurationResponseParameter> responses; + absl::optional<SctpPacket> p = SctpPacket::Parse(payload, DcSctpOptions()); + if (!p.has_value()) { + EXPECT_TRUE(false); + return {}; + } + if (p->descriptors().size() != 1) { + EXPECT_TRUE(false); + return {}; + } + absl::optional<ReConfigChunk> response_chunk = + ReConfigChunk::Parse(p->descriptors()[0].data); + if (!response_chunk.has_value()) { + EXPECT_TRUE(false); + return {}; + } + for (const auto& desc : response_chunk->parameters().descriptors()) { + if (desc.type == ReconfigurationResponseParameter::kType) { + absl::optional<ReconfigurationResponseParameter> response = + ReconfigurationResponseParameter::Parse(desc.data); + if (!response.has_value()) { + EXPECT_TRUE(false); + return {}; + } + responses.emplace_back(*std::move(response)); + } + } + return responses; + } + + void PerformHandover() { + EXPECT_TRUE(handler_->GetHandoverReadiness().IsReady()); + EXPECT_TRUE(data_tracker_->GetHandoverReadiness().IsReady()); + EXPECT_TRUE(reasm_->GetHandoverReadiness().IsReady()); + EXPECT_TRUE(retransmission_queue_->GetHandoverReadiness().IsReady()); + + DcSctpSocketHandoverState state; + handler_->AddHandoverState(state); + data_tracker_->AddHandoverState(state); + reasm_->AddHandoverState(state); + + retransmission_queue_->AddHandoverState(state); + + g_handover_state_transformer_for_test(&state); + + data_tracker_ = std::make_unique<DataTracker>( + "log: ", delayed_ack_timer_.get(), kPeerInitialTsn); + data_tracker_->RestoreFromState(state); + reasm_ = + std::make_unique<ReassemblyQueue>("log: ", kPeerInitialTsn, kArwnd); + reasm_->RestoreFromState(state); + retransmission_queue_ = std::make_unique<RetransmissionQueue>( + "", &callbacks_, kMyInitialTsn, kArwnd, producer_, + [](DurationMs rtt_ms) {}, []() {}, *t3_rtx_timer_, DcSctpOptions(), + /*supports_partial_reliability=*/true, + /*use_message_interleaving=*/false); + retransmission_queue_->RestoreFromState(state); + handler_ = std::make_unique<StreamResetHandler>( + "log: ", &ctx_, &timer_manager_, data_tracker_.get(), reasm_.get(), + retransmission_queue_.get(), &state); + } + + DataGenerator gen_; + NiceMock<MockDcSctpSocketCallbacks> callbacks_; + NiceMock<MockContext> ctx_; + NiceMock<MockSendQueue> producer_; + TimerManager timer_manager_; + std::unique_ptr<Timer> delayed_ack_timer_; + std::unique_ptr<Timer> t3_rtx_timer_; + std::unique_ptr<DataTracker> data_tracker_; + std::unique_ptr<ReassemblyQueue> reasm_; + std::unique_ptr<RetransmissionQueue> retransmission_queue_; + std::unique_ptr<StreamResetHandler> handler_; +}; + +TEST_F(StreamResetHandlerTest, ChunkWithNoParametersReturnsError) { + EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0); + EXPECT_CALL(callbacks_, OnError).Times(1); + handler_->HandleReConfig(ReConfigChunk(Parameters())); +} + +TEST_F(StreamResetHandlerTest, ChunkWithInvalidParametersReturnsError) { + Parameters::Builder builder; + // Two OutgoingSSNResetRequestParameter in a RE-CONFIG is not valid. + builder.Add(OutgoingSSNResetRequestParameter(ReconfigRequestSN(1), + ReconfigRequestSN(10), + kPeerInitialTsn, {StreamID(1)})); + builder.Add(OutgoingSSNResetRequestParameter(ReconfigRequestSN(2), + ReconfigRequestSN(10), + kPeerInitialTsn, {StreamID(2)})); + + EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0); + EXPECT_CALL(callbacks_, OnError).Times(1); + handler_->HandleReConfig(ReConfigChunk(builder.Build())); +} + +TEST_F(StreamResetHandlerTest, FailToDeliverWithoutResettingStream) { + reasm_->Add(kPeerInitialTsn, gen_.Ordered({1, 2, 3, 4}, "BE")); + reasm_->Add(AddTo(kPeerInitialTsn, 1), gen_.Ordered({1, 2, 3, 4}, "BE")); + + data_tracker_->Observe(kPeerInitialTsn); + data_tracker_->Observe(AddTo(kPeerInitialTsn, 1)); + EXPECT_THAT(reasm_->FlushMessages(), + UnorderedElementsAre( + SctpMessageIs(StreamID(1), PPID(53), kShortPayload), + SctpMessageIs(StreamID(1), PPID(53), kShortPayload))); + + gen_.ResetStream(); + reasm_->Add(AddTo(kPeerInitialTsn, 2), gen_.Ordered({1, 2, 3, 4}, "BE")); + EXPECT_THAT(reasm_->FlushMessages(), IsEmpty()); +} + +TEST_F(StreamResetHandlerTest, ResetStreamsNotDeferred) { + reasm_->Add(kPeerInitialTsn, gen_.Ordered({1, 2, 3, 4}, "BE")); + reasm_->Add(AddTo(kPeerInitialTsn, 1), gen_.Ordered({1, 2, 3, 4}, "BE")); + + data_tracker_->Observe(kPeerInitialTsn); + data_tracker_->Observe(AddTo(kPeerInitialTsn, 1)); + EXPECT_THAT(reasm_->FlushMessages(), + UnorderedElementsAre( + SctpMessageIs(StreamID(1), PPID(53), kShortPayload), + SctpMessageIs(StreamID(1), PPID(53), kShortPayload))); + + Parameters::Builder builder; + builder.Add(OutgoingSSNResetRequestParameter( + kPeerInitialReqSn, ReconfigRequestSN(3), AddTo(kPeerInitialTsn, 1), + {StreamID(1)})); + + std::vector<ReconfigurationResponseParameter> responses = + HandleAndCatchResponse(ReConfigChunk(builder.Build())); + EXPECT_THAT(responses, SizeIs(1)); + EXPECT_EQ(responses[0].result(), ResponseResult::kSuccessPerformed); + + gen_.ResetStream(); + reasm_->Add(AddTo(kPeerInitialTsn, 2), gen_.Ordered({1, 2, 3, 4}, "BE")); + EXPECT_THAT(reasm_->FlushMessages(), + UnorderedElementsAre( + SctpMessageIs(StreamID(1), PPID(53), kShortPayload))); +} + +TEST_F(StreamResetHandlerTest, ResetStreamsDeferred) { + constexpr StreamID kStreamId = StreamID(1); + data_tracker_->Observe(TSN(10)); + reasm_->Add(TSN(10), gen_.Ordered({1, 2, 3, 4}, "BE", {.mid = MID(0)})); + + data_tracker_->Observe(TSN(11)); + reasm_->Add(TSN(11), gen_.Ordered({1, 2, 3, 4}, "BE", {.mid = MID(1)})); + + EXPECT_THAT( + reasm_->FlushMessages(), + UnorderedElementsAre(SctpMessageIs(kStreamId, PPID(53), kShortPayload), + SctpMessageIs(kStreamId, PPID(53), kShortPayload))); + + Parameters::Builder builder; + builder.Add(OutgoingSSNResetRequestParameter( + ReconfigRequestSN(10), ReconfigRequestSN(3), TSN(13), {kStreamId})); + EXPECT_THAT(HandleAndCatchResponse(ReConfigChunk(builder.Build())), + ElementsAre(Property(&ReconfigurationResponseParameter::result, + ResponseResult::kInProgress))); + + data_tracker_->Observe(TSN(15)); + reasm_->Add(TSN(15), gen_.Ordered({1, 2, 3, 4}, "BE", + {.mid = MID(1), .ppid = PPID(5)})); + + data_tracker_->Observe(TSN(14)); + reasm_->Add(TSN(14), gen_.Ordered({1, 2, 3, 4}, "BE", + {.mid = MID(0), .ppid = PPID(4)})); + + data_tracker_->Observe(TSN(13)); + reasm_->Add(TSN(13), gen_.Ordered({1, 2, 3, 4}, "BE", + {.mid = MID(3), .ppid = PPID(3)})); + + data_tracker_->Observe(TSN(12)); + reasm_->Add(TSN(12), gen_.Ordered({1, 2, 3, 4}, "BE", + {.mid = MID(2), .ppid = PPID(2)})); + + builder.Add(OutgoingSSNResetRequestParameter( + ReconfigRequestSN(11), ReconfigRequestSN(4), TSN(13), {kStreamId})); + EXPECT_THAT(HandleAndCatchResponse(ReConfigChunk(builder.Build())), + ElementsAre(Property(&ReconfigurationResponseParameter::result, + ResponseResult::kSuccessPerformed))); + + EXPECT_THAT( + reasm_->FlushMessages(), + UnorderedElementsAre(SctpMessageIs(kStreamId, PPID(2), kShortPayload), + SctpMessageIs(kStreamId, PPID(3), kShortPayload), + SctpMessageIs(kStreamId, PPID(4), kShortPayload), + SctpMessageIs(kStreamId, PPID(5), kShortPayload))); +} + +TEST_F(StreamResetHandlerTest, ResetStreamsDeferredOnlySelectedStreams) { + // This test verifies the receiving behavior of receiving messages on + // streams 1, 2 and 3, and receiving a reset request on stream 1, 2, causing + // deferred reset processing. + + // Reset stream 1,2 with "last assigned TSN=12" + Parameters::Builder builder; + builder.Add(OutgoingSSNResetRequestParameter(ReconfigRequestSN(10), + ReconfigRequestSN(3), TSN(12), + {StreamID(1), StreamID(2)})); + EXPECT_THAT(HandleAndCatchResponse(ReConfigChunk(builder.Build())), + ElementsAre(Property(&ReconfigurationResponseParameter::result, + ResponseResult::kInProgress))); + + // TSN 10, SID 1 - before TSN 12 -> deliver + data_tracker_->Observe(TSN(10)); + reasm_->Add(TSN(10), gen_.Ordered({1, 2, 3, 4}, "BE", + {.stream_id = StreamID(1), + .mid = MID(0), + .ppid = PPID(1001)})); + + // TSN 11, SID 2 - before TSN 12 -> deliver + data_tracker_->Observe(TSN(11)); + reasm_->Add(TSN(11), gen_.Ordered({1, 2, 3, 4}, "BE", + {.stream_id = StreamID(2), + .mid = MID(0), + .ppid = PPID(1002)})); + + // TSN 12, SID 3 - at TSN 12 -> deliver + data_tracker_->Observe(TSN(12)); + reasm_->Add(TSN(12), gen_.Ordered({1, 2, 3, 4}, "BE", + {.stream_id = StreamID(3), + .mid = MID(0), + .ppid = PPID(1003)})); + + // TSN 13, SID 1 - after TSN 12 and SID=1 -> defer + data_tracker_->Observe(TSN(13)); + reasm_->Add(TSN(13), gen_.Ordered({1, 2, 3, 4}, "BE", + {.stream_id = StreamID(1), + .mid = MID(0), + .ppid = PPID(1004)})); + + // TSN 14, SID 2 - after TSN 12 and SID=2 -> defer + data_tracker_->Observe(TSN(14)); + reasm_->Add(TSN(14), gen_.Ordered({1, 2, 3, 4}, "BE", + {.stream_id = StreamID(2), + .mid = MID(0), + .ppid = PPID(1005)})); + + // TSN 15, SID 3 - after TSN 12, but SID 3 is not reset -> deliver + data_tracker_->Observe(TSN(15)); + reasm_->Add(TSN(15), gen_.Ordered({1, 2, 3, 4}, "BE", + {.stream_id = StreamID(3), + .mid = MID(1), + .ppid = PPID(1006)})); + + EXPECT_THAT(reasm_->FlushMessages(), + UnorderedElementsAre( + SctpMessageIs(StreamID(1), PPID(1001), kShortPayload), + SctpMessageIs(StreamID(2), PPID(1002), kShortPayload), + SctpMessageIs(StreamID(3), PPID(1003), kShortPayload), + SctpMessageIs(StreamID(3), PPID(1006), kShortPayload))); + + builder.Add(OutgoingSSNResetRequestParameter(ReconfigRequestSN(11), + ReconfigRequestSN(3), TSN(13), + {StreamID(1), StreamID(2)})); + EXPECT_THAT(HandleAndCatchResponse(ReConfigChunk(builder.Build())), + ElementsAre(Property(&ReconfigurationResponseParameter::result, + ResponseResult::kSuccessPerformed))); + + EXPECT_THAT(reasm_->FlushMessages(), + UnorderedElementsAre( + SctpMessageIs(StreamID(1), PPID(1004), kShortPayload), + SctpMessageIs(StreamID(2), PPID(1005), kShortPayload))); +} + +TEST_F(StreamResetHandlerTest, ResetStreamsDefersForwardTsn) { + // This test verifies that FORWARD-TSNs are deferred if they want to move + // the cumulative ack TSN point past sender's last assigned TSN. + static constexpr StreamID kStreamId = StreamID(42); + + // Simulate sender sends: + // * TSN 10 (SSN=0, BE, lost), + // * TSN 11 (SSN=1, BE, lost), + // * TSN 12 (SSN=2, BE, lost) + // * RESET THE STREAM + // * TSN 13 (SSN=0, B, received) + // * TSN 14 (SSN=0, E, lost), + // * TSN 15 (SSN=1, BE, received) + Parameters::Builder builder; + builder.Add(OutgoingSSNResetRequestParameter( + ReconfigRequestSN(10), ReconfigRequestSN(3), TSN(12), {kStreamId})); + EXPECT_THAT(HandleAndCatchResponse(ReConfigChunk(builder.Build())), + ElementsAre(Property(&ReconfigurationResponseParameter::result, + ResponseResult::kInProgress))); + + // TSN 13, B, after TSN=12 -> defer + data_tracker_->Observe(TSN(13)); + reasm_->Add(TSN(13), + gen_.Ordered( + {1, 2, 3, 4}, "B", + {.stream_id = kStreamId, .mid = MID(0), .ppid = PPID(1004)})); + + // TSN 15, BE, after TSN=12 -> defer + data_tracker_->Observe(TSN(15)); + reasm_->Add(TSN(15), + gen_.Ordered( + {1, 2, 3, 4}, "BE", + {.stream_id = kStreamId, .mid = MID(1), .ppid = PPID(1005)})); + + // Time passes, sender decides to send FORWARD-TSN up to the RESET. + data_tracker_->HandleForwardTsn(TSN(12)); + reasm_->HandleForwardTsn( + TSN(12), std::vector<SkippedStream>({SkippedStream(kStreamId, SSN(2))})); + + // The receiver sends a SACK in response to that. The stream hasn't been + // reset yet, but the sender now decides that TSN=13-14 is to be skipped. + // As this has a TSN 14, after TSN=12 -> defer it. + data_tracker_->HandleForwardTsn(TSN(14)); + reasm_->HandleForwardTsn( + TSN(14), std::vector<SkippedStream>({SkippedStream(kStreamId, SSN(0))})); + + // Reset the stream -> deferred TSNs should be re-added. + builder.Add(OutgoingSSNResetRequestParameter( + ReconfigRequestSN(11), ReconfigRequestSN(3), TSN(12), {kStreamId})); + EXPECT_THAT(HandleAndCatchResponse(ReConfigChunk(builder.Build())), + ElementsAre(Property(&ReconfigurationResponseParameter::result, + ResponseResult::kSuccessPerformed))); + + EXPECT_THAT(reasm_->FlushMessages(), + UnorderedElementsAre( + SctpMessageIs(kStreamId, PPID(1005), kShortPayload))); +} + +TEST_F(StreamResetHandlerTest, SendOutgoingRequestDirectly) { + EXPECT_CALL(producer_, PrepareResetStream(StreamID(42))); + handler_->ResetStreams(std::vector<StreamID>({StreamID(42)})); + + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true)); + EXPECT_CALL(producer_, GetStreamsReadyToBeReset()) + .WillOnce(Return(std::vector<StreamID>({StreamID(42)}))); + + absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest(); + ASSERT_TRUE(reconfig.has_value()); + ASSERT_HAS_VALUE_AND_ASSIGN( + OutgoingSSNResetRequestParameter req, + reconfig->parameters().get<OutgoingSSNResetRequestParameter>()); + + EXPECT_EQ(req.request_sequence_number(), kMyInitialReqSn); + EXPECT_EQ(req.sender_last_assigned_tsn(), + TSN(*retransmission_queue_->next_tsn() - 1)); + EXPECT_THAT(req.stream_ids(), UnorderedElementsAre(StreamID(42))); +} + +TEST_F(StreamResetHandlerTest, ResetMultipleStreamsInOneRequest) { + EXPECT_CALL(producer_, PrepareResetStream(StreamID(40))); + EXPECT_CALL(producer_, PrepareResetStream(StreamID(41))); + EXPECT_CALL(producer_, PrepareResetStream(StreamID(42))).Times(2); + EXPECT_CALL(producer_, PrepareResetStream(StreamID(43))); + EXPECT_CALL(producer_, PrepareResetStream(StreamID(44))); + handler_->ResetStreams(std::vector<StreamID>({StreamID(42)})); + handler_->ResetStreams( + std::vector<StreamID>({StreamID(43), StreamID(44), StreamID(41)})); + handler_->ResetStreams(std::vector<StreamID>({StreamID(42), StreamID(40)})); + + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true)); + EXPECT_CALL(producer_, GetStreamsReadyToBeReset()) + .WillOnce(Return( + std::vector<StreamID>({StreamID(40), StreamID(41), StreamID(42), + StreamID(43), StreamID(44)}))); + absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest(); + ASSERT_TRUE(reconfig.has_value()); + ASSERT_HAS_VALUE_AND_ASSIGN( + OutgoingSSNResetRequestParameter req, + reconfig->parameters().get<OutgoingSSNResetRequestParameter>()); + + EXPECT_EQ(req.request_sequence_number(), kMyInitialReqSn); + EXPECT_EQ(req.sender_last_assigned_tsn(), + TSN(*retransmission_queue_->next_tsn() - 1)); + EXPECT_THAT(req.stream_ids(), + UnorderedElementsAre(StreamID(40), StreamID(41), StreamID(42), + StreamID(43), StreamID(44))); +} + +TEST_F(StreamResetHandlerTest, SendOutgoingRequestDeferred) { + EXPECT_CALL(producer_, PrepareResetStream(StreamID(42))); + handler_->ResetStreams(std::vector<StreamID>({StreamID(42)})); + + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()) + .WillOnce(Return(false)) + .WillOnce(Return(false)) + .WillOnce(Return(true)); + + EXPECT_FALSE(handler_->MakeStreamResetRequest().has_value()); + EXPECT_FALSE(handler_->MakeStreamResetRequest().has_value()); + EXPECT_TRUE(handler_->MakeStreamResetRequest().has_value()); +} + +TEST_F(StreamResetHandlerTest, SendOutgoingResettingOnPositiveResponse) { + EXPECT_CALL(producer_, PrepareResetStream(StreamID(42))); + handler_->ResetStreams(std::vector<StreamID>({StreamID(42)})); + + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true)); + EXPECT_CALL(producer_, GetStreamsReadyToBeReset()) + .WillOnce(Return(std::vector<StreamID>({StreamID(42)}))); + + absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest(); + ASSERT_TRUE(reconfig.has_value()); + ASSERT_HAS_VALUE_AND_ASSIGN( + OutgoingSSNResetRequestParameter req, + reconfig->parameters().get<OutgoingSSNResetRequestParameter>()); + + Parameters::Builder builder; + builder.Add(ReconfigurationResponseParameter( + req.request_sequence_number(), ResponseResult::kSuccessPerformed)); + ReConfigChunk response_reconfig(builder.Build()); + + EXPECT_CALL(producer_, CommitResetStreams); + EXPECT_CALL(producer_, RollbackResetStreams).Times(0); + + // Processing a response shouldn't result in sending anything. + EXPECT_CALL(callbacks_, OnError).Times(0); + EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0); + handler_->HandleReConfig(std::move(response_reconfig)); +} + +TEST_F(StreamResetHandlerTest, SendOutgoingResetRollbackOnError) { + EXPECT_CALL(producer_, PrepareResetStream(StreamID(42))); + handler_->ResetStreams(std::vector<StreamID>({StreamID(42)})); + + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true)); + EXPECT_CALL(producer_, GetStreamsReadyToBeReset()) + .WillOnce(Return(std::vector<StreamID>({StreamID(42)}))); + + absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest(); + ASSERT_TRUE(reconfig.has_value()); + ASSERT_HAS_VALUE_AND_ASSIGN( + OutgoingSSNResetRequestParameter req, + reconfig->parameters().get<OutgoingSSNResetRequestParameter>()); + + Parameters::Builder builder; + builder.Add(ReconfigurationResponseParameter( + req.request_sequence_number(), ResponseResult::kErrorBadSequenceNumber)); + ReConfigChunk response_reconfig(builder.Build()); + + EXPECT_CALL(producer_, CommitResetStreams).Times(0); + EXPECT_CALL(producer_, RollbackResetStreams); + + // Only requests should result in sending responses. + EXPECT_CALL(callbacks_, OnError).Times(0); + EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0); + handler_->HandleReConfig(std::move(response_reconfig)); +} + +TEST_F(StreamResetHandlerTest, SendOutgoingResetRetransmitOnInProgress) { + static constexpr StreamID kStreamToReset = StreamID(42); + + EXPECT_CALL(producer_, PrepareResetStream(kStreamToReset)); + handler_->ResetStreams(std::vector<StreamID>({kStreamToReset})); + + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true)); + EXPECT_CALL(producer_, GetStreamsReadyToBeReset()) + .WillOnce(Return(std::vector<StreamID>({kStreamToReset}))); + + absl::optional<ReConfigChunk> reconfig1 = handler_->MakeStreamResetRequest(); + ASSERT_TRUE(reconfig1.has_value()); + ASSERT_HAS_VALUE_AND_ASSIGN( + OutgoingSSNResetRequestParameter req1, + reconfig1->parameters().get<OutgoingSSNResetRequestParameter>()); + + // Simulate that the peer responded "In Progress". + Parameters::Builder builder; + builder.Add(ReconfigurationResponseParameter(req1.request_sequence_number(), + ResponseResult::kInProgress)); + ReConfigChunk response_reconfig(builder.Build()); + + EXPECT_CALL(producer_, CommitResetStreams()).Times(0); + EXPECT_CALL(producer_, RollbackResetStreams()).Times(0); + + // Processing a response shouldn't result in sending anything. + EXPECT_CALL(callbacks_, OnError).Times(0); + EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0); + handler_->HandleReConfig(std::move(response_reconfig)); + + // Let some time pass, so that the reconfig timer expires, and retries the + // same request. + EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(1); + AdvanceTime(kRto); + + std::vector<uint8_t> payload = callbacks_.ConsumeSentPacket(); + ASSERT_FALSE(payload.empty()); + + ASSERT_HAS_VALUE_AND_ASSIGN(SctpPacket packet, + SctpPacket::Parse(payload, DcSctpOptions())); + ASSERT_THAT(packet.descriptors(), SizeIs(1)); + ASSERT_HAS_VALUE_AND_ASSIGN( + ReConfigChunk reconfig2, + ReConfigChunk::Parse(packet.descriptors()[0].data)); + + ASSERT_HAS_VALUE_AND_ASSIGN( + OutgoingSSNResetRequestParameter req2, + reconfig2.parameters().get<OutgoingSSNResetRequestParameter>()); + + EXPECT_EQ(req2.request_sequence_number(), + AddTo(req1.request_sequence_number(), 1)); + EXPECT_THAT(req2.stream_ids(), UnorderedElementsAre(kStreamToReset)); +} + +TEST_F(StreamResetHandlerTest, ResetWhileRequestIsSentWillQueue) { + EXPECT_CALL(producer_, PrepareResetStream(StreamID(42))); + handler_->ResetStreams(std::vector<StreamID>({StreamID(42)})); + + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true)); + EXPECT_CALL(producer_, GetStreamsReadyToBeReset()) + .WillOnce(Return(std::vector<StreamID>({StreamID(42)}))); + + absl::optional<ReConfigChunk> reconfig1 = handler_->MakeStreamResetRequest(); + ASSERT_TRUE(reconfig1.has_value()); + ASSERT_HAS_VALUE_AND_ASSIGN( + OutgoingSSNResetRequestParameter req1, + reconfig1->parameters().get<OutgoingSSNResetRequestParameter>()); + EXPECT_EQ(req1.request_sequence_number(), kMyInitialReqSn); + EXPECT_EQ(req1.sender_last_assigned_tsn(), + AddTo(retransmission_queue_->next_tsn(), -1)); + EXPECT_THAT(req1.stream_ids(), UnorderedElementsAre(StreamID(42))); + + // Streams reset while the request is in-flight will be queued. + EXPECT_CALL(producer_, PrepareResetStream(StreamID(41))); + EXPECT_CALL(producer_, PrepareResetStream(StreamID(43))); + StreamID stream_ids[] = {StreamID(41), StreamID(43)}; + handler_->ResetStreams(stream_ids); + EXPECT_EQ(handler_->MakeStreamResetRequest(), absl::nullopt); + + Parameters::Builder builder; + builder.Add(ReconfigurationResponseParameter( + req1.request_sequence_number(), ResponseResult::kSuccessPerformed)); + ReConfigChunk response_reconfig(builder.Build()); + + EXPECT_CALL(producer_, CommitResetStreams()).Times(1); + EXPECT_CALL(producer_, RollbackResetStreams()).Times(0); + + // Processing a response shouldn't result in sending anything. + EXPECT_CALL(callbacks_, OnError).Times(0); + EXPECT_CALL(callbacks_, SendPacketWithStatus).Times(0); + handler_->HandleReConfig(std::move(response_reconfig)); + + // Response has been processed. A new request can be sent. + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true)); + EXPECT_CALL(producer_, GetStreamsReadyToBeReset()) + .WillOnce(Return(std::vector<StreamID>({StreamID(41), StreamID(43)}))); + + absl::optional<ReConfigChunk> reconfig2 = handler_->MakeStreamResetRequest(); + ASSERT_TRUE(reconfig2.has_value()); + ASSERT_HAS_VALUE_AND_ASSIGN( + OutgoingSSNResetRequestParameter req2, + reconfig2->parameters().get<OutgoingSSNResetRequestParameter>()); + EXPECT_EQ(req2.request_sequence_number(), AddTo(kMyInitialReqSn, 1)); + EXPECT_EQ(req2.sender_last_assigned_tsn(), + TSN(*retransmission_queue_->next_tsn() - 1)); + EXPECT_THAT(req2.stream_ids(), + UnorderedElementsAre(StreamID(41), StreamID(43))); +} + +TEST_F(StreamResetHandlerTest, SendIncomingResetJustReturnsNothingPerformed) { + Parameters::Builder builder; + builder.Add( + IncomingSSNResetRequestParameter(kPeerInitialReqSn, {StreamID(1)})); + + std::vector<ReconfigurationResponseParameter> responses = + HandleAndCatchResponse(ReConfigChunk(builder.Build())); + ASSERT_THAT(responses, SizeIs(1)); + EXPECT_THAT(responses[0].response_sequence_number(), kPeerInitialReqSn); + EXPECT_THAT(responses[0].result(), ResponseResult::kSuccessNothingToDo); +} + +TEST_F(StreamResetHandlerTest, SendSameRequestTwiceIsIdempotent) { + // Simulate that receiving the same chunk twice (due to network issues, + // or retransmissions, causing a RECONFIG to be re-received) is idempotent. + for (int i = 0; i < 2; ++i) { + Parameters::Builder builder; + builder.Add(OutgoingSSNResetRequestParameter( + kPeerInitialReqSn, ReconfigRequestSN(3), AddTo(kPeerInitialTsn, 1), + {StreamID(1)})); + + std::vector<ReconfigurationResponseParameter> responses1 = + HandleAndCatchResponse(ReConfigChunk(builder.Build())); + EXPECT_THAT(responses1, SizeIs(1)); + EXPECT_EQ(responses1[0].result(), ResponseResult::kInProgress); + } +} + +TEST_F(StreamResetHandlerTest, + HandoverIsAllowedOnlyWhenNoStreamIsBeingOrWillBeReset) { + EXPECT_CALL(producer_, PrepareResetStream(StreamID(42))); + handler_->ResetStreams(std::vector<StreamID>({StreamID(42)})); + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true)); + EXPECT_EQ( + handler_->GetHandoverReadiness(), + HandoverReadinessStatus(HandoverUnreadinessReason::kPendingStreamReset)); + + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()) + .WillOnce(Return(true)) + .WillOnce(Return(false)); + EXPECT_CALL(producer_, GetStreamsReadyToBeReset()) + .WillOnce(Return(std::vector<StreamID>({StreamID(42)}))); + + ASSERT_TRUE(handler_->MakeStreamResetRequest().has_value()); + EXPECT_EQ(handler_->GetHandoverReadiness(), + HandoverReadinessStatus( + HandoverUnreadinessReason::kPendingStreamResetRequest)); + + // Reset more streams while the request is in-flight. + EXPECT_CALL(producer_, PrepareResetStream(StreamID(41))); + EXPECT_CALL(producer_, PrepareResetStream(StreamID(43))); + StreamID stream_ids[] = {StreamID(41), StreamID(43)}; + handler_->ResetStreams(stream_ids); + + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true)); + EXPECT_EQ(handler_->GetHandoverReadiness(), + HandoverReadinessStatus() + .Add(HandoverUnreadinessReason::kPendingStreamResetRequest) + .Add(HandoverUnreadinessReason::kPendingStreamReset)); + + // Processing a response to first request. + EXPECT_CALL(producer_, CommitResetStreams()).Times(1); + handler_->HandleReConfig( + ReConfigChunk(Parameters::Builder() + .Add(ReconfigurationResponseParameter( + kMyInitialReqSn, ResponseResult::kSuccessPerformed)) + .Build())); + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true)); + EXPECT_EQ( + handler_->GetHandoverReadiness(), + HandoverReadinessStatus(HandoverUnreadinessReason::kPendingStreamReset)); + + // Second request can be sent. + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()) + .WillOnce(Return(true)) + .WillOnce(Return(false)); + EXPECT_CALL(producer_, GetStreamsReadyToBeReset()) + .WillOnce(Return(std::vector<StreamID>({StreamID(41), StreamID(43)}))); + + ASSERT_TRUE(handler_->MakeStreamResetRequest().has_value()); + EXPECT_EQ(handler_->GetHandoverReadiness(), + HandoverReadinessStatus( + HandoverUnreadinessReason::kPendingStreamResetRequest)); + + // Processing a response to second request. + EXPECT_CALL(producer_, CommitResetStreams()).Times(1); + handler_->HandleReConfig(ReConfigChunk( + Parameters::Builder() + .Add(ReconfigurationResponseParameter( + AddTo(kMyInitialReqSn, 1), ResponseResult::kSuccessPerformed)) + .Build())); + + // Seconds response has been processed. No pending resets. + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(false)); + + EXPECT_TRUE(handler_->GetHandoverReadiness().IsReady()); +} + +TEST_F(StreamResetHandlerTest, HandoverInInitialState) { + PerformHandover(); + + EXPECT_CALL(producer_, PrepareResetStream(StreamID(42))); + handler_->ResetStreams(std::vector<StreamID>({StreamID(42)})); + + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true)); + EXPECT_CALL(producer_, GetStreamsReadyToBeReset()) + .WillOnce(Return(std::vector<StreamID>({StreamID(42)}))); + + absl::optional<ReConfigChunk> reconfig = handler_->MakeStreamResetRequest(); + ASSERT_TRUE(reconfig.has_value()); + ASSERT_HAS_VALUE_AND_ASSIGN( + OutgoingSSNResetRequestParameter req, + reconfig->parameters().get<OutgoingSSNResetRequestParameter>()); + + EXPECT_EQ(req.request_sequence_number(), kMyInitialReqSn); + EXPECT_EQ(req.sender_last_assigned_tsn(), + TSN(*retransmission_queue_->next_tsn() - 1)); + EXPECT_THAT(req.stream_ids(), UnorderedElementsAre(StreamID(42))); +} + +TEST_F(StreamResetHandlerTest, HandoverAfterHavingResetOneStream) { + // Reset one stream + { + EXPECT_CALL(producer_, PrepareResetStream(StreamID(42))); + handler_->ResetStreams(std::vector<StreamID>({StreamID(42)})); + + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()) + .WillOnce(Return(true)) + .WillOnce(Return(false)); + EXPECT_CALL(producer_, GetStreamsReadyToBeReset()) + .WillOnce(Return(std::vector<StreamID>({StreamID(42)}))); + + ASSERT_HAS_VALUE_AND_ASSIGN(ReConfigChunk reconfig, + handler_->MakeStreamResetRequest()); + ASSERT_HAS_VALUE_AND_ASSIGN( + OutgoingSSNResetRequestParameter req, + reconfig.parameters().get<OutgoingSSNResetRequestParameter>()); + EXPECT_EQ(req.request_sequence_number(), kMyInitialReqSn); + EXPECT_EQ(req.sender_last_assigned_tsn(), + TSN(*retransmission_queue_->next_tsn() - 1)); + EXPECT_THAT(req.stream_ids(), UnorderedElementsAre(StreamID(42))); + + EXPECT_CALL(producer_, CommitResetStreams()).Times(1); + handler_->HandleReConfig( + ReConfigChunk(Parameters::Builder() + .Add(ReconfigurationResponseParameter( + req.request_sequence_number(), + ResponseResult::kSuccessPerformed)) + .Build())); + } + + PerformHandover(); + + // Reset another stream after handover + { + EXPECT_CALL(producer_, PrepareResetStream(StreamID(43))); + handler_->ResetStreams(std::vector<StreamID>({StreamID(43)})); + + EXPECT_CALL(producer_, HasStreamsReadyToBeReset()).WillOnce(Return(true)); + EXPECT_CALL(producer_, GetStreamsReadyToBeReset()) + .WillOnce(Return(std::vector<StreamID>({StreamID(43)}))); + + ASSERT_HAS_VALUE_AND_ASSIGN(ReConfigChunk reconfig, + handler_->MakeStreamResetRequest()); + ASSERT_HAS_VALUE_AND_ASSIGN( + OutgoingSSNResetRequestParameter req, + reconfig.parameters().get<OutgoingSSNResetRequestParameter>()); + + EXPECT_EQ(req.request_sequence_number(), + ReconfigRequestSN(kMyInitialReqSn.value() + 1)); + EXPECT_EQ(req.sender_last_assigned_tsn(), + TSN(*retransmission_queue_->next_tsn() - 1)); + EXPECT_THAT(req.stream_ids(), UnorderedElementsAre(StreamID(43))); + } +} + +TEST_F(StreamResetHandlerTest, PerformCloseAfterOneFirstFailing) { + // Inject a stream reset on the first expected TSN (which hasn't been seen). + Parameters::Builder builder; + builder.Add(OutgoingSSNResetRequestParameter( + kPeerInitialReqSn, ReconfigRequestSN(3), kPeerInitialTsn, {StreamID(1)})); + + // The socket is expected to say "in progress" as that TSN hasn't been seen. + std::vector<ReconfigurationResponseParameter> responses = + HandleAndCatchResponse(ReConfigChunk(builder.Build())); + EXPECT_THAT(responses, SizeIs(1)); + EXPECT_EQ(responses[0].result(), ResponseResult::kInProgress); + + // Let the socket receive the TSN. + DataGeneratorOptions opts; + opts.mid = MID(0); + reasm_->Add(kPeerInitialTsn, gen_.Ordered({1, 2, 3, 4}, "BE", opts)); + data_tracker_->Observe(kPeerInitialTsn); + + // And emulate that time has passed, and the peer retries the stream reset, + // but now with an incremented request sequence number. + Parameters::Builder builder2; + builder2.Add(OutgoingSSNResetRequestParameter( + ReconfigRequestSN(*kPeerInitialReqSn + 1), ReconfigRequestSN(3), + kPeerInitialTsn, {StreamID(1)})); + + // This is supposed to be handled well. + std::vector<ReconfigurationResponseParameter> responses2 = + HandleAndCatchResponse(ReConfigChunk(builder2.Build())); + EXPECT_THAT(responses2, SizeIs(1)); + EXPECT_EQ(responses2[0].result(), ResponseResult::kSuccessPerformed); +} +} // namespace +} // namespace dcsctp |