WIP: Implement 5xx Error handling.

chrome/browser/google_apis/drive_uploader.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "chrome/browser/google_apis/drive_uploader.h"
 
 #include <algorithm>
 
 #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"
 #include "content/public/browser/browser_thread.h"
 #include "net/base/file_stream.h"
 #include "net/base/net_errors.h"
 
 using content::BrowserThread;
 
 namespace {
 
 // Google Documents List API requires uploading in chunks of 512kB.
 const int64 kUploadChunkSize = 512 * 1024;
 
 // Opens |path| with |file_stream| and returns the file size.
 // If failed, returns an error code in a negative value.
 int64 OpenFileStreamAndGetSizeOnBlockingPool(net::FileStream* file_stream,
                                              const FilePath& path) {
   int result = file_stream->OpenSync(
       path, base::PLATFORM_FILE_OPEN | base::PLATFORM_FILE_READ);
   if (result != net::OK)
     return result;
   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 {
 
 // Structure containing current upload information of file, passed between
 // DriveServiceInterface methods and callbacks.
 struct DriveUploader::UploadFileInfo {
   UploadFileInfo(scoped_refptr<base::SequencedTaskRunner> task_runner,
                  UploadMode upload_mode,
                  const GURL& initial_upload_location,
                  const FilePath& drive_path,
                  const FilePath& local_path,
                  const std::string& title,
                  const std::string& content_type,
                  const std::string& etag,
                  const UploadCompletionCallback& callback)
       : upload_mode(upload_mode),
         initial_upload_location(initial_upload_location),
         drive_path(drive_path),
         file_path(local_path),
         title(title),
         content_type(content_type),
         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) {
   }
 
   ~UploadFileInfo() {
     blocking_task_runner->DeleteSoon(FROM_HERE, file_stream.release());
   }
 
   // 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.
   std::string DebugString() const {
     return "title=[" + title +
            "], file_path=[" + file_path.AsUTF8Unsafe() +
            "], content_type=[" + content_type +
            "], content_length=[" + base::UintToString(content_length) +
            "], drive_path=[" + drive_path.AsUTF8Unsafe() +
            "]";
@@ skipping 22 matching lines @@
   // Callback to be invoked once the upload has finished.
   const UploadCompletionCallback completion_callback;
 
   // Location URL where file is to be uploaded to, returned from
   // InitiateUpload. Used for the subsequent ResumeUpload requests.
   GURL upload_location;
 
   // 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
   // of the stream is held in UI thread. At the point when this UploadFileInfo
   // is destroyed, the ownership of the stream is passed to the worker pool.
   // TODO(kinaba): We should switch to async API of FileStream once
   // crbug.com/164312 is fixed.
   scoped_ptr<net::FileStream> file_stream;
 
   // 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;
 };
 
 DriveUploader::DriveUploader(DriveServiceInterface* drive_service)
     : drive_service_(drive_service),
       ALLOW_THIS_IN_INITIALIZER_LIST(weak_ptr_factory_(this)) {
   base::SequencedWorkerPool* blocking_pool = BrowserThread::GetBlockingPool();
   blocking_task_runner_ = blocking_pool->GetSequencedTaskRunner(
       blocking_pool->GetSequenceToken());
@@ skipping 119 matching lines @@
     UploadFailed(upload_file_info.Pass(), DRIVE_UPLOAD_ERROR_ABORT);
     return;
   }
 
   upload_file_info->upload_location = upload_location;
 
   // Start the upload from the beginning of the file.
   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));
 
   // Determine number of bytes to read for this upload iteration.
   const int bytes_to_read = std::min(upload_file_info->SizeRemaining(),
                                      kUploadChunkSize);
 
   if (bytes_to_read == 0) {
     // net::FileStream doesn't allow to read 0 bytes, so directly proceed to the
@@ skipping 32 matching lines @@
   DCHECK_EQ(bytes_to_read, bytes_read);
   DVLOG(1) << "ReadCompletionCallback bytes read=" << bytes_read;
 
   if (bytes_read < 0) {
     LOG(ERROR) << "Error reading from file "
                << upload_file_info->file_path.value();
     UploadFailed(upload_file_info.Pass(), DRIVE_UPLOAD_ERROR_ABORT);
     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,
                  weak_ptr_factory_.GetWeakPtr(),
                  base::Passed(&upload_file_info)));
 }
 
 void DriveUploader::OnResumeUploadResponseReceived(
     scoped_ptr<UploadFileInfo> upload_file_info,
     const ResumeUploadResponse& response,
@@ skipping 13 matching lines @@
                                               entry.Pass());
     return;
   }
 
   // ETag mismatch.
   if (response.code == HTTP_PRECONDITION) {
     UploadFailed(upload_file_info.Pass(), DRIVE_UPLOAD_ERROR_CONFLICT);
     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);
     return;
   }
 
   DVLOG(1) << "Received range " << response.start_position_received
            << "-" << response.end_position_received
            << " 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,
                                  DriveUploadError error) {
   DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
 
   LOG(ERROR) << "Upload failed " << upload_file_info->DebugString();
 
   upload_file_info->completion_callback.Run(error,
                                             upload_file_info->drive_path,
                                             upload_file_info->file_path,
                                             scoped_ptr<ResourceEntry>());
 }
 
 }  // namespace google_apis