WIP: Implement 5xx Error handling.

chrome/browser/google_apis/drive_uploader.cc

Index: chrome/browser/google_apis/drive_uploader.cc
diff --git a/chrome/browser/google_apis/drive_uploader.cc b/chrome/browser/google_apis/drive_uploader.cc
index f575d42f2ea689d1b56be885dacd230b524a2511..fb3e5f7cafb337bb2c02d3a32907785a99761d20 100644
--- a/chrome/browser/google_apis/drive_uploader.cc
+++ b/chrome/browser/google_apis/drive_uploader.cc
@@ -9,7 +9,9 @@
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/message_loop.h"
+#include "base/rand_util.h"
 #include "base/string_number_conversions.h"
+#include "base/time.h"
 #include "chrome/browser/google_apis/drive_service_interface.h"
 #include "chrome/browser/google_apis/drive_upload_mode.h"
 #include "chrome/browser/google_apis/gdata_wapi_parser.h"
@@ -35,6 +37,11 @@ int64 OpenFileStreamAndGetSizeOnBlockingPool(net::FileStream* file_stream,
   return file_stream->Available();
 }
 
+// Returns true if the status code is 5xx (SERVER ERROR).
+bool IsHttp5xxStatusCode(int status_code) {
+  return status_code >= 500 && status_code < 600;
+}
+
 }  // namespace
 
 namespace google_apis {
@@ -60,9 +67,11 @@ struct DriveUploader::UploadFileInfo {
         etag(etag),
         completion_callback(callback),
         content_length(0),
-        next_send_position(0),
         file_stream(new net::FileStream(NULL)),
         buf(new net::IOBuffer(kUploadChunkSize)),
+        buffer_position(0),
+        buffer_size(0),
+        trial_count(0),
         blocking_task_runner(task_runner) {
   }
 
@@ -72,8 +81,8 @@ struct DriveUploader::UploadFileInfo {
 
   // Bytes left to upload.
   int64 SizeRemaining()  const {
-    DCHECK(content_length >= next_send_position);
-    return content_length - next_send_position;
+    DCHECK(content_length >= buffer_position + buffer_size);
+    return content_length - (buffer_position + buffer_size);
   }
 
   // Useful for printf debugging.
@@ -116,9 +125,6 @@ struct DriveUploader::UploadFileInfo {
   // Header content-Length.
   int64 content_length;
 
-  // The start position of the contents to be sent as the next upload chunk.
-  int64 next_send_position;
-
   // For opening and reading from physical file.
   //
   // File operations are posted to |blocking_task_runner|, while the ownership
@@ -131,6 +137,17 @@ struct DriveUploader::UploadFileInfo {
   // Holds current content to be uploaded.
   const scoped_refptr<net::IOBuffer> buf;
 
+  // The current |buf|'s head position in the file.
+  int64 buffer_position;
+
+  // The size of available data in the |buf|.
+  int64 buffer_size;
+
+  // The number of current trial to send the current |buf| content.
+  // This is used to calculate waiting duration. Please see also
+  // ResumeInterruptedUpload.
+  int trial_count;
+
   // Runner for net::FileStream tasks.
   const scoped_refptr<base::SequencedTaskRunner> blocking_task_runner;
 };
@@ -270,6 +287,31 @@ void DriveUploader::OnUploadLocationReceived(
   UploadNextChunk(upload_file_info.Pass());
 }
 
+void DriveUploader::UploadCurrentChunk(
+    scoped_ptr<UploadFileInfo> upload_file_info,
+    int64 restart_position) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+
+  int64 buffer_offset =
+      restart_position - upload_file_info->buffer_position;
+  DCHECK_GE(buffer_offset, 0);
+
+  UploadFileInfo* info_ptr = upload_file_info.get();
+  drive_service_->ResumeUpload(
+      ResumeUploadParams(info_ptr->upload_mode,
+                         restart_position,
+                         info_ptr->buffer_position + info_ptr->buffer_size,
+                         info_ptr->content_length,
+                         info_ptr->content_type,
+                         info_ptr->buf,
+                         buffer_offset,
+                         info_ptr->upload_location,
+                         info_ptr->drive_path),
+      base::Bind(&DriveUploader::OnResumeUploadResponseReceived,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 base::Passed(&upload_file_info)));
+}
+
 void DriveUploader::UploadNextChunk(
     scoped_ptr<UploadFileInfo> upload_file_info) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
@@ -322,18 +364,22 @@ void DriveUploader::ReadCompletionCallback(
     return;
   }
 
-  int64 start_position = upload_file_info->next_send_position;
-  upload_file_info->next_send_position += bytes_read;
-  int64 end_position = upload_file_info->next_send_position;
+  // Update the buffer's position.
+  upload_file_info->buffer_position += upload_file_info->buffer_size;
+  upload_file_info->buffer_size = bytes_read;
+
+  // Reset the trial count.
+  upload_file_info->trial_count = 0;
 
   UploadFileInfo* info_ptr = upload_file_info.get();
   drive_service_->ResumeUpload(
       ResumeUploadParams(info_ptr->upload_mode,
-                         start_position,
-                         end_position,
+                         info_ptr->buffer_position,
+                         info_ptr->buffer_position + info_ptr->buffer_size,
                          info_ptr->content_length,
                          info_ptr->content_type,
                          info_ptr->buf,
+                         0,  // buffer's offset.
                          info_ptr->upload_location,
                          info_ptr->drive_path),
       base::Bind(&DriveUploader::OnResumeUploadResponseReceived,
@@ -367,19 +413,27 @@ void DriveUploader::OnResumeUploadResponseReceived(
     return;
   }
 
+  // The uploading is interrupted. Move to resuming flow.
+  if (IsHttp5xxStatusCode(response.code)) {
+    ResumeInterruptedUpload(upload_file_info.Pass());
+    return;
+  }
+
   // If code is 308 (RESUME_INCOMPLETE) and range_received is what has been
   // previously uploaded (i.e. = upload_file_info->end_position), proceed to
   // upload the next chunk.
   if (response.code != HTTP_RESUME_INCOMPLETE ||
       response.start_position_received != 0 ||
-      response.end_position_received != upload_file_info->next_send_position) {
+      response.end_position_received < upload_file_info->buffer_position ||
+      response.end_position_received >
+          upload_file_info->buffer_position + upload_file_info->buffer_size) {
     // TODO(achuith): Handle error cases, e.g.
     // - when previously uploaded data wasn't received by Google Docs server,
     //   i.e. when end_position_received < upload_file_info->end_position
     LOG(ERROR) << "UploadNextChunk http code=" << response.code
         << ", start_position_received=" << response.start_position_received
-        << ", end_position_received=" << response.end_position_received
-        << ", expected end range=" << upload_file_info->next_send_position;
+        << ", end_position_received=" << response.end_position_received;
+    //  TODO
     UploadFailed(upload_file_info.Pass(),
                  response.code == HTTP_FORBIDDEN ?
                      DRIVE_UPLOAD_ERROR_NO_SPACE : DRIVE_UPLOAD_ERROR_ABORT);
@@ -391,7 +445,65 @@ void DriveUploader::OnResumeUploadResponseReceived(
            << " for [" << upload_file_info->title << "]";
 
   // Continue uploading.
-  UploadNextChunk(upload_file_info.Pass());
+  if (upload_file_info->buffer_position + upload_file_info->buffer_size ==
+      response.end_position_received) {
+    // The current chunk has been sent successuflly, so send the next chunk.
+    UploadNextChunk(upload_file_info.Pass());
+  } else {
+    // There is remaining data in the current chunk. Re-send it.
+    //
+    // Note that PostTask is necessary here because we have to finish an
+    // uploading callback before calling ResumeUpload again due to the
+    // implementation of OperationRegistry (http://crbug.com/134814).
+    MessageLoop::current()->PostTask(
+        FROM_HERE,
+        base::Bind(&DriveUploader::UploadCurrentChunk,
+                   weak_ptr_factory_.GetWeakPtr(),
+                   base::Passed(&upload_file_info),
+                   response.end_position_received));
+  }
+}
+
+void DriveUploader::ResumeInterruptedUpload(
+    scoped_ptr<UploadFileInfo> upload_file_info) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+
+  // When the upload is interrupted, we'll wait some period and restart
+  // the upload, following the document:
+  // https://developers.google.com/drive/manage-uploads#exp-backoff
+  // The waiting duration is calculated as follows:
+  // First trial: 1sec + random millisecs.
+  // Second trial: 2secs + random millisecs.
+  // Third trial: 4secs + random millisecs.
+  // Forth trial: 8secs + random millisecs.
+  //   :
+  // n-th trial: 2^(n-1)secs + random millisecs.
+  base::TimeDelta wait_duration =
+      base::TimeDelta::FromSeconds(1 << upload_file_info->trial_count) +
+      base::TimeDelta::FromMilliseconds(base::RandInt(0, 9));
+  ++upload_file_info->trial_count;
+
+  MessageLoop::current()->PostDelayedTask(
+        FROM_HERE,
+        base::Bind(&DriveUploader::ResumeInterruptedUploadAfterWaiting,
+                   weak_ptr_factory_.GetWeakPtr(),
+                   base::Passed(&upload_file_info)),
+        wait_duration);
+}
+
+void DriveUploader::ResumeInterruptedUploadAfterWaiting(
+    scoped_ptr<UploadFileInfo> upload_file_info) {
+  DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
+
+  UploadFileInfo* info_ptr = upload_file_info.get();
+  drive_service_->GetUploadState(
+      info_ptr->upload_mode,
+      info_ptr->drive_path,
+      info_ptr->upload_location,
+      info_ptr->content_length,
+      base::Bind(&DriveUploader::OnResumeUploadResponseReceived,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 base::Passed(&upload_file_info)));
 }
 
 void DriveUploader::UploadFailed(scoped_ptr<UploadFileInfo> upload_file_info,