dcsctp: Don't generate FORWARD-TSN across stream resets

This was a fun bug which proved to be challenging to find a good
solution for. The issue comes from the combination of partial
reliability and stream resetting, which are covered in different RFCs,
and where they don't refer to each other...

Stream resetting (RFC 6525) is used in WebRTC for closing a Data
Channel, and is done by signaling to the receiver that the stream
sequence number (SSN) should be set to zero (0) at some time. Partial
reliability (RFC 3758) - and expiring messages that will not be
retransmitted - is done by signaling that the SSN should be set to a
certain value at a certain TSN, as the messages up until the provided
SSN are not to be expected to be sent again.

As these two functionalities both work by signaling to the receiver
what the next expected SSN should be, they need to do it correctly not
to overwrite each others' intent. And here was the bug. An example
scenario where this caused issues, where we are Z (the receiver),
getting packets from the sender (A):

 5  A->Z          DATA (TSN=30, B, SID=2, SSN=0)
 6          Z->A  SACK (Ack=30)
 7  A->Z          DATA (TSN=31, E, SID=2, SSN=0)
 8  A->Z          RE_CONFIG (REQ=30, TSN=31, SID=2)
 9          Z->A  RE_CONFIG (RESP=30, Performed)
10          Z->A  SACK (Ack=31)
11  A->Z          DATA (TSN=32, SID=1)
12  A->Z          FORWARD_TSN (TSN=32, SID=2, SSN=0)

