Fixing bug with PseudoTcp that corrupts data if initial packet is lost.

The issue occurred if a control segment is received after a non-control
segment received out-of-order, which only happens if:
* The initial "connect" segment is lost, and retransmitted later.
* Both sides send "connect"s simultaneously (rather than having
  designated server/client roles), such that the local side thinks a
  connection is established even before its "connect" has been
  acknowledged.
* Nagle algorithm disabled, allowing a data segment to be sent before
  the "connect" has been acknowledged.

This may seem like a pretty specific set of circumstances, but it can
happen with chromoting.

See the linked bug for more details.

Bug: webrtc:9208
Change-Id: I3cfe26e02158fcc5843f32d4e2ef7c511d58d9c9
Reviewed-on: https://webrtc-review.googlesource.com/78861
Reviewed-by: Sergey Ulanov <sergeyu@google.com>
Reviewed-by: Sergey Ulanov <sergeyu@chromium.org>
Commit-Queue: Taylor Brandstetter <deadbeef@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#23477}

p2p/base/pseudotcp.cc

@@ -899,9 +899,28 @@ bool PseudoTcp::process(Segment& seg) {
   bool bNewData = false;
 
   if (seg.len > 0) {
+    bool bRecover = false;
     if (bIgnoreData) {
       if (seg.seq == m_rcv_nxt) {
         m_rcv_nxt += seg.len;
+        // If we received a data segment out of order relative to a control
+        // segment, then we wrote it into the receive buffer at an offset (see
+        // "WriteOffset") below. So we need to advance the position in the
+        // buffer to avoid corrupting data. See bugs.webrtc.org/9208
+        //
+        // We advance the position in the buffer by N bytes by acting like we
+        // wrote N bytes and then immediately read them. We can only do this if
+        // there's not already data ready to read, but this should always be
+        // true in the problematic scenario, since control frames are always
+        // sent first in the stream.
+        size_t rcv_buffered;
+        if (m_rbuf.GetBuffered(&rcv_buffered) && rcv_buffered == 0) {
+          m_rbuf.ConsumeWriteBuffer(seg.len);
+          m_rbuf.ConsumeReadData(seg.len);
+          // After shifting the position in the buffer, we may have
+          // out-of-order packets ready to be recovered.
+          bRecover = true;
+        }
       }
     } else {
       uint32_t nOffset = seg.seq - m_rcv_nxt;
@@ -920,23 +939,9 @@ bool PseudoTcp::process(Segment& seg) {
         m_rcv_nxt += seg.len;
         m_rcv_wnd -= seg.len;
         bNewData = true;
-
+        // May be able to recover packets previously received out-of-order
-        RList::iterator it = m_rlist.begin();
+        // now.
-        while ((it != m_rlist.end()) && (it->seq <= m_rcv_nxt)) {
+        bRecover = true;
-          if (it->seq + it->len > m_rcv_nxt) {
-            sflags = sfImmediateAck;  // (Fast Recovery)
-            uint32_t nAdjust = (it->seq + it->len) - m_rcv_nxt;
-#if _DEBUGMSG >= _DBG_NORMAL
-            RTC_LOG(LS_INFO)
-                << "Recovered " << nAdjust << " bytes (" << m_rcv_nxt << " -> "
-                << m_rcv_nxt + nAdjust << ")";
-#endif  // _DEBUGMSG
-            m_rbuf.ConsumeWriteBuffer(nAdjust);
-            m_rcv_nxt += nAdjust;
-            m_rcv_wnd -= nAdjust;
-          }
-          it = m_rlist.erase(it);
-        }
       } else {
 #if _DEBUGMSG >= _DBG_NORMAL
         RTC_LOG(LS_INFO) << "Saving " << seg.len << " bytes (" << seg.seq
@@ -952,6 +957,24 @@ bool PseudoTcp::process(Segment& seg) {
         m_rlist.insert(it, rseg);
       }
     }
+    if (bRecover) {
+      RList::iterator it = m_rlist.begin();
+      while ((it != m_rlist.end()) && (it->seq <= m_rcv_nxt)) {
+        if (it->seq + it->len > m_rcv_nxt) {
+          sflags = sfImmediateAck;  // (Fast Recovery)
+          uint32_t nAdjust = (it->seq + it->len) - m_rcv_nxt;
+#if _DEBUGMSG >= _DBG_NORMAL
+          RTC_LOG(LS_INFO) << "Recovered " << nAdjust << " bytes (" << m_rcv_nxt
+                           << " -> " << m_rcv_nxt + nAdjust << ")";
+#endif  // _DEBUGMSG
+          m_rbuf.ConsumeWriteBuffer(nAdjust);
+          m_rcv_nxt += nAdjust;
+          m_rcv_wnd -= nAdjust;
+          bNewData = true;
+        }
+        it = m_rlist.erase(it);
+      }
+    }
   }
 
   attemptSend(sflags);

p2p/base/pseudotcp_unittest.cc

@@ -49,13 +49,18 @@ class PseudoTcpTestBase : public testing::Test,
         remote_mtu_(65535),
         delay_(0),
         loss_(0) {
-    // Set use of the test RNG to get predictable loss patterns.
+    // Set use of the test RNG to get predictable loss patterns. Otherwise,
+    // this test would occasionally get really unlucky loss and time out.
     rtc::SetRandomTestMode(true);
   }
   ~PseudoTcpTestBase() {
     // Put it back for the next test.
     rtc::SetRandomTestMode(false);
   }
+  // If true, both endpoints will send the "connect" segment simultaneously,
+  // rather than |local_| sending it followed by a response from |remote_|.
+  // Note that this is what chromoting ends up doing.
+  void SetSimultaneousOpen(bool enabled) { simultaneous_open_ = enabled; }
   void SetLocalMtu(int mtu) {
     local_.NotifyMTU(mtu);
     local_mtu_ = mtu;
@@ -66,6 +71,9 @@ class PseudoTcpTestBase : public testing::Test,
   }
   void SetDelay(int delay) { delay_ = delay; }
   void SetLoss(int percent) { loss_ = percent; }
+  // Used to cause the initial "connect" segment to be lost, needed for a
+  // regression test.
+  void DropNextPacket() { drop_next_packet_ = true; }
   void SetOptNagling(bool enable_nagles) {
     local_.SetOption(PseudoTcp::OPT_NODELAY, !enable_nagles);
     remote_.SetOption(PseudoTcp::OPT_NODELAY, !enable_nagles);
@@ -93,6 +101,12 @@ class PseudoTcpTestBase : public testing::Test,
     if (ret == 0) {
       UpdateLocalClock();
     }
+    if (simultaneous_open_) {
+      ret = remote_.Connect();
+      if (ret == 0) {
+        UpdateRemoteClock();
+      }
+    }
     return ret;
   }
   void Close() {
@@ -134,19 +148,28 @@ class PseudoTcpTestBase : public testing::Test,
   virtual WriteResult TcpWritePacket(PseudoTcp* tcp,
                                      const char* buffer,
                                      size_t len) {
+    // Drop a packet if the test called DropNextPacket.
+    if (drop_next_packet_) {
+      drop_next_packet_ = false;
+      RTC_LOG(LS_VERBOSE) << "Dropping packet due to DropNextPacket, size="
+                          << len;
+      return WR_SUCCESS;
+    }
     // Randomly drop the desired percentage of packets.
-    // Also drop packets that are larger than the configured MTU.
     if (rtc::CreateRandomId() % 100 < static_cast<uint32_t>(loss_)) {
       RTC_LOG(LS_VERBOSE) << "Randomly dropping packet, size=" << len;
-    } else if (len > static_cast<size_t>(std::min(local_mtu_, remote_mtu_))) {
+      return WR_SUCCESS;
+    }
+    // Also drop packets that are larger than the configured MTU.
+    if (len > static_cast<size_t>(std::min(local_mtu_, remote_mtu_))) {
       RTC_LOG(LS_VERBOSE) << "Dropping packet that exceeds path MTU, size="
                           << len;
-    } else {
+      return WR_SUCCESS;
-      int id = (tcp == &local_) ? MSG_RPACKET : MSG_LPACKET;
-      std::string packet(buffer, len);
-      rtc::Thread::Current()->PostDelayed(RTC_FROM_HERE, delay_, this, id,
-                                          rtc::WrapMessageData(packet));
     }
+    int id = (tcp == &local_) ? MSG_RPACKET : MSG_LPACKET;
+    std::string packet(buffer, len);
+    rtc::Thread::Current()->PostDelayed(RTC_FROM_HERE, delay_, this, id,
+                                        rtc::WrapMessageData(packet));
     return WR_SUCCESS;
   }
 
@@ -198,6 +221,8 @@ class PseudoTcpTestBase : public testing::Test,
   int remote_mtu_;
   int delay_;
   int loss_;
+  bool drop_next_packet_ = false;
+  bool simultaneous_open_ = false;
 };
 
 class PseudoTcpTest : public PseudoTcpTestBase {
@@ -620,6 +645,27 @@ TEST_F(PseudoTcpTest, TestSendWithLossAndOptNaglingOff) {
   TestTransfer(100000);  // less data so test runs faster
 }
 
+// Regression test for bugs.webrtc.org/9208.
+//
+// This bug resulted in corrupted data if a "connect" segment was received after
+// a data segment. This is only possible if:
+//
+// * The initial "connect" segment is lost, and retransmitted later.
+// * Both sides send "connect"s simultaneously, such that the local side thinks
+//   a connection is established even before its "connect" has been
+//   acknowledged.
+// * Nagle algorithm disabled, allowing a data segment to be sent before the
+//   "connect" has been acknowledged.
+TEST_F(PseudoTcpTest,
+       TestSendWhenFirstPacketLostWithOptNaglingOffAndSimultaneousOpen) {
+  SetLocalMtu(1500);
+  SetRemoteMtu(1500);
+  DropNextPacket();
+  SetOptNagling(false);
+  SetSimultaneousOpen(true);
+  TestTransfer(10000);
+}
+
 // Test sending data with 10% packet loss and Delayed ACK disabled.
 // Transmission should be slightly faster than with it enabled.
 TEST_F(PseudoTcpTest, TestSendWithLossAndOptAckDelayOff) {