Let's assume that the path Z->A had packet loss and A never really
received our responses (#6, #9, #10) in time.

At #5, Z receives a DATA fragment, which it acks, and at #7 the end of
that message. The stream is then reset (#8) which it signals that it
was performed (#9) and acked (#10), and data on another stream (2) was
received (#11). Since A hasn't received any ACKS yet, and those chunks
on SID=2 all expired, A sends a FORWARD-TSN saying that "Skip to TSN=32,
and don't expect SID=2, SSN=0". That makes the receiver expect the SSN
on SID=2 to be SSN=1 next time at TSN=32.

But that's not good at all - A reset the stream at #8 and will want to
send the next message on SID=2 using SSN=0 - not 1. The FORWARD-TSN
clearly can't have a TSN that is beyond the stream reset TSN for that
stream.

This is just one example - combining stream resetting and partial
reliability, together with a lossy network, and different variants of
this can occur, which results in the receiver possibly not delivering
packets because it expects a different SSN than the one the sender is
later using.

So this CL adds "breakpoints" to how far a FORWARD-TSN can stretch. It
will simply not cross any Stream Reset last assigned TSNs, and only when
a receiver has acked that all TSNs up till the Stream Reset last
assigned TSN has been received, it will proceed expiring chunks after
that.

Bug: webrtc:14600
Change-Id: Ibae8c9308f5dfe8d734377d42cce653e69e95731
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/321600
Commit-Queue: Victor Boivie <[email protected]>
Reviewed-by: Harald Alvestrand <[email protected]>
Cr-Commit-Position: refs/heads/main@{#40829}

net/dcsctp/socket/stream_reset_handler.cc

@@ -278,8 +278,8 @@ absl::optional<ReConfigChunk> StreamResetHandler::MakeStreamResetRequest() {
     return absl::nullopt;
   }
 
-  current_request_.emplace(TSN(*retransmission_queue_->next_tsn() - 1),
-                           retransmission_queue_->GetStreamsReadyToBeReset());
+  current_request_.emplace(retransmission_queue_->last_assigned_tsn(),
+                           retransmission_queue_->BeginResetStreams());
   reconfig_timer_->set_duration(ctx_->current_rto());
   reconfig_timer_->Start();
   return MakeReconfigChunk();

net/dcsctp/tx/BUILD.gn

@@ -104,6 +104,7 @@ rtc_library("outstanding_data") {
     "../../../api:array_view",
     "../../../rtc_base:checks",
     "../../../rtc_base:logging",
+    "../../../rtc_base/containers:flat_set",
     "../common:math",
     "../common:sequence_numbers",
     "../common:str_join",

net/dcsctp/tx/outstanding_data.cc

@@ -159,6 +159,9 @@ void OutstandingData::RemoveAcked(UnwrappedTSN cumulative_tsn_ack,
 
   outstanding_data_.erase(outstanding_data_.begin(), first_unacked);
   last_cumulative_tsn_ack_ = cumulative_tsn_ack;
+  stream_reset_breakpoint_tsns_.erase(stream_reset_breakpoint_tsns_.begin(),
+                                      stream_reset_breakpoint_tsns_.upper_bound(
+                                          cumulative_tsn_ack.next_value()));
 }
 
 void OutstandingData::AckGapBlocks(
@@ -487,7 +490,8 @@ ForwardTsnChunk OutstandingData::CreateForwardTsn() const {
   UnwrappedTSN new_cumulative_ack = last_cumulative_tsn_ack_;
 
   for (const auto& [tsn, item] : outstanding_data_) {
-    if ((tsn != new_cumulative_ack.next_value()) || !item.is_abandoned()) {
+    if (stream_reset_breakpoint_tsns_.contains(tsn) ||
+        (tsn != new_cumulative_ack.next_value()) || !item.is_abandoned()) {
       break;
     }
     new_cumulative_ack = tsn;
@@ -510,7 +514,8 @@ IForwardTsnChunk OutstandingData::CreateIForwardTsn() const {
   UnwrappedTSN new_cumulative_ack = last_cumulative_tsn_ack_;
 
   for (const auto& [tsn, item] : outstanding_data_) {
-    if ((tsn != new_cumulative_ack.next_value()) || !item.is_abandoned()) {
+    if (stream_reset_breakpoint_tsns_.contains(tsn) ||
+        (tsn != new_cumulative_ack.next_value()) || !item.is_abandoned()) {
       break;
     }
     new_cumulative_ack = tsn;
@@ -540,4 +545,8 @@ void OutstandingData::ResetSequenceNumbers(UnwrappedTSN next_tsn,
   next_tsn_ = next_tsn;
   last_cumulative_tsn_ack_ = last_cumulative_tsn;
 }
+
+void OutstandingData::BeginResetStreams() {
+  stream_reset_breakpoint_tsns_.insert(next_tsn_);
+}
 }  // namespace dcsctp

net/dcsctp/tx/outstanding_data.h

@@ -22,6 +22,7 @@
 #include "net/dcsctp/packet/chunk/sack_chunk.h"
 #include "net/dcsctp/packet/data.h"
 #include "net/dcsctp/public/types.h"
+#include "rtc_base/containers/flat_set.h"
 
 namespace dcsctp {
 
@@ -159,6 +160,10 @@ class OutstandingData {
   void ResetSequenceNumbers(UnwrappedTSN next_tsn,
                             UnwrappedTSN last_cumulative_tsn);
 
+  // Called when an outgoing stream reset is sent, marking the last assigned TSN
+  // as a breakpoint that a FORWARD-TSN shouldn't cross.
+  void BeginResetStreams();
+
  private:
   // A fragmented message's DATA chunk while in the retransmission queue, and
   // its associated metadata.
@@ -345,6 +350,10 @@ class OutstandingData {
   std::set<UnwrappedTSN> to_be_fast_retransmitted_;
   // Data chunks that are to be retransmitted.
   std::set<UnwrappedTSN> to_be_retransmitted_;
+  // Wben a stream reset has begun, the "next TSN to assign" is added to this
+  // set, and removed when the cum-ack TSN reaches it. This is used to limit a
+  // FORWARD-TSN to reset streams past a "stream reset last assigned TSN".
+  webrtc::flat_set<UnwrappedTSN> stream_reset_breakpoint_tsns_;
 };
 }  // namespace dcsctp
 #endif  // NET_DCSCTP_TX_OUTSTANDING_DATA_H_

net/dcsctp/tx/outstanding_data_test.cc

@@ -27,11 +27,14 @@ namespace {
 using ::testing::MockFunction;
 using State = ::dcsctp::OutstandingData::State;
 using ::testing::_;
+using ::testing::AllOf;
 using ::testing::ElementsAre;
 using ::testing::IsEmpty;
 using ::testing::Pair;
+using ::testing::Property;
 using ::testing::Return;
 using ::testing::StrictMock;
+using ::testing::UnorderedElementsAre;
 
 constexpr TimeMs kNow(42);
 
@@ -587,5 +590,97 @@ TEST_F(OutstandingDataTest, LifecycleReturnsAbandonedAfterT3rtxExpired) {
   EXPECT_FALSE(ack2.has_packet_loss);
   EXPECT_THAT(ack2.abandoned_lifecycle_ids, ElementsAre(LifecycleId(42)));
 }
+
+TEST_F(OutstandingDataTest, GeneratesForwardTsnUntilNextStreamResetTsn) {
+  // This test generates:
+  // * Stream 1: TSN 10, 11, 12 <RESET>
+  // * Stream 2: TSN 13, 14 <RESET>
+  // * Stream 3: TSN 15, 16
+  //
+  // Then it expires chunk 12-15, and ensures that the generated FORWARD-TSN
+  // only includes up till TSN 12 until the cum ack TSN has reached 12, and then
+  // 13 and 14 are included, and then after the cum ack TSN has reached 14, then
+  // 15 is included.
+  //
+  // What it shouldn't do, is to generate a FORWARD-TSN directly at the start
+  // with new TSN=15, and setting [(sid=1, ssn=44), (sid=2, ssn=46),
+  // (sid=3, ssn=47)], because that will confuse the receiver at TSN=17,
+  // receiving SID=1, SSN=0 (it's reset!), expecting SSN to be 45.
+  constexpr DataGeneratorOptions kStream1 = {.stream_id = StreamID(1)};
+  constexpr DataGeneratorOptions kStream2 = {.stream_id = StreamID(2)};
+  constexpr DataGeneratorOptions kStream3 = {.stream_id = StreamID(3)};
+  EXPECT_CALL(on_discard_, Call).WillRepeatedly(Return(false));
+
+  // TSN 10-12
+  buf_.Insert(gen_.Ordered({1}, "BE", kStream1), kNow);
+  buf_.Insert(gen_.Ordered({1}, "BE", kStream1), kNow);
+  buf_.Insert(gen_.Ordered({1}, "BE", kStream1), kNow, MaxRetransmits(0));
+
+  buf_.BeginResetStreams();
+
+  // TSN 13, 14
+  buf_.Insert(gen_.Ordered({1}, "BE", kStream2), kNow, MaxRetransmits(0));
+  buf_.Insert(gen_.Ordered({1}, "BE", kStream2), kNow, MaxRetransmits(0));
+
+  buf_.BeginResetStreams();
+
+  // TSN 15, 16
+  buf_.Insert(gen_.Ordered({1}, "BE", kStream3), kNow, MaxRetransmits(0));
+  buf_.Insert(gen_.Ordered({1}, "BE", kStream3), kNow);
+
+  EXPECT_FALSE(buf_.ShouldSendForwardTsn());
+
+  buf_.HandleSack(unwrapper_.Unwrap(TSN(11)), {}, false);
+  buf_.NackAll();
+  EXPECT_THAT(buf_.GetChunkStatesForTesting(),
+              ElementsAre(Pair(TSN(11), State::kAcked),      //
+                          Pair(TSN(12), State::kAbandoned),  //
+                          Pair(TSN(13), State::kAbandoned),  //
+                          Pair(TSN(14), State::kAbandoned),  //
+                          Pair(TSN(15), State::kAbandoned),  //
+                          Pair(TSN(16), State::kToBeRetransmitted)));
+
+  EXPECT_TRUE(buf_.ShouldSendForwardTsn());
+  EXPECT_THAT(
+      buf_.CreateForwardTsn(),
+      AllOf(Property(&ForwardTsnChunk::new_cumulative_tsn, TSN(12)),
+            Property(&ForwardTsnChunk::skipped_streams,
+                     UnorderedElementsAre(ForwardTsnChunk::SkippedStream(
+                         StreamID(1), SSN(44))))));
+
+  // Ack 12, allowing a FORWARD-TSN that spans to TSN=14 to be created.
+  buf_.HandleSack(unwrapper_.Unwrap(TSN(12)), {}, false);
+  EXPECT_TRUE(buf_.ShouldSendForwardTsn());
+  EXPECT_THAT(
+      buf_.CreateForwardTsn(),
+      AllOf(Property(&ForwardTsnChunk::new_cumulative_tsn, TSN(14)),
+            Property(&ForwardTsnChunk::skipped_streams,
+                     UnorderedElementsAre(ForwardTsnChunk::SkippedStream(
+                         StreamID(2), SSN(46))))));
+
+  // Ack 13, allowing a FORWARD-TSN that spans to TSN=14 to be created.
+  buf_.HandleSack(unwrapper_.Unwrap(TSN(13)), {}, false);
+  EXPECT_TRUE(buf_.ShouldSendForwardTsn());
+  EXPECT_THAT(
+      buf_.CreateForwardTsn(),
+      AllOf(Property(&ForwardTsnChunk::new_cumulative_tsn, TSN(14)),
+            Property(&ForwardTsnChunk::skipped_streams,
+                     UnorderedElementsAre(ForwardTsnChunk::SkippedStream(
+                         StreamID(2), SSN(46))))));
+
+  // Ack 14, allowing a FORWARD-TSN that spans to TSN=15 to be created.
+  buf_.HandleSack(unwrapper_.Unwrap(TSN(14)), {}, false);
+  EXPECT_TRUE(buf_.ShouldSendForwardTsn());
+  EXPECT_THAT(
+      buf_.CreateForwardTsn(),
+      AllOf(Property(&ForwardTsnChunk::new_cumulative_tsn, TSN(15)),
+            Property(&ForwardTsnChunk::skipped_streams,
+                     UnorderedElementsAre(ForwardTsnChunk::SkippedStream(
+                         StreamID(3), SSN(47))))));
+
+  buf_.HandleSack(unwrapper_.Unwrap(TSN(15)), {}, false);
+  EXPECT_FALSE(buf_.ShouldSendForwardTsn());
+}
+
 }  // namespace
 }  // namespace dcsctp

net/dcsctp/tx/retransmission_queue.cc

@@ -573,6 +573,10 @@ void RetransmissionQueue::PrepareResetStream(StreamID stream_id) {
 bool RetransmissionQueue::HasStreamsReadyToBeReset() const {
   return send_queue_.HasStreamsReadyToBeReset();
 }
+std::vector<StreamID> RetransmissionQueue::BeginResetStreams() {
+  outstanding_data_.BeginResetStreams();
+  return send_queue_.GetStreamsReadyToBeReset();
+}
 void RetransmissionQueue::CommitResetStreams() {
   send_queue_.CommitResetStreams();
 }

net/dcsctp/tx/retransmission_queue.h

@@ -103,6 +103,10 @@ class RetransmissionQueue {
   // Returns the next TSN that will be allocated for sent DATA chunks.
   TSN next_tsn() const { return outstanding_data_.next_tsn().Wrap(); }
 
+  TSN last_assigned_tsn() const {
+    return UnwrappedTSN::AddTo(outstanding_data_.next_tsn(), -1).Wrap();
+  }
+
   // Returns the size of the congestion window, in bytes. This is the number of
   // bytes that may be in-flight.
   size_t cwnd() const { return cwnd_; }
@@ -148,9 +152,7 @@ class RetransmissionQueue {
   // to stream resetting.
   void PrepareResetStream(StreamID stream_id);
   bool HasStreamsReadyToBeReset() const;
-  std::vector<StreamID> GetStreamsReadyToBeReset() const {
-    return send_queue_.GetStreamsReadyToBeReset();
-  }
+  std::vector<StreamID> BeginResetStreams();
   void CommitResetStreams();
   void RollbackResetStreams